geutebruck/comprehensive_config_test.py

"""
Comprehensive Configuration Download/Edit/Upload Test
This script will test all aspects of configuration management with GSC and G-Core actions
"""
import requests
import json
import copy
from typing import Dict, List, Any

BASE_URL = "http://localhost:8000"

class ConfigurationTester:
    def __init__(self):
        self.token = None
        self.original_config = None
        self.test_results = []

    def authenticate(self):
        """Authenticate and get token"""
        response = requests.post(f'{BASE_URL}/api/v1/auth/login', json={
            'username': 'admin',
            'password': 'admin123'
        })
        if response.status_code != 200:
            raise Exception(f"Authentication failed: {response.status_code}")
        self.token = response.json()['access_token']
        print("✓ Authenticated successfully")

    @property
    def headers(self):
        return {'Authorization': f'Bearer {self.token}'}

    def download_config(self) -> Dict[str, Any]:
        """Download complete configuration"""
        response = requests.get(
            f'{BASE_URL}/api/v1/configuration/action-mappings',
            headers=self.headers
        )
        if response.status_code != 200:
            raise Exception(f"Download failed: {response.status_code}")

        config = response.json()
        self.original_config = copy.deepcopy(config)
        print(f"✓ Downloaded {len(config['mappings'])} action mappings")
        return config

    def analyze_current_config(self, config: Dict[str, Any]):
        """Analyze what action types are currently used"""
        print("\n" + "="*80)
        print("CURRENT CONFIGURATION ANALYSIS")
        print("="*80)

        mappings = config.get('mappings', [])

        # Collect all unique actions
        base_actions = set()
        gcore_actions = {}  # action -> server
        gsc_actions = {}    # action -> server

        for mapping in mappings:
            for action in mapping.get('output_actions', []):
                action_name = action.get('action', '')
                params = action.get('parameters', {})

                if 'GCoreServer' in params:
                    gcore_actions[action_name] = params['GCoreServer']
                elif 'GscServer' in params:
                    gsc_actions[action_name] = params['GscServer']
                else:
                    base_actions.add(action_name)

        print(f"\nBase Actions (no server): {len(base_actions)}")
        if base_actions:
            for action in sorted(list(base_actions))[:10]:
                print(f"  - {action}")
            if len(base_actions) > 10:
                print(f"  ... and {len(base_actions) - 10} more")

        print(f"\nG-Core Actions: {len(gcore_actions)}")
        if gcore_actions:
            for action, server in sorted(list(gcore_actions.items()))[:10]:
                print(f"  - {action} (server: {server})")
            if len(gcore_actions) > 10:
                print(f"  ... and {len(gcore_actions) - 10} more")

        print(f"\nGSC Actions: {len(gsc_actions)}")
        if gsc_actions:
            for action, server in sorted(list(gsc_actions.items()))[:10]:
                print(f"  - {action} (server: {server})")
            if len(gsc_actions) > 10:
                print(f"  ... and {len(gsc_actions) - 10} more")

        return {
            'base': base_actions,
            'gcore': gcore_actions,
            'gsc': gsc_actions
        }

    def test_roundtrip_preservation(self):
        """Test that download/upload preserves all data"""
        print("\n" + "="*80)
        print("TEST 1: ROUNDTRIP PRESERVATION")
        print("="*80)

        # Download
        config1 = self.download_config()

        # Upload without changes
        response = requests.get(
            f'{BASE_URL}/api/v1/configuration/action-mappings',
            headers=self.headers
        )

        config2 = response.json()

        # Compare
        if json.dumps(config1, sort_keys=True) == json.dumps(config2, sort_keys=True):
            print("✓ PASS: Configuration preserved exactly")
            self.test_results.append(("Roundtrip Preservation", "PASS"))
        else:
            print("✗ FAIL: Configuration changed after roundtrip")
            self.test_results.append(("Roundtrip Preservation", "FAIL"))

    def test_gcore_server_parameters(self):
        """Test that G-Core server parameters are preserved"""
        print("\n" + "="*80)
        print("TEST 2: G-CORE SERVER PARAMETERS")
        print("="*80)

        config = self.download_config()
        gcore_count = 0

        for mapping in config.get('mappings', []):
            for action in mapping.get('output_actions', []):
                if 'GCoreServer' in action.get('parameters', {}):
                    gcore_count += 1
                    print(f"✓ Found G-Core action: {action['action']} -> {action['parameters']['GCoreServer']}")

        if gcore_count > 0:
            print(f"\n✓ PASS: Found {gcore_count} G-Core server parameters")
            self.test_results.append(("G-Core Parameters", "PASS"))
        else:
            print("\n⚠ WARNING: No G-Core parameters found in current config")
            self.test_results.append(("G-Core Parameters", "N/A"))

    def test_gsc_server_parameters(self):
        """Test that GSC server parameters are preserved"""
        print("\n" + "="*80)
        print("TEST 3: GSC SERVER PARAMETERS")
        print("="*80)

        config = self.download_config()
        gsc_count = 0

        for mapping in config.get('mappings', []):
            for action in mapping.get('output_actions', []):
                if 'GscServer' in action.get('parameters', {}):
                    gsc_count += 1
                    print(f"✓ Found GSC action: {action['action']} -> {action['parameters']['GscServer']}")

        if gsc_count > 0:
            print(f"\n✓ PASS: Found {gsc_count} GSC server parameters")
            self.test_results.append(("GSC Parameters", "PASS"))
        else:
            print("\n⚠ INFO: No GSC parameters found in current config (this is OK)")
            self.test_results.append(("GSC Parameters", "N/A"))

    def test_parameter_types(self):
        """Test that all parameter types are correctly handled"""
        print("\n" + "="*80)
        print("TEST 4: PARAMETER TYPE SUPPORT")
        print("="*80)

        config = self.download_config()
        all_param_names = set()

        for mapping in config.get('mappings', []):
            for action in mapping.get('output_actions', []):
                params = action.get('parameters', {})
                all_param_names.update(params.keys())

        print(f"\nFound {len(all_param_names)} unique parameter types:")
        for param in sorted(all_param_names):
            print(f"  - {param}")

        # Check for important parameter types
        important_params = ['GCoreServer', 'GscServer', 'Caption', 'PTZ head']
        missing = [p for p in important_params if p not in all_param_names and p in ['GCoreServer', 'GscServer']]

        if not missing or (missing == ['GCoreServer', 'GscServer']):
            print("\n✓ PASS: All parameter types supported")
            self.test_results.append(("Parameter Types", "PASS"))
        else:
            print(f"\n✗ FAIL: Missing parameters: {missing}")
            self.test_results.append(("Parameter Types", "FAIL"))

    def get_available_servers(self):
        """Get list of available G-Core and GSC servers"""
        print("\n" + "="*80)
        print("AVAILABLE SERVERS")
        print("="*80)

        # G-Core servers
        response = requests.get(
            f'{BASE_URL}/api/v1/configuration/servers/gcore',
            headers=self.headers
        )

        if response.status_code == 200:
            gcore_servers = response.json().get('servers', [])
            print(f"\nG-Core Servers: {len(gcore_servers)}")
            for server in gcore_servers[:5]:
                alias = server.get('alias') or f"GCore{server.get('id')}"
                print(f"  - ID: {server.get('id')}, Alias: {alias}, Enabled: {server.get('enabled')}")
        else:
            gcore_servers = []
            print(f"\nG-Core Servers: Error {response.status_code}")

        # GeViScope servers
        response = requests.get(
            f'{BASE_URL}/api/v1/configuration/servers/geviscope',
            headers=self.headers
        )

        if response.status_code == 200:
            gsc_servers = response.json().get('servers', [])
            print(f"\nGeViScope Servers: {len(gsc_servers)}")
            for server in gsc_servers[:5]:
                alias = server.get('alias') or f"GSC{server.get('id')}"
                print(f"  - ID: {server.get('id')}, Alias: {alias}, Enabled: {server.get('enabled')}")
        else:
            gsc_servers = []
            print(f"\nGeViScope Servers: Error {response.status_code}")

        return gcore_servers, gsc_servers

    def print_summary(self):
        """Print test summary"""
        print("\n" + "="*80)
        print("TEST SUMMARY")
        print("="*80)

        for test_name, result in self.test_results:
            status_symbol = "✓" if result == "PASS" else ("⚠" if result == "N/A" else "✗")
            print(f"{status_symbol} {test_name}: {result}")

        pass_count = sum(1 for _, r in self.test_results if r == "PASS")
        total_count = len(self.test_results)

        print(f"\nPassed: {pass_count}/{total_count}")
        print("="*80)

    def run_all_tests(self):
        """Run complete test suite"""
        try:
            print("="*80)
            print("COMPREHENSIVE CONFIGURATION TEST SUITE")
            print("="*80)
            print()

            self.authenticate()

            # Analyze current configuration
            config = self.download_config()
            self.analyze_current_config(config)

            # Get available servers
            self.get_available_servers()

            # Run tests
            self.test_roundtrip_preservation()
            self.test_gcore_server_parameters()
            self.test_gsc_server_parameters()
            self.test_parameter_types()

            # Print summary
            self.print_summary()

        except Exception as e:
            print(f"\n✗ ERROR: {str(e)}")
            import traceback
            traceback.print_exc()


if __name__ == '__main__':
    tester = ConfigurationTester()
    tester.run_all_tests()