99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for bulk_update_movies.py
|
|
This tests the bulk update functionality without modifying data
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
def test_bulk_update_dry_run():
|
|
"""Test bulk update script in dry-run mode"""
|
|
print("🧪 Testing Bulk Update Script")
|
|
print("=" * 40)
|
|
|
|
# Check if we can import the required modules
|
|
try:
|
|
from clients.radarr_db_client import RadarrDbClient
|
|
from core.database import NFOGuardDatabase
|
|
print("✅ Successfully imported required modules")
|
|
except Exception as e:
|
|
print(f"❌ Import error: {e}")
|
|
return False
|
|
|
|
# Test database connections
|
|
print("\n🔌 Testing Database Connections...")
|
|
|
|
# Test Radarr database
|
|
try:
|
|
radarr_db = RadarrDbClient.from_env()
|
|
if not radarr_db:
|
|
print("❌ Radarr database connection failed - check environment variables")
|
|
return False
|
|
print("✅ Radarr database connection successful")
|
|
except Exception as e:
|
|
print(f"❌ Radarr database error: {e}")
|
|
return False
|
|
|
|
# Test NFOGuard database
|
|
try:
|
|
db_path = Path(os.environ.get("DB_PATH", "/app/data/media_dates.db"))
|
|
nfo_db = NFOGuardDatabase(db_path)
|
|
print(f"✅ NFOGuard database connection successful: {db_path}")
|
|
except Exception as e:
|
|
print(f"❌ NFOGuard database error: {e}")
|
|
return False
|
|
|
|
# Test query execution (dry run)
|
|
print("\n📊 Testing Query Execution...")
|
|
|
|
try:
|
|
query = """
|
|
SELECT
|
|
mm."ImdbId" as imdb_id,
|
|
m."Id" as movie_id,
|
|
mm."Title" as title,
|
|
mm."Year" as year,
|
|
m."Path" as path
|
|
FROM "Movies" m
|
|
JOIN "MovieMetadata" mm ON m."MovieMetadataId" = mm."Id"
|
|
WHERE mm."ImdbId" IS NOT NULL AND mm."ImdbId" != ''
|
|
ORDER BY mm."Title"
|
|
LIMIT 5
|
|
"""
|
|
|
|
with radarr_db._get_connection() as conn:
|
|
if radarr_db.db_type == "postgresql":
|
|
import psycopg2.extras
|
|
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
|
else:
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(query)
|
|
movies = cursor.fetchall()
|
|
|
|
print(f"✅ Successfully queried Radarr database: {len(movies)} sample movies found")
|
|
|
|
# Show sample data
|
|
for i, movie in enumerate(movies):
|
|
if radarr_db.db_type == "postgresql":
|
|
imdb_id, title, year = movie['imdb_id'], movie['title'], movie['year']
|
|
else:
|
|
imdb_id, title, year = movie[0], movie[2], movie[3]
|
|
print(f" Sample {i+1}: {title} ({year}) - {imdb_id}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Query execution failed: {e}")
|
|
return False
|
|
|
|
print("\n✅ All tests passed! Bulk update script should work correctly.")
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
success = test_bulk_update_dry_run()
|
|
if not success:
|
|
print("\n❌ Bulk update test failed!")
|
|
sys.exit(1)
|
|
else:
|
|
print("\n🎉 Bulk update test successful!") |