db updates

test_end_to_end.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""
+End-to-end testing script for NFOGuard workflow
+Tests the complete flow: Database -> Manual Scan -> NFO Creation
+"""
+
+import os
+import sys
+import requests
+import time
+from pathlib import Path
+
+def test_server_health():
+    """Test if the server is running and healthy"""
+    try:
+        response = requests.get("http://localhost:8080/health", timeout=10)
+        if response.status_code == 200:
+            health_data = response.json()
+            print(f"✅ Server is healthy: {health_data['status']}")
+            print(f"   Version: {health_data['version']}")
+            print(f"   Database: {health_data['database_status']}")
+            if health_data.get('radarr_database'):
+                radarr_status = health_data['radarr_database']['status']
+                print(f"   Radarr DB: {radarr_status}")
+            return True
+        else:
+            print(f"❌ Server health check failed: {response.status_code}")
+            return False
+    except Exception as e:
+        print(f"❌ Cannot connect to server: {e}")
+        return False
+
+def test_database_performance():
+    """Test the database performance endpoint"""
+    try:
+        response = requests.get("http://localhost:8080/debug/movie/tt1674782", timeout=10)
+        if response.status_code == 200:
+            debug_data = response.json()
+            if debug_data.get('detected_import_date'):
+                print(f"✅ Database query works: {debug_data['detected_import_date']}")
+                print(f"   Source: {debug_data['import_source']}")
+                return True
+            else:
+                print(f"❌ No import date found for test movie")
+                return False
+        else:
+            print(f"❌ Debug endpoint failed: {response.status_code}")
+            return False
+    except Exception as e:
+        print(f"❌ Database test failed: {e}")
+        return False
+
+def test_manual_scan():
+    """Test the manual scan functionality"""
+    try:
+        print("🔍 Starting manual movie scan...")
+        response = requests.post(
+            "http://localhost:8080/manual/scan?scan_type=movies", 
+            timeout=30
+        )
+        
+        if response.status_code == 200:
+            scan_data = response.json()
+            print(f"✅ Manual scan started: {scan_data['message']}")
+            
+            # Wait a bit for processing
+            print("⏳ Waiting for scan to process...")
+            time.sleep(10)
+            
+            # Check batch status
+            batch_response = requests.get("http://localhost:8080/batch/status", timeout=10)
+            if batch_response.status_code == 200:
+                batch_data = batch_response.json()
+                print(f"   Pending items: {batch_data['pending_count']}")
+                print(f"   Processing items: {batch_data['processing_count']}")
+            
+            return True
+        else:
+            print(f"❌ Manual scan failed: {response.status_code}")
+            return False
+    except Exception as e:
+        print(f"❌ Manual scan test failed: {e}")
+        return False
+
+def test_stats_endpoint():
+    """Test database stats"""
+    try:
+        response = requests.get("http://localhost:8080/stats", timeout=10)
+        if response.status_code == 200:
+            stats = response.json()
+            print(f"✅ Database stats:")
+            print(f"   Movies: {stats.get('movie_count', 0)}")
+            print(f"   TV Series: {stats.get('series_count', 0)}")
+            print(f"   Episodes: {stats.get('episode_count', 0)}")
+            return True
+        else:
+            print(f"❌ Stats endpoint failed: {response.status_code}")
+            return False
+    except Exception as e:
+        print(f"❌ Stats test failed: {e}")
+        return False
+
+def run_end_to_end_test():
+    """Run complete end-to-end test suite"""
+    print("🎯 NFOGuard End-to-End Testing")
+    print("=" * 50)
+    
+    tests = [
+        ("Server Health", test_server_health),
+        ("Database Performance", test_database_performance),
+        ("Manual Scan", test_manual_scan),
+        ("Database Stats", test_stats_endpoint)
+    ]
+    
+    passed = 0
+    failed = 0
+    
+    for test_name, test_func in tests:
+        print(f"\n🧪 Running: {test_name}")
+        try:
+            if test_func():
+                passed += 1
+                print(f"✅ {test_name} passed")
+            else:
+                failed += 1
+                print(f"❌ {test_name} failed")
+        except Exception as e:
+            failed += 1
+            print(f"❌ {test_name} crashed: {e}")
+    
+    print(f"\n📊 Test Results:")
+    print(f"   ✅ Passed: {passed}")
+    print(f"   ❌ Failed: {failed}")
+    print(f"   📈 Success Rate: {passed}/{passed+failed} ({100*passed/(passed+failed):.1f}%)")
+    
+    return failed == 0
+
+if __name__ == "__main__":
+    success = run_end_to_end_test()
+    
+    if success:
+        print("\n🎉 All end-to-end tests passed!")
+        print("\n📋 Next Steps:")
+        print("1. Run bulk update: python3 bulk_update_movies.py")
+        print("2. Test with real movie: curl http://localhost:8080/debug/movie/tt1674782") 
+        print("3. Check NFO file creation in /media/Movies/movies/")
+        print("4. Verify file timestamps match import dates")
+    else:
+        print("\n💥 Some tests failed - check the logs above")
+        sys.exit(1)