148 lines
5.0 KiB
Python
148 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify that the file locking issue is fixed.
|
|
This script simulates the scenario where a download fails and then tries to redownload.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# Add the bdfr module to the path
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from bdfr.api import BDFRManager, DownloadType, LoggingCallback, ProgressEvent
|
|
|
|
class TestProgressCallback(LoggingCallback):
|
|
"""Test callback that simulates a failure"""
|
|
|
|
def __init__(self):
|
|
super().__init__("test_logger")
|
|
self.events = []
|
|
|
|
async def on_progress(self, event: ProgressEvent):
|
|
self.events.append(event)
|
|
await super().on_progress(event)
|
|
|
|
async def on_error(self, event: ProgressEvent):
|
|
self.events.append(event)
|
|
await super().on_error(event)
|
|
|
|
async def on_completed(self, event: ProgressEvent):
|
|
self.events.append(event)
|
|
await super().on_completed(event)
|
|
|
|
def test_file_locking_fix():
|
|
"""Test that the file locking issue is resolved"""
|
|
print("Testing file locking fix...")
|
|
|
|
# Create a temporary directory for testing
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
temp_path = Path(temp_dir)
|
|
print(f"Using temporary directory: {temp_path}")
|
|
|
|
# Create BDFR manager
|
|
manager = BDFRManager(temp_path)
|
|
|
|
# Test 1: Create a download that will fail
|
|
print("\n1. Creating first download (will fail)...")
|
|
download_id1 = manager.create_download(
|
|
DownloadType.USER,
|
|
"test_user_12345", # This user doesn't exist, should fail
|
|
progress_callbacks=[TestProgressCallback()]
|
|
)
|
|
|
|
# Start the download (it should fail)
|
|
manager.start_download(download_id1)
|
|
|
|
# Wait a bit for the download to start and fail
|
|
time.sleep(2)
|
|
|
|
# Check status
|
|
status1 = manager.get_download_status(download_id1)
|
|
print(f"First download status: {status1['status'] if status1 else 'Not found'}")
|
|
|
|
# Test 2: Try to create a second download immediately after
|
|
print("\n2. Creating second download (should work without file locking error)...")
|
|
download_id2 = manager.create_download(
|
|
DownloadType.USER,
|
|
"test_user_67890", # This user also doesn't exist, should fail
|
|
progress_callbacks=[TestProgressCallback()]
|
|
)
|
|
|
|
# Start the second download
|
|
success = manager.start_download(download_id2)
|
|
|
|
if success:
|
|
print("SUCCESS: Second download started successfully (no file locking error)")
|
|
else:
|
|
print("FAILED: Failed to start second download")
|
|
return False
|
|
|
|
# Wait for second download to fail
|
|
time.sleep(2)
|
|
|
|
# Check status
|
|
status2 = manager.get_download_status(download_id2)
|
|
print(f"Second download status: {status2['status'] if status2 else 'Not found'}")
|
|
|
|
# Test 3: Check that log files are unique
|
|
print("\n3. Checking for unique log files...")
|
|
logs_dir = temp_path / "logs"
|
|
if logs_dir.exists():
|
|
log_files = list(logs_dir.glob("*.log"))
|
|
print(f"Found {len(log_files)} log files:")
|
|
for log_file in log_files:
|
|
print(f" - {log_file.name}")
|
|
# Check if file is accessible (not locked)
|
|
try:
|
|
with open(log_file, 'r') as f:
|
|
content = f.read()
|
|
print(f" SUCCESS: Log file is accessible ({len(content)} characters)")
|
|
except PermissionError:
|
|
print(f" FAILED: Log file is still locked!")
|
|
return False
|
|
else:
|
|
print("No logs directory found")
|
|
|
|
# Test 4: Try to create a third download to ensure cleanup worked
|
|
print("\n4. Creating third download to verify cleanup...")
|
|
download_id3 = manager.create_download(
|
|
DownloadType.USER,
|
|
"test_user_cleanup",
|
|
progress_callbacks=[TestProgressCallback()]
|
|
)
|
|
|
|
success3 = manager.start_download(download_id3)
|
|
if success3:
|
|
print("SUCCESS: Third download started successfully (cleanup worked)")
|
|
else:
|
|
print("FAILED: Third download failed to start")
|
|
return False
|
|
|
|
# Wait and check final status
|
|
time.sleep(2)
|
|
status3 = manager.get_download_status(download_id3)
|
|
print(f"Third download status: {status3['status'] if status3 else 'Not found'}")
|
|
|
|
print("\nSUCCESS: All tests passed! File locking issue appears to be fixed.")
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
success = test_file_locking_fix()
|
|
if success:
|
|
print("\nTest completed successfully!")
|
|
sys.exit(0)
|
|
else:
|
|
print("\nTest failed!")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"\nTest failed with exception: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1) |