feat(UI): initial working frontend UI
This commit is contained in:
@@ -0,0 +1,148 @@
|
||||
#!/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)
|
||||
@@ -0,0 +1,161 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
Test script to verify user folder structure changes.
|
||||
|
||||
This script tests that:
|
||||
1. Subreddit downloads go to: downloads/subreddit_name/files
|
||||
2. User downloads go to: downloads/username/subreddit_name/files
|
||||
"""
|
||||
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
# Set UTF-8 encoding for Windows console
|
||||
if sys.platform == 'win32':
|
||||
import codecs
|
||||
sys.stdout = codecs.getwriter('utf-8')(sys.stdout.buffer, 'strict')
|
||||
sys.stderr = codecs.getwriter('utf-8')(sys.stderr.buffer, 'strict')
|
||||
|
||||
from bdfr.api import BDFRManager, DownloadType
|
||||
from bdfr.configuration import Configuration
|
||||
|
||||
|
||||
def test_subreddit_directory_structure():
|
||||
"""Test that subreddit downloads use correct directory structure"""
|
||||
print("\n=== Testing Subreddit Directory Structure ===")
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
manager = BDFRManager(download_directory=tmpdir)
|
||||
|
||||
# Create a download for a subreddit
|
||||
download_id = manager.create_download(
|
||||
DownloadType.SUBREDDIT,
|
||||
"test_subreddit"
|
||||
)
|
||||
|
||||
download_info = manager.get_download_status(download_id)
|
||||
config = download_info["config"]
|
||||
|
||||
expected_dir = str(Path(tmpdir))
|
||||
actual_dir = config.directory
|
||||
|
||||
print(f"Expected directory: {expected_dir}")
|
||||
print(f"Actual directory: {actual_dir}")
|
||||
print(f"Subreddit config: {config.subreddit}")
|
||||
print(f"Folder scheme: {config.folder_scheme}")
|
||||
|
||||
assert actual_dir == expected_dir, f"Subreddit directory mismatch!"
|
||||
print("[OK] Subreddit directory structure is correct")
|
||||
print(f" Files will be saved to: {actual_dir}/{{SUBREDDIT}}/{{files}}")
|
||||
|
||||
|
||||
def test_user_directory_structure():
|
||||
"""Test that user downloads use correct directory structure"""
|
||||
print("\n=== Testing User Directory Structure ===")
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
manager = BDFRManager(download_directory=tmpdir)
|
||||
|
||||
# Create a download for a user
|
||||
username = "test_user"
|
||||
download_id = manager.create_download(
|
||||
DownloadType.USER,
|
||||
username
|
||||
)
|
||||
|
||||
download_info = manager.get_download_status(download_id)
|
||||
config = download_info["config"]
|
||||
|
||||
expected_dir = str(Path(tmpdir) / username)
|
||||
actual_dir = config.directory
|
||||
|
||||
print(f"Expected directory: {expected_dir}")
|
||||
print(f"Actual directory: {actual_dir}")
|
||||
print(f"User config: {config.user}")
|
||||
print(f"Folder scheme: {config.folder_scheme}")
|
||||
|
||||
assert actual_dir == expected_dir, f"User directory mismatch!"
|
||||
print("[OK] User directory structure is correct")
|
||||
print(f" Files will be saved to: {actual_dir}/{{SUBREDDIT}}/{{files}}")
|
||||
|
||||
|
||||
def test_convenience_method():
|
||||
"""Test the convenience method download_user()"""
|
||||
print("\n=== Testing download_user() Convenience Method ===")
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
manager = BDFRManager(download_directory=tmpdir)
|
||||
|
||||
# Don't actually start the download, just check the config
|
||||
username = "convenience_test_user"
|
||||
|
||||
# Create config manually like the convenience method does
|
||||
config = Configuration()
|
||||
user_directory = Path(tmpdir) / username
|
||||
config.directory = str(user_directory)
|
||||
config.user = [username]
|
||||
|
||||
expected_dir = str(Path(tmpdir) / username)
|
||||
actual_dir = config.directory
|
||||
|
||||
print(f"Expected directory: {expected_dir}")
|
||||
print(f"Actual directory: {actual_dir}")
|
||||
print(f"User config: {config.user}")
|
||||
|
||||
assert actual_dir == expected_dir, f"Convenience method directory mismatch!"
|
||||
print("[OK] Convenience method directory structure is correct")
|
||||
|
||||
|
||||
def demonstrate_folder_structure():
|
||||
"""Demonstrate the folder structure for both download types"""
|
||||
print("\n=== Folder Structure Demonstration ===")
|
||||
print("\nWhen downloading from a SUBREDDIT 'python':")
|
||||
print(" downloads/")
|
||||
print(" └── python/")
|
||||
print(" ├── file1.jpg")
|
||||
print(" ├── file2.png")
|
||||
print(" └── file3.mp4")
|
||||
|
||||
print("\nWhen downloading from a USER 'spez' who posts to multiple subreddits:")
|
||||
print(" downloads/")
|
||||
print(" └── spez/")
|
||||
print(" ├── python/")
|
||||
print(" │ ├── file1.jpg")
|
||||
print(" │ └── file2.png")
|
||||
print(" ├── announcements/")
|
||||
print(" │ └── file3.jpg")
|
||||
print(" └── pics/")
|
||||
print(" └── file4.png")
|
||||
|
||||
print("\n[OK] This structure allows:")
|
||||
print(" 1. Easy identification of user-specific downloads")
|
||||
print(" 2. Organization by subreddit within each user folder")
|
||||
print(" 3. No conflicts between subreddit and user downloads")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("=" * 60)
|
||||
print("Testing User Folder Structure Changes")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
test_subreddit_directory_structure()
|
||||
test_user_directory_structure()
|
||||
test_convenience_method()
|
||||
demonstrate_folder_structure()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("[SUCCESS] All tests passed!")
|
||||
print("=" * 60)
|
||||
|
||||
except AssertionError as e:
|
||||
print(f"\n[FAIL] Test failed: {e}")
|
||||
exit(1)
|
||||
except Exception as e:
|
||||
print(f"\n[ERROR] Unexpected error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
exit(1)
|
||||
Reference in New Issue
Block a user