backend_test.py

#!/usr/bin/env python3
"""
Studio Dashboard Emby Plugin - Backend API Testing
==================================================

This test suite validates the Studio Dashboard plugin API endpoints.
The plugin should be running in an Emby Server instance for these tests to work.

Test Focus:
1. Build verification (requires .NET SDK)
2. API endpoint accessibility without authentication
3. Data structure validation
4. Parameter filtering functionality
5. Collections sync functionality

Expected Emby Server URL: http://localhost:8096 (or configured server)
API Base Path: /emby/StudioDashboard/
"""

import requests
import json
import sys
import subprocess
import os
from typing import Dict, List, Any, Optional
from datetime import datetime

class StudioDashboardTester:
    def __init__(self, base_url: str = "http://localhost:8096"):
        self.base_url = base_url.rstrip('/')
        self.api_base = f"{self.base_url}/emby/StudioDashboard"
        self.session = requests.Session()
        self.test_results = []
        
    def log_test(self, test_name: str, success: bool, message: str, details: Any = None):
        """Log test result"""
        result = {
            "test": test_name,
            "success": success,
            "message": message,
            "details": details,
            "timestamp": datetime.now().isoformat()
        }
        self.test_results.append(result)
        status = "✅ PASS" if success else "❌ FAIL"
        print(f"{status} {test_name}: {message}")
        if details and not success:
            print(f"   Details: {details}")
    
    def test_build_compilation(self) -> bool:
        """Test 1: Verify plugin compilation"""
        print("\n=== BUILD COMPILATION TEST ===")
        
        try:
            # Check if .NET SDK is available
            result = subprocess.run(['dotnet', '--version'], 
                                  capture_output=True, text=True, timeout=10)
            if result.returncode != 0:
                self.log_test("Build - .NET SDK Check", False, 
                            ".NET SDK not available in environment")
                return False
                
            dotnet_version = result.stdout.strip()
            self.log_test("Build - .NET SDK Check", True, 
                        f".NET SDK version: {dotnet_version}")
            
            # Clean the project
            clean_result = subprocess.run(['dotnet', 'clean', 'StudioDashboard.csproj'], 
                                        cwd='/app', capture_output=True, text=True, timeout=30)
            if clean_result.returncode != 0:
                self.log_test("Build - Clean", False, 
                            "Clean failed", clean_result.stderr)
                return False
                
            self.log_test("Build - Clean", True, "Project cleaned successfully")
            
            # Build the project
            build_result = subprocess.run(['dotnet', 'build', 'StudioDashboard.csproj', '-c', 'Release'], 
                                        cwd='/app', capture_output=True, text=True, timeout=60)
            if build_result.returncode != 0:
                self.log_test("Build - Compilation", False, 
                            "Compilation failed", build_result.stderr)
                return False
                
            self.log_test("Build - Compilation", True, 
                        "Plugin compiled successfully")
            
            # Check if DLL was created
            dll_path = "/app/bin/Release/net48/StudioDashboard.dll"
            if os.path.exists(dll_path):
                self.log_test("Build - DLL Creation", True, 
                            f"DLL created at {dll_path}")
                return True
            else:
                self.log_test("Build - DLL Creation", False, 
                            f"DLL not found at expected path: {dll_path}")
                return False
                
        except subprocess.TimeoutExpired:
            self.log_test("Build - Compilation", False, "Build process timed out")
            return False
        except FileNotFoundError:
            self.log_test("Build - .NET SDK Check", False, 
                        ".NET SDK not found in system PATH")
            return False
        except Exception as e:
            self.log_test("Build - Compilation", False, 
                        f"Unexpected error during build: {str(e)}")
            return False
    
    def test_server_connectivity(self) -> bool:
        """Test 2: Check if Emby Server is accessible"""
        print("\n=== SERVER CONNECTIVITY TEST ===")
        
        try:
            response = self.session.get(f"{self.base_url}/System/Info", timeout=10)
            if response.status_code == 200:
                server_info = response.json()
                server_name = server_info.get('ServerName', 'Unknown')
                version = server_info.get('Version', 'Unknown')
                self.log_test("Server - Connectivity", True, 
                            f"Connected to {server_name} v{version}")
                return True
            else:
                self.log_test("Server - Connectivity", False, 
                            f"Server returned status {response.status_code}")
                return False
        except requests.exceptions.RequestException as e:
            self.log_test("Server - Connectivity", False, 
                        f"Cannot connect to Emby Server: {str(e)}")
            return False
    
    def test_api_test_endpoint(self) -> bool:
        """Test 3: Test the /Test endpoint"""
        print("\n=== API TEST ENDPOINT ===")
        
        try:
            response = self.session.get(f"{self.api_base}/Test", timeout=10)
            
            if response.status_code == 401:
                self.log_test("API - Test Endpoint Auth", False, 
                            "Authentication required - [AllowAnonymous] not working")
                return False
            elif response.status_code == 404:
                self.log_test("API - Test Endpoint", False, 
                            "Endpoint not found - plugin may not be loaded")
                return False
            elif response.status_code != 200:
                self.log_test("API - Test Endpoint", False, 
                            f"Unexpected status code: {response.status_code}")
                return False
            
            try:
                data = response.json()
                expected_fields = ['Status', 'Message', 'Timestamp', 'PluginVersion']
                
                if data.get('Status') == 'OK':
                    self.log_test("API - Test Endpoint", True, 
                                f"Test endpoint working: {data.get('Message')}")
                    
                    # Validate response structure
                    missing_fields = [field for field in expected_fields if field not in data]
                    if missing_fields:
                        self.log_test("API - Test Response Structure", False, 
                                    f"Missing fields: {missing_fields}", data)
                    else:
                        self.log_test("API - Test Response Structure", True, 
                                    "All expected fields present")
                    return True
                else:
                    self.log_test("API - Test Endpoint", False, 
                                f"Status not OK: {data.get('Status')}", data)
                    return False
                    
            except json.JSONDecodeError:
                self.log_test("API - Test Endpoint", False, 
                            "Response is not valid JSON", response.text)
                return False
                
        except requests.exceptions.RequestException as e:
            self.log_test("API - Test Endpoint", False, 
                        f"Request failed: {str(e)}")
            return False
    
    def test_studios_with_logos_endpoint(self) -> bool:
        """Test 4: Test the main StudiosWithLogos endpoint"""
        print("\n=== STUDIOS WITH LOGOS ENDPOINT ===")
        
        success_count = 0
        total_tests = 4
        
        # Test 4a: Basic endpoint without parameters
        try:
            response = self.session.get(f"{self.api_base}/StudiosWithLogos", timeout=15)
            
            if response.status_code == 401:
                self.log_test("API - StudiosWithLogos Auth", False, 
                            "Authentication required - [AllowAnonymous] not working")
                return False
            elif response.status_code == 404:
                self.log_test("API - StudiosWithLogos", False, 
                            "Endpoint not found - plugin may not be loaded")
                return False
            elif response.status_code != 200:
                self.log_test("API - StudiosWithLogos", False, 
                            f"Unexpected status code: {response.status_code}")
                return False
            
            try:
                data = response.json()
                if isinstance(data, list):
                    self.log_test("API - StudiosWithLogos Basic", True, 
                                f"Returned {len(data)} studios")
                    success_count += 1
                    
                    # Validate structure of first item if available
                    if data:
                        studio = data[0]
                        expected_fields = ['Studio', 'Count', 'LogoUrl', 'BackgroundColor', 
                                         'TextColor', 'IsStreamingService']
                        missing_fields = [field for field in expected_fields if field not in studio]
                        
                        if missing_fields:
                            self.log_test("API - StudiosWithLogos Structure", False, 
                                        f"Missing fields in studio object: {missing_fields}", studio)
                        else:
                            self.log_test("API - StudiosWithLogos Structure", True, 
                                        "Studio object structure is correct")
                            success_count += 1
                    else:
                        self.log_test("API - StudiosWithLogos Structure", False, 
                                    "No studios returned - may indicate empty library")
                else:
                    self.log_test("API - StudiosWithLogos Basic", False, 
                                "Response is not a list", type(data))
                    
            except json.JSONDecodeError:
                self.log_test("API - StudiosWithLogos Basic", False, 
                            "Response is not valid JSON", response.text)
                
        except requests.exceptions.RequestException as e:
            self.log_test("API - StudiosWithLogos Basic", False, 
                        f"Request failed: {str(e)}")
        
        # Test 4b: Test with streaming filter
        try:
            response = self.session.get(f"{self.api_base}/StudiosWithLogos?Filter=streaming", timeout=15)
            
            if response.status_code == 200:
                data = response.json()
                if isinstance(data, list):
                    streaming_count = sum(1 for studio in data if studio.get('IsStreamingService', False))
                    self.log_test("API - StudiosWithLogos Streaming Filter", True, 
                                f"Returned {len(data)} studios, {streaming_count} are streaming services")
                    success_count += 1
                else:
                    self.log_test("API - StudiosWithLogos Streaming Filter", False, 
                                "Response is not a list")
            else:
                self.log_test("API - StudiosWithLogos Streaming Filter", False, 
                            f"Status code: {response.status_code}")
                
        except Exception as e:
            self.log_test("API - StudiosWithLogos Streaming Filter", False, 
                        f"Request failed: {str(e)}")
        
        # Test 4c: Test with Top parameter
        try:
            response = self.session.get(f"{self.api_base}/StudiosWithLogos?Top=5", timeout=15)
            
            if response.status_code == 200:
                data = response.json()
                if isinstance(data, list):
                    if len(data) <= 5:
                        self.log_test("API - StudiosWithLogos Top Parameter", True, 
                                    f"Correctly limited to {len(data)} studios (requested 5)")
                        success_count += 1
                    else:
                        self.log_test("API - StudiosWithLogos Top Parameter", False, 
                                    f"Returned {len(data)} studios, expected max 5")
                else:
                    self.log_test("API - StudiosWithLogos Top Parameter", False, 
                                "Response is not a list")
            else:
                self.log_test("API - StudiosWithLogos Top Parameter", False, 
                            f"Status code: {response.status_code}")
                
        except Exception as e:
            self.log_test("API - StudiosWithLogos Top Parameter", False, 
                        f"Request failed: {str(e)}")
        
        return success_count >= 2  # At least basic functionality should work
    
    def test_sync_collections_endpoint(self) -> bool:
        """Test 5: Test the SyncCollections endpoint"""
        print("\n=== SYNC COLLECTIONS ENDPOINT ===")
        
        try:
            response = self.session.post(f"{self.api_base}/SyncCollections", timeout=20)
            
            if response.status_code == 401:
                self.log_test("API - SyncCollections Auth", False, 
                            "Authentication required - [AllowAnonymous] not working")
                return False
            elif response.status_code == 404:
                self.log_test("API - SyncCollections", False, 
                            "Endpoint not found - plugin may not be loaded")
                return False
            elif response.status_code != 200:
                self.log_test("API - SyncCollections", False, 
                            f"Unexpected status code: {response.status_code}")
                return False
            
            try:
                data = response.json()
                if data.get('Started') is True:
                    self.log_test("API - SyncCollections", True, 
                                f"Collections sync started: {data.get('Message')}")
                    return True
                else:
                    self.log_test("API - SyncCollections", False, 
                                f"Sync not started: {data.get('Message')}", data)
                    return False
                    
            except json.JSONDecodeError:
                self.log_test("API - SyncCollections", False, 
                            "Response is not valid JSON", response.text)
                return False
                
        except requests.exceptions.RequestException as e:
            self.log_test("API - SyncCollections", False, 
                        f"Request failed: {str(e)}")
            return False
    
    def test_additional_endpoints(self) -> bool:
        """Test 6: Test additional endpoints"""
        print("\n=== ADDITIONAL ENDPOINTS ===")
        
        success_count = 0
        total_tests = 3
        
        # Test StudiosNormalized endpoint
        try:
            response = self.session.get(f"{self.api_base}/StudiosNormalized", timeout=15)
            if response.status_code == 200:
                data = response.json()
                if isinstance(data, list):
                    self.log_test("API - StudiosNormalized", True, 
                                f"Returned {len(data)} normalized studios")
                    success_count += 1
                else:
                    self.log_test("API - StudiosNormalized", False, 
                                "Response is not a list")
            else:
                self.log_test("API - StudiosNormalized", False, 
                            f"Status code: {response.status_code}")
        except Exception as e:
            self.log_test("API - StudiosNormalized", False, f"Request failed: {str(e)}")
        
        # Test ExportJson endpoint
        try:
            response = self.session.get(f"{self.api_base}/ExportJson", timeout=15)
            if response.status_code == 200:
                data = response.json()
                if isinstance(data, list):
                    self.log_test("API - ExportJson", True, 
                                f"Exported {len(data)} studios as JSON")
                    success_count += 1
                else:
                    self.log_test("API - ExportJson", False, 
                                "Response is not a list")
            else:
                self.log_test("API - ExportJson", False, 
                            f"Status code: {response.status_code}")
        except Exception as e:
            self.log_test("API - ExportJson", False, f"Request failed: {str(e)}")
        
        # Test ExportCsv endpoint
        try:
            response = self.session.get(f"{self.api_base}/ExportCsv", timeout=15)
            if response.status_code == 200:
                csv_data = response.text
                if csv_data.startswith("Studio,Count"):
                    self.log_test("API - ExportCsv", True, 
                                f"Exported CSV data ({len(csv_data)} characters)")
                    success_count += 1
                else:
                    self.log_test("API - ExportCsv", False, 
                                "CSV format incorrect", csv_data[:100])
            else:
                self.log_test("API - ExportCsv", False, 
                            f"Status code: {response.status_code}")
        except Exception as e:
            self.log_test("API - ExportCsv", False, f"Request failed: {str(e)}")
        
        return success_count >= 1  # At least one additional endpoint should work
    
    def run_all_tests(self) -> Dict[str, Any]:
        """Run all tests and return summary"""
        print("🚀 Starting Studio Dashboard Plugin Backend Tests")
        print(f"Target Server: {self.base_url}")
        print("=" * 60)
        
        test_functions = [
            ("Build Compilation", self.test_build_compilation),
            ("Server Connectivity", self.test_server_connectivity),
            ("API Test Endpoint", self.test_api_test_endpoint),
            ("Studios With Logos", self.test_studios_with_logos_endpoint),
            ("Sync Collections", self.test_sync_collections_endpoint),
            ("Additional Endpoints", self.test_additional_endpoints)
        ]
        
        results = {}
        total_passed = 0
        total_tests = len(test_functions)
        
        for test_name, test_func in test_functions:
            try:
                passed = test_func()
                results[test_name] = passed
                if passed:
                    total_passed += 1
            except Exception as e:
                print(f"❌ CRITICAL ERROR in {test_name}: {str(e)}")
                results[test_name] = False
        
        print("\n" + "=" * 60)
        print("📊 TEST SUMMARY")
        print("=" * 60)
        
        for test_name, passed in results.items():
            status = "✅ PASS" if passed else "❌ FAIL"
            print(f"{status} {test_name}")
        
        print(f"\nOverall: {total_passed}/{total_tests} test categories passed")
        
        # Detailed results
        summary = {
            "total_tests": len(self.test_results),
            "passed_tests": len([r for r in self.test_results if r["success"]]),
            "failed_tests": len([r for r in self.test_results if not r["success"]]),
            "test_categories": results,
            "detailed_results": self.test_results,
            "overall_success": total_passed >= 3  # At least half should pass
        }
        
        return summary

def main():
    """Main test execution"""
    import argparse
    
    parser = argparse.ArgumentParser(description='Test Studio Dashboard Emby Plugin')
    parser.add_argument('--server', default='http://localhost:8096', 
                       help='Emby Server URL (default: http://localhost:8096)')
    parser.add_argument('--output', help='Output file for test results (JSON)')
    
    args = parser.parse_args()
    
    tester = StudioDashboardTester(args.server)
    results = tester.run_all_tests()
    
    if args.output:
        with open(args.output, 'w') as f:
            json.dump(results, f, indent=2)
        print(f"\n📄 Detailed results saved to: {args.output}")
    
    # Exit with appropriate code
    sys.exit(0 if results["overall_success"] else 1)

if __name__ == "__main__":
    main()