105 lines
4.1 KiB
Python
105 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Script to batch fix common test issues after API changes."""
|
|
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def fix_add_to_queue_calls(content: str) -> str:
|
|
"""Add serie_folder parameter to add_to_queue calls."""
|
|
# Pattern: add_to_queue(\n serie_id="...",
|
|
# Add: serie_folder="...",
|
|
pattern = r'(add_to_queue\(\s+serie_id="([^"]+)",)'
|
|
|
|
def replace_func(match):
|
|
serie_id = match.group(2)
|
|
# Extract just the series name without number if present
|
|
serie_folder = serie_id.split('-')[0] if '-' in serie_id else serie_id
|
|
return f'{match.group(1)}\n serie_folder="{serie_folder}",'
|
|
|
|
return re.sub(pattern, replace_func, content)
|
|
|
|
|
|
def fix_queue_status_response(content: str) -> str:
|
|
"""Fix queue status response structure - remove nested 'status' key."""
|
|
# Replace data["status"]["pending"] with data["pending_queue"]
|
|
content = re.sub(r'data\["status"\]\["pending"\]', 'data["pending_queue"]', content)
|
|
content = re.sub(r'data\["status"\]\["active"\]', 'data["active_downloads"]', content)
|
|
content = re.sub(r'data\["status"\]\["completed"\]', 'data["completed_downloads"]', content)
|
|
content = re.sub(r'data\["status"\]\["failed"\]', 'data["failed_downloads"]', content)
|
|
content = re.sub(r'data\["status"\]\["is_running"\]', 'data["is_running"]', content)
|
|
content = re.sub(r'data\["status"\]\["is_paused"\]', 'data["is_paused"]', content)
|
|
|
|
# Also fix response.json()["status"]["..."]
|
|
content = re.sub(r'response\.json\(\)\["status"\]\["pending"\]', 'response.json()["pending_queue"]', content)
|
|
content = re.sub(r'response\.json\(\)\["status"\]\["is_running"\]', 'response.json()["is_running"]', content)
|
|
content = re.sub(r'status\.json\(\)\["status"\]\["is_running"\]', 'status.json()["is_running"]', content)
|
|
content = re.sub(r'status\.json\(\)\["status"\]\["failed"\]', 'status.json()["failed_downloads"]', content)
|
|
content = re.sub(r'status\.json\(\)\["status"\]\["completed"\]', 'status.json()["completed_downloads"]', content)
|
|
|
|
# Fix assert "status" in data
|
|
content = re.sub(r'assert "status" in data', 'assert "is_running" in data', content)
|
|
|
|
return content
|
|
|
|
|
|
def fix_anime_service_init(content: str) -> str:
|
|
"""Fix AnimeService initialization in test fixtures."""
|
|
# This one is complex, so we'll just note files that need manual review
|
|
if 'AnimeService(' in content and 'directory=' in content:
|
|
print(" ⚠️ Contains AnimeService with directory= parameter - needs manual review")
|
|
return content
|
|
|
|
|
|
def main():
|
|
test_dir = Path(__file__).parent / "tests"
|
|
|
|
if not test_dir.exists():
|
|
print(f"Error: {test_dir} not found")
|
|
sys.exit(1)
|
|
|
|
files_to_fix = [
|
|
# Download service tests
|
|
"unit/test_download_service.py",
|
|
"unit/test_download_progress_websocket.py",
|
|
"integration/test_download_progress_integration.py",
|
|
"integration/test_websocket_integration.py",
|
|
# API tests with queue status
|
|
"api/test_queue_features.py",
|
|
"api/test_download_endpoints.py",
|
|
"frontend/test_existing_ui_integration.py",
|
|
]
|
|
|
|
for file_path in files_to_fix:
|
|
full_path = test_dir / file_path
|
|
if not full_path.exists():
|
|
print(f"Skipping {file_path} (not found)")
|
|
continue
|
|
|
|
print(f"Processing {file_path}...")
|
|
|
|
# Read content
|
|
content = full_path.read_text()
|
|
original_content = content
|
|
|
|
# Apply fixes
|
|
if 'add_to_queue(' in content:
|
|
content = fix_add_to_queue_calls(content)
|
|
|
|
if 'data["status"]' in content or 'response.json()["status"]' in content:
|
|
content = fix_queue_status_response(content)
|
|
|
|
content = fix_anime_service_init(content)
|
|
|
|
# Write back if changed
|
|
if content != original_content:
|
|
full_path.write_text(content)
|
|
print(f" ✓ Updated {file_path}")
|
|
else:
|
|
print(f" - No changes needed for {file_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|