geutebruck/safe_delete_test.py

"""
Safe Delete Test - Verify the cascade deletion fix
Only creates and deletes 2 test mappings to verify fix works
"""
import asyncio
import sys
sys.path.insert(0, r'C:\DEV\COPILOT\geutebruck-api\src\api\protos')

import grpc
import configuration_pb2
import configuration_pb2_grpc


async def count_mappings():
    """Count total action mappings"""
    channel = grpc.aio.insecure_channel('localhost:50051')
    stub = configuration_pb2_grpc.ConfigurationServiceStub(channel)

    try:
        request = configuration_pb2.ReadActionMappingsRequest()
        response = await stub.ReadActionMappings(request, timeout=10.0)
        count = len(response.mappings)
        await channel.close()
        return count
    except Exception as e:
        print(f"Error reading mappings: {e}")
        await channel.close()
        return 0


async def create_test_mapping(name, contact_num):
    """Create a simple test mapping"""
    channel = grpc.aio.insecure_channel('localhost:50051')
    stub = configuration_pb2_grpc.ConfigurationServiceStub(channel)

    request = configuration_pb2.CreateActionMappingRequest(
        mapping=configuration_pb2.ActionMappingInput(
            name=name,
            input_actions=[
                configuration_pb2.ActionDefinition(
                    action="DigitalContactChanged",
                    parameters=[
                        configuration_pb2.ActionParameter(name="Contact", value=str(contact_num)),
                        configuration_pb2.ActionParameter(name="State", value="closed")
                    ]
                )
            ],
            output_actions=[
                configuration_pb2.ActionDefinition(
                    action="SystemInfo",
                    parameters=[
                        configuration_pb2.ActionParameter(name="Message", value=f"Test {contact_num}")
                    ]
                )
            ]
        )
    )

    try:
        response = await stub.CreateActionMapping(request, timeout=10.0)
        await channel.close()
        return response.mapping.name
    except Exception as e:
        print(f"Error creating mapping: {e}")
        await channel.close()
        return None


async def get_test_mapping_ids():
    """Get IDs of all SAFE_DELETE_TEST mappings"""
    channel = grpc.aio.insecure_channel('localhost:50051')
    stub = configuration_pb2_grpc.ConfigurationServiceStub(channel)

    try:
        request = configuration_pb2.ReadActionMappingsRequest()
        response = await stub.ReadActionMappings(request, timeout=10.0)

        test_ids = []
        for i, mapping in enumerate(response.mappings, 1):
            if "SAFE_DELETE_TEST" in mapping.name:
                test_ids.append(i)

        await channel.close()
        return test_ids
    except Exception as e:
        print(f"Error reading mappings: {e}")
        await channel.close()
        return []


async def delete_mapping(mapping_id):
    """Delete a mapping by ID"""
    channel = grpc.aio.insecure_channel('localhost:50051')
    stub = configuration_pb2_grpc.ConfigurationServiceStub(channel)

    try:
        request = configuration_pb2.DeleteActionMappingRequest(mapping_id=mapping_id)
        response = await stub.DeleteActionMapping(request, timeout=10.0)
        await channel.close()
        return response.success
    except Exception as e:
        print(f"Error deleting mapping {mapping_id}: {e}")
        await channel.close()
        return False


async def main():
    print("=" * 80)
    print("SAFE DELETE TEST - Verifying Cascade Deletion Fix")
    print("=" * 80)

    # 1. Count initial mappings
    print("\n[1] Counting initial mappings...")
    initial_count = await count_mappings()
    print(f"    Initial count: {initial_count}")

    # 2. Create 2 test mappings
    print("\n[2] Creating 2 test mappings...")
    created1 = await create_test_mapping("SAFE_DELETE_TEST - Mapping 1", 91)
    created2 = await create_test_mapping("SAFE_DELETE_TEST - Mapping 2", 92)

    if created1 and created2:
        print(f"    [OK] Created: {created1}")
        print(f"    [OK] Created: {created2}")
    else:
        print("    [ERROR] Failed to create test mappings")
        return

    # 3. Verify they were added
    print("\n[3] Verifying mappings were added...")
    after_create_count = await count_mappings()
    print(f"    Count after create: {after_create_count}")
    print(f"    Expected: {initial_count + 2}")

    if after_create_count == initial_count + 2:
        print("    [PASS] Both mappings created successfully")
    else:
        print(f"    [FAIL] Expected {initial_count + 2}, got {after_create_count}")
        return

    # 4. Get test mapping IDs
    print("\n[4] Finding test mapping IDs...")
    test_ids = await get_test_mapping_ids()
    print(f"    Found test mappings at IDs: {test_ids}")

    if len(test_ids) != 2:
        print(f"    [ERROR] Expected 2 test mappings, found {len(test_ids)}")
        return

    # 5. Delete in REVERSE order (CRITICAL FIX!)
    print("\n[5] Deleting test mappings in REVERSE order (highest first)...")
    test_ids_sorted = sorted(test_ids, reverse=True)
    print(f"    Delete order: {test_ids_sorted}")

    for mapping_id in test_ids_sorted:
        success = await delete_mapping(mapping_id)
        if success:
            print(f"    [OK] Deleted mapping #{mapping_id}")
        else:
            print(f"    [ERROR] Failed to delete mapping #{mapping_id}")

    # 6. Verify they were deleted
    print("\n[6] Verifying mappings were deleted...")
    after_delete_count = await count_mappings()
    print(f"    Count after delete: {after_delete_count}")
    print(f"    Expected: {initial_count}")

    if after_delete_count == initial_count:
        print("    [PASS] Both mappings deleted successfully")
    else:
        print(f"    [FAIL] Expected {initial_count}, got {after_delete_count}")
        delta = initial_count - after_delete_count
        if delta > 0:
            print(f"    [CRITICAL] {delta} additional mappings were deleted!")
        elif delta < 0:
            print(f"    [WARNING] {-delta} mappings remain")

    # 7. Double-check no test mappings remain
    print("\n[7] Final verification...")
    remaining_test_ids = await get_test_mapping_ids()
    if not remaining_test_ids:
        print("    [PASS] No test mappings remain")
    else:
        print(f"    [FAIL] {len(remaining_test_ids)} test mappings still exist: {remaining_test_ids}")

    # Summary
    print("\n" + "=" * 80)
    print("TEST SUMMARY")
    print("=" * 80)
    print(f"  Initial count:      {initial_count}")
    print(f"  After create (+2):  {after_create_count}")
    print(f"  After delete (-2):  {after_delete_count}")
    print(f"  Expected final:     {initial_count}")

    if after_delete_count == initial_count and not remaining_test_ids:
        print("\n  [SUCCESS] Delete fix verified - no cascade deletion occurred!")
    else:
        print("\n  [FAILURE] Delete issue detected - fix may not be working")

    print("=" * 80)


if __name__ == "__main__":
    asyncio.run(main())