92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
"""
|
|
File handling utilities for Cognitive Prism Automation Tests
|
|
"""
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from config.config import DOWNLOAD_DIR, DOWNLOAD_TIMEOUT
|
|
|
|
|
|
class FileHandlers:
|
|
"""Helper class for file operations"""
|
|
|
|
@staticmethod
|
|
def wait_for_download(filename, timeout=DOWNLOAD_TIMEOUT):
|
|
"""
|
|
Wait for file to be downloaded
|
|
|
|
Args:
|
|
filename: Name of the file to wait for
|
|
timeout: Maximum time to wait in seconds
|
|
|
|
Returns:
|
|
Path: Path to downloaded file
|
|
|
|
Raises:
|
|
TimeoutError: If file is not downloaded within timeout
|
|
"""
|
|
download_path = Path(DOWNLOAD_DIR) / filename
|
|
start_time = time.time()
|
|
|
|
while not download_path.exists():
|
|
if time.time() - start_time > timeout:
|
|
raise TimeoutError(f"File {filename} was not downloaded within {timeout} seconds")
|
|
time.sleep(0.5)
|
|
|
|
# Wait for file to be completely downloaded (no .crdownload extension)
|
|
while download_path.suffix == ".crdownload" or download_path.name.endswith(".tmp"):
|
|
if time.time() - start_time > timeout:
|
|
raise TimeoutError(f"File {filename} download did not complete within {timeout} seconds")
|
|
time.sleep(0.5)
|
|
download_path = Path(DOWNLOAD_DIR) / filename
|
|
|
|
return download_path
|
|
|
|
@staticmethod
|
|
def get_latest_downloaded_file(pattern="*"):
|
|
"""
|
|
Get the most recently downloaded file matching pattern
|
|
|
|
Args:
|
|
pattern: File pattern to match (e.g., "*.csv", "*.json")
|
|
|
|
Returns:
|
|
Path: Path to latest file or None if no file found
|
|
"""
|
|
download_path = Path(DOWNLOAD_DIR)
|
|
files = list(download_path.glob(pattern))
|
|
|
|
if not files:
|
|
return None
|
|
|
|
# Sort by modification time and return the latest
|
|
latest_file = max(files, key=lambda f: f.stat().st_mtime)
|
|
return latest_file
|
|
|
|
@staticmethod
|
|
def cleanup_downloads(pattern="*", older_than_days=7):
|
|
"""
|
|
Clean up old downloaded files
|
|
|
|
Args:
|
|
pattern: File pattern to match
|
|
older_than_days: Delete files older than this many days
|
|
"""
|
|
import time
|
|
download_path = Path(DOWNLOAD_DIR)
|
|
current_time = time.time()
|
|
cutoff_time = current_time - (older_than_days * 24 * 60 * 60)
|
|
|
|
for file in download_path.glob(pattern):
|
|
if file.stat().st_mtime < cutoff_time:
|
|
try:
|
|
file.unlink()
|
|
except Exception as e:
|
|
print(f"Error deleting file {file}: {e}")
|
|
|
|
|
|
|
|
|
|
|
|
|