geutebruck/test_comprehensive_roundtrip.py

"""
Comprehensive test of upload/download roundtrip for GSC/G-Core actions
Tests the complete flow and logs every step
"""
import requests
import json

BASE_URL = "http://localhost:8000"

def authenticate():
    response = requests.post(f'{BASE_URL}/api/v1/auth/login', json={
        'username': 'admin',
        'password': 'admin123'
    })
    return response.json()['access_token']

def test_roundtrip():
    token = authenticate()
    headers = {'Authorization': f'Bearer {token}'}

    print("=" * 80)
    print("COMPREHENSIVE ROUNDTRIP TEST")
    print("=" * 80)
    print()

    # Step 1: Delete any existing test mapping
    print("Step 1: Cleaning up existing test mappings...")
    response = requests.get(f'{BASE_URL}/api/v1/configuration/action-mappings', headers=headers)
    for mapping in response.json()['mappings']:
        if mapping['name'] == 'ROUNDTRIP_TEST':
            print(f"  Deleting existing mapping ID {mapping['id']}")
            requests.delete(f'{BASE_URL}/api/v1/configuration/action-mappings/{mapping['id']}', headers=headers)
    print()

    # Step 2: Create new mapping with GSC and G-Core actions
    print("Step 2: Creating new mapping with GSC and G-Core actions...")
    new_mapping = {
        "name": "ROUNDTRIP_TEST",
        "input_action": "ROUNDTRIP_TEST",
        "output_actions": [
            {
                "action": "PanLeft",
                "parameters": {
                    "Caption": "GSC PanLeft Test",
                    "GscServer": "gscope-cdx-3",
                    "PTZ head": "101027"
                }
            },
            {
                "action": "PanRight",
                "parameters": {
                    "Caption": "G-Core PanRight Test",
                    "GCoreServer": "gscope-cdu-3",
                    "PTZ head": "101028"
                }
            }
        ],
        "enabled": True
    }

    print("Creating mapping:")
    print(json.dumps(new_mapping, indent=2))
    print()

    response = requests.post(
        f'{BASE_URL}/api/v1/configuration/action-mappings',
        headers=headers,
        json=new_mapping
    )

    if response.status_code not in [200, 201]:
        print(f"ERROR: Failed to create mapping: {response.status_code}")
        print(response.text)
        return

    result = response.json()
    print("Response:")
    print(json.dumps(result, indent=2))
    mapping_id = result.get('mapping', {}).get('id') or result.get('id')
    print(f"[OK] Mapping created with ID: {mapping_id}")
    print()

    # Step 3: Read back from database BEFORE upload
    print("Step 3: Reading back from database BEFORE upload...")
    response = requests.get(
        f'{BASE_URL}/api/v1/configuration/action-mappings/{mapping_id}',
        headers=headers
    )

    if response.status_code != 200:
        print(f"ERROR: Failed to read mapping: {response.status_code}")
        return

    db_mapping = response.json()
    print("Output actions from database:")
    for i, action in enumerate(db_mapping['output_actions'], 1):
        print(f"\n  Action {i}: {action['action']}")
        print(f"  Parameters: {json.dumps(action.get('parameters', {}), indent=4)}")
    print()

    # Step 4: Upload to GeViServer
    print("Step 4: Uploading to GeViServer...")
    response = requests.post(
        f'{BASE_URL}/api/v1/configuration/upload',
        headers=headers
    )

    if response.status_code != 200:
        print(f"ERROR: Upload failed: {response.status_code}")
        print(response.text)
        return

    print("[OK] Upload successful")
    print()

    # Step 5: Download from GeViServer
    print("Step 5: Downloading from GeViServer...")
    response = requests.get(
        f'{BASE_URL}/api/v1/configuration/download',
        headers=headers
    )

    if response.status_code != 200:
        print(f"ERROR: Download failed: {response.status_code}")
        print(response.text)
        return

    print("[OK] Download successful")
    print()

    # Step 6: Find our mapping in the downloaded data
    print("Step 6: Finding ROUNDTRIP_TEST in downloaded configuration...")
    downloaded_config = response.json()

    found_mapping = None
    for mapping in downloaded_config.get('actionMappings', {}).get('mappings', []):
        if mapping.get('name') == 'ROUNDTRIP_TEST':
            found_mapping = mapping
            break

    if not found_mapping:
        print("ERROR: ROUNDTRIP_TEST not found in downloaded configuration!")
        return

    print("[OK] Found mapping in downloaded configuration")
    print()

    # Step 7: Compare actions
    print("Step 7: Comparing output actions...")
    print()

    downloaded_actions = found_mapping.get('outputActions', [])
    print(f"Number of output actions: {len(downloaded_actions)}")
    print()

    for i, action in enumerate(downloaded_actions, 1):
        print(f"Output Action {i}:")
        print(f"  action: {action.get('action', 'MISSING')}")
        print(f"  parameters:")
        for key, value in action.get('parameters', {}).items():
            print(f"    {key}: {value}")
        print()

        # Check for issues
        issues = []
        if action.get('action') == 'CrossSwitch':
            issues.append("[WARNING]  Action shows as 'CrossSwitch' instead of original action name!")
        if 'SwitchMode' in action.get('parameters', {}):
            issues.append("[WARNING]  Has invalid 'SwitchMode' parameter!")
        if 'VideoInput' in action.get('parameters', {}):
            issues.append("[WARNING]  Has invalid 'VideoInput' parameter!")
        if 'VideoOutput' in action.get('parameters', {}):
            issues.append("[WARNING]  Has invalid 'VideoOutput' parameter!")
        if 'GscServer' not in action.get('parameters', {}) and 'GCoreServer' not in action.get('parameters', {}):
            issues.append("[WARNING]  Missing server parameter (GscServer or GCoreServer)!")

        if issues:
            print("  ISSUES FOUND:")
            for issue in issues:
                print(f"    {issue}")
        else:
            print("  [OK] Action looks good!")
        print()

    # Step 8: Read back from database AFTER upload/download
    print("Step 8: Reading back from database AFTER upload/download cycle...")
    response = requests.get(
        f'{BASE_URL}/api/v1/configuration/action-mappings/{mapping_id}',
        headers=headers
    )

    if response.status_code != 200:
        print(f"ERROR: Failed to read mapping: {response.status_code}")
        return

    db_mapping_after = response.json()
    print("Output actions from database AFTER cycle:")
    for i, action in enumerate(db_mapping_after['output_actions'], 1):
        print(f"\n  Action {i}: {action['action']}")
        print(f"  Parameters: {json.dumps(action.get('parameters', {}), indent=4)}")
    print()

    print("=" * 80)
    print("TEST COMPLETE")
    print("=" * 80)

if __name__ == '__main__':
    test_roundtrip()