geutebruck/test_grpc_mapping.py

"""
Test script for creating action mapping directly via gRPC (SDK Bridge)
This bypasses the REST API and authentication layer
"""
import grpc
import sys
sys.path.append(r'C:\DEV\COPILOT\geutebruck-api\src\api')

from protos import configuration_pb2
from protos import configuration_pb2_grpc

def test_create_mapping_grpc():
    """Test creating mapping via gRPC"""

    print("=" * 70)
    print("TESTING ACTION MAPPING CREATION VIA GRPC")
    print("=" * 70)

    # Connect to gRPC service
    channel = grpc.insecure_channel('localhost:50051')
    stub = configuration_pb2_grpc.ConfigurationServiceStub(channel)

    # Step 1: Get current mappings count
    print("\n1. Getting current action mappings count...")
    try:
        list_request = configuration_pb2.ReadActionMappingsRequest()
        list_response = stub.ReadActionMappings(list_request)

        initial_count = len(list_response.mappings)
        print(f"   Initial mapping count: {initial_count}")

        print(f"\n   Current mappings:")
        for i, mapping in enumerate(list_response.mappings[:5]):  # Show first 5
            try:
                print(f"      [{i+1}] {mapping.name}")
            except UnicodeEncodeError:
                print(f"      [{i+1}] <mapping with special characters>")
        if len(list_response.mappings) > 5:
            print(f"      ... and {len(list_response.mappings) - 5} more")

    except grpc.RpcError as e:
        print(f"   ERROR: {e.code()}: {e.details()}")
        return False

    # Step 2: Create new mapping
    print("\n2. Creating new action mapping via gRPC...")

    # Create action definitions with parameters
    param1_1 = configuration_pb2.ActionParameter(name="PreAlarm", value="10")
    param1_2 = configuration_pb2.ActionParameter(name="Alarm", value="60")
    output_action1 = configuration_pb2.ActionDefinition(
        action="GCoreDataBase",
        parameters=[param1_1, param1_2]
    )

    param2_1 = configuration_pb2.ActionParameter(name="Receiver", value="test@example.com")
    param2_2 = configuration_pb2.ActionParameter(name="Subject", value="Test Alert from gRPC")
    output_action2 = configuration_pb2.ActionDefinition(
        action="GscMail",
        parameters=[param2_1, param2_2]
    )

    new_mapping = configuration_pb2.ActionMappingInput(
        name="GRPC_TEST_FOLDER_TREE_MAPPING",
        output_actions=[output_action1, output_action2]
    )

    create_request = configuration_pb2.CreateActionMappingRequest(
        mapping=new_mapping
    )

    print(f"   Mapping name: {new_mapping.name}")
    print(f"   Output actions: {len(new_mapping.output_actions)}")
    for i, action_def in enumerate(new_mapping.output_actions):
        print(f"      [{i+1}] {action_def.action} ({len(action_def.parameters)} params)")

    try:
        create_response = stub.CreateActionMapping(create_request)

        print(f"\n   gRPC Response:")
        print(f"      Success: {create_response.success}")
        print(f"      Message: {create_response.message}")

        if not create_response.success:
            print("   ERROR: CreateActionMapping returned success=False")
            return False

    except grpc.RpcError as e:
        print(f"   ERROR: {e.code()}: {e.details()}")
        return False

    # Step 3: Verify persistence
    print("\n3. Verifying persistence by reading back...")
    try:
        verify_request = configuration_pb2.ReadActionMappingsRequest()
        verify_response = stub.ReadActionMappings(verify_request)

        final_count = len(verify_response.mappings)
        print(f"   Initial count: {initial_count}")
        print(f"   Final count: {final_count}")

        if final_count == initial_count + 1:
            print("   [SUCCESS] Mapping count increased by 1 - SUCCESS!")
        else:
            print(f"   [FAIL] Mapping count did NOT increase! Expected {initial_count + 1}, got {final_count}")
            print("   THIS WAS THE CRITICAL BUG IN PREVIOUS IMPLEMENTATION!")
            return False

    except grpc.RpcError as e:
        print(f"   ERROR: {e.code()}: {e.details()}")
        return False

    # Step 4: Find the new mapping
    print("\n4. Searching for new mapping in list...")
    print(f"   Showing last 5 mappings (most recent):")
    for i, mapping in enumerate(verify_response.mappings[-5:]):
        idx = len(verify_response.mappings) - 5 + i + 1
        try:
            print(f"      [{idx}] {mapping.name[:50]}")
        except (UnicodeEncodeError, UnicodeDecodeError):
            print(f"      [{idx}] <special characters>")

    found = False
    for mapping in verify_response.mappings:
        try:
            if mapping.name == "GRPC_TEST_FOLDER_TREE_MAPPING":
                found = True
                print(f"\n   [SUCCESS] Found new mapping!")
                print(f"      Name: {mapping.name}")
                print(f"      Output actions: {len(mapping.output_actions)}")
                for i, action in enumerate(mapping.output_actions):
                    print(f"         [{i+1}] {action.action}")
                break
        except (UnicodeEncodeError, UnicodeDecodeError):
            continue

    if not found:
        print("   [FAIL] New mapping NOT found in list!")
        return False

    print("\n" + "=" * 70)
    print("SUCCESS! ALL TESTS PASSED!")
    print("=" * 70)
    print("\nThe folder tree implementation is working correctly:")
    print("  ✓ Configuration parsed completely")
    print("  ✓ New mapping added to tree")
    print("  ✓ Binary serialization successful")
    print("  ✓ GeViServer accepted changes")
    print("  ✓ Mapping persisted across read-back")
    print("=" * 70)

    return True

if __name__ == "__main__":
    try:
        success = test_create_mapping_grpc()
        exit(0 if success else 1)
    except Exception as e:
        print(f"\nEXCEPTION: {e}")
        import traceback
        traceback.print_exc()
        exit(1)