geutebruck/test_action_mapping_upload.py

"""
Test script to verify action mapping upload/download
Tests if Pan Left action is correctly stored and retrieved
"""
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'
    })
    if response.status_code != 200:
        raise Exception(f"Authentication failed")
    return response.json()['access_token']

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

    print("="*80)
    print("TEST: Action Mapping Upload/Download Roundtrip")
    print("="*80)
    print()

    # Step 1: Get the existing mapping
    mapping_name = "GeVi PanLeft_101027"
    print(f"Step 1: Getting existing mapping '{mapping_name}'...")

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

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

    mappings = response.json()['mappings']
    target_mapping = None
    for mapping in mappings:
        if mapping['name'] == mapping_name:
            target_mapping = mapping
            break

    if not target_mapping:
        print(f"ERROR: Mapping '{mapping_name}' not found!")
        print(f"Available mappings: {[m['name'] for m in mappings]}")
        return

    mapping_id = target_mapping['id']
    print(f"Found mapping ID: {mapping_id}")
    print(f"Current output actions:")
    for action in target_mapping.get('output_actions', []):
        print(f"  - {action['action']}")
        print(f"    Parameters: {action.get('parameters', {})}")
    print()

    # Step 2: Update with Pan Left action
    print("Step 2: Updating with GSC Pan Left action...")

    new_output_actions = [
        {
            "action": "PanLeft",
            "parameters": {
                "GscServer": "gscope-cdx-3",
                "PTZ head": "101027"
            }
        }
    ]

    update_data = {
        "name": mapping_name,
        "output_actions": new_output_actions
    }

    print(f"Sending update:")
    print(json.dumps(update_data, indent=2))
    print()

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

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

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

    # Step 3: Download the mapping back
    print("Step 3: Downloading mapping back from server...")

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

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

    mappings = response.json()['mappings']
    downloaded_mapping = None
    for mapping in mappings:
        if mapping['id'] == mapping_id:
            downloaded_mapping = mapping
            break

    if not downloaded_mapping:
        print(f"ERROR: Mapping with ID {mapping_id} not found after update!")
        return

    print(f"Downloaded mapping:")
    print(f"  Name: {downloaded_mapping['name']}")
    print(f"  Output actions:")
    for action in downloaded_mapping.get('output_actions', []):
        print(f"    - Action: {action['action']}")
        print(f"      Parameters: {action.get('parameters', {})}")
    print()

    # Step 4: Verify
    print("Step 4: Verification...")
    downloaded_action = downloaded_mapping['output_actions'][0]

    if downloaded_action['action'] == "PanLeft":
        print("[OK] Action name preserved: PanLeft")
    else:
        print(f"[ERROR] Action name changed: Expected 'PanLeft', got '{downloaded_action['action']}'")

    if 'GscServer' in downloaded_action.get('parameters', {}):
        print(f"[OK] GscServer parameter preserved: {downloaded_action['parameters']['GscServer']}")
    else:
        print(f"[ERROR] GscServer parameter missing or changed")
        print(f"   Parameters: {downloaded_action.get('parameters', {})}")

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

if __name__ == '__main__':
    try:
        test_action_mapping_roundtrip()
    except Exception as e:
        print(f"ERROR: {str(e)}")
        import traceback
        traceback.print_exc()