reformatting
formatting_check / formatting_check (push) Failing after 3s
Python Test / test (.ps1, windows-latest, 3.9) (push) Has been cancelled
Python Test / test (.sh, macos-latest, 3.9) (push) Has been cancelled
Python Test / test (.sh, ubuntu-latest, 3.9) (push) Has been cancelled
formatting_check / formatting_check (push) Failing after 3s
Python Test / test (.ps1, windows-latest, 3.9) (push) Has been cancelled
Python Test / test (.sh, macos-latest, 3.9) (push) Has been cancelled
Python Test / test (.sh, ubuntu-latest, 3.9) (push) Has been cancelled
This commit is contained in:
@@ -7,16 +7,18 @@ This script simulates the scenario where a download fails and then tries to redo
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
|
||||
# Add the bdfr module to the path
|
||||
import sys
|
||||
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"""
|
||||
|
||||
@@ -36,6 +38,7 @@ class TestProgressCallback(LoggingCallback):
|
||||
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...")
|
||||
@@ -53,7 +56,7 @@ def test_file_locking_fix():
|
||||
download_id1 = manager.create_download(
|
||||
DownloadType.USER,
|
||||
"test_user_12345", # This user doesn't exist, should fail
|
||||
progress_callbacks=[TestProgressCallback()]
|
||||
progress_callbacks=[TestProgressCallback()],
|
||||
)
|
||||
|
||||
# Start the download (it should fail)
|
||||
@@ -71,7 +74,7 @@ def test_file_locking_fix():
|
||||
download_id2 = manager.create_download(
|
||||
DownloadType.USER,
|
||||
"test_user_67890", # This user also doesn't exist, should fail
|
||||
progress_callbacks=[TestProgressCallback()]
|
||||
progress_callbacks=[TestProgressCallback()],
|
||||
)
|
||||
|
||||
# Start the second download
|
||||
@@ -100,7 +103,7 @@ def test_file_locking_fix():
|
||||
print(f" - {log_file.name}")
|
||||
# Check if file is accessible (not locked)
|
||||
try:
|
||||
with open(log_file, 'r') as f:
|
||||
with open(log_file, "r") as f:
|
||||
content = f.read()
|
||||
print(f" SUCCESS: Log file is accessible ({len(content)} characters)")
|
||||
except PermissionError:
|
||||
@@ -112,9 +115,7 @@ def test_file_locking_fix():
|
||||
# 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()]
|
||||
DownloadType.USER, "test_user_cleanup", progress_callbacks=[TestProgressCallback()]
|
||||
)
|
||||
|
||||
success3 = manager.start_download(download_id3)
|
||||
@@ -132,6 +133,7 @@ def test_file_locking_fix():
|
||||
print("\nSUCCESS: All tests passed! File locking issue appears to be fixed.")
|
||||
return True
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
success = test_file_locking_fix()
|
||||
@@ -144,5 +146,6 @@ if __name__ == "__main__":
|
||||
except Exception as e:
|
||||
print(f"\nTest failed with exception: {e}")
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
Reference in New Issue
Block a user