BDFR_Web/tests/test_file_locking_fix.py

#!/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

# Add the bdfr module to the path
import sys
import tempfile
import time
from pathlib import Path

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)