add proxy tests for the API Gateway v2 service

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: whummer

aws-proxy/tests/proxy/test_apigatewayv2.py

@@ -0,0 +1,282 @@
+# Note/disclosure: This file has been (partially or fully) generated by an AI agent.
+import logging
+
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+from localstack.aws.connect import connect_to
+from localstack.utils.strings import short_uid
+
+from aws_proxy.shared.models import ProxyConfig
+
+logger = logging.getLogger(__name__)
+
+
+def test_apigatewayv2_http_api_requests(start_aws_proxy, cleanups):
+    """Test basic API Gateway v2 HTTP API operations with proxy."""
+    api_name_aws = f"test-http-api-{short_uid()}"
+
+    # start proxy - forwarding all API Gateway v2 requests
+    config = ProxyConfig(services={"apigatewayv2": {"resources": ".*"}})
+    start_aws_proxy(config)
+
+    # create clients
+    region_name = "us-east-1"
+    apigwv2_client = connect_to(region_name=region_name).apigatewayv2
+    apigwv2_client_aws = boto3.client("apigatewayv2", region_name=region_name)
+
+    # create HTTP API in AWS
+    create_response_aws = apigwv2_client_aws.create_api(
+        Name=api_name_aws, ProtocolType="HTTP", Description="Test HTTP API for proxy"
+    )
+    api_id_aws = create_response_aws["ApiId"]
+    cleanups.append(lambda: apigwv2_client_aws.delete_api(ApiId=api_id_aws))
+
+    # assert that local call for this API is proxied
+    get_api_local = apigwv2_client.get_api(ApiId=api_id_aws)
+    get_api_aws = apigwv2_client_aws.get_api(ApiId=api_id_aws)
+    assert get_api_local["Name"] == get_api_aws["Name"] == api_name_aws
+    assert get_api_local["ApiId"] == get_api_aws["ApiId"] == api_id_aws
+
+    # negative test: verify that requesting a non-existent API fails
+    with pytest.raises(ClientError) as ctx:
+        apigwv2_client_aws.get_api(ApiId="nonexistent123")
+    assert ctx.value.response["Error"]["Code"] == "NotFoundException"
+
+    # list APIs from AWS should include the created API
+    apis_aws = apigwv2_client_aws.get_apis()
+    aws_api_ids = [api["ApiId"] for api in apis_aws.get("Items", [])]
+    assert api_id_aws in aws_api_ids
+
+    # update API description via LocalStack client (should proxy to AWS)
+    updated_description = "Updated description via proxy"
+    apigwv2_client.update_api(ApiId=api_id_aws, Description=updated_description)
+
+    # verify update is reflected in AWS
+    get_api_aws = apigwv2_client_aws.get_api(ApiId=api_id_aws)
+    assert get_api_aws["Description"] == updated_description
+
+
+def test_apigatewayv2_routes_and_integrations(start_aws_proxy, cleanups):
+    """Test API Gateway v2 routes and integrations with proxy."""
+    api_name_aws = f"test-http-api-routes-{short_uid()}"
+
+    # start proxy - forwarding all API Gateway v2 requests
+    config = ProxyConfig(services={"apigatewayv2": {"resources": ".*"}})
+    start_aws_proxy(config)
+
+    # create clients
+    region_name = "us-east-1"
+    apigwv2_client = connect_to(region_name=region_name).apigatewayv2
+    apigwv2_client_aws = boto3.client("apigatewayv2", region_name=region_name)
+
+    # create HTTP API in AWS
+    create_response_aws = apigwv2_client_aws.create_api(
+        Name=api_name_aws, ProtocolType="HTTP"
+    )
+    api_id_aws = create_response_aws["ApiId"]
+    cleanups.append(lambda: apigwv2_client_aws.delete_api(ApiId=api_id_aws))
+
+    # create integration via AWS client
+    integration_response = apigwv2_client_aws.create_integration(
+        ApiId=api_id_aws,
+        IntegrationType="HTTP_PROXY",
+        IntegrationMethod="GET",
+        IntegrationUri="https://httpbin.org/get",
+        PayloadFormatVersion="1.0",
+    )
+    integration_id = integration_response["IntegrationId"]
+
+    # verify integration via LocalStack (proxied)
+    integration_local = apigwv2_client.get_integration(
+        ApiId=api_id_aws, IntegrationId=integration_id
+    )
+    assert integration_local["IntegrationType"] == "HTTP_PROXY"
+    assert integration_local["IntegrationId"] == integration_id
+
+    # create route via AWS client
+    route_response = apigwv2_client_aws.create_route(
+        ApiId=api_id_aws,
+        RouteKey="GET /users",
+        Target=f"integrations/{integration_id}",
+    )
+    route_id = route_response["RouteId"]
+
+    # verify route via LocalStack (proxied)
+    route_local = apigwv2_client.get_route(ApiId=api_id_aws, RouteId=route_id)
+    assert route_local["RouteKey"] == "GET /users"
+    assert route_local["RouteId"] == route_id
+
+    # list routes via LocalStack (should proxy to AWS)
+    routes_local = apigwv2_client.get_routes(ApiId=api_id_aws)
+    route_ids_local = [r["RouteId"] for r in routes_local.get("Items", [])]
+    assert route_id in route_ids_local
+
+    # delete route via AWS client
+    apigwv2_client_aws.delete_route(ApiId=api_id_aws, RouteId=route_id)
+
+    # verify route is deleted via LocalStack (proxied)
+    with pytest.raises(ClientError) as ctx:
+        apigwv2_client.get_route(ApiId=api_id_aws, RouteId=route_id)
+    assert ctx.value.response["Error"]["Code"] == "NotFoundException"
+
+
+def test_apigatewayv2_stages_and_deployments(start_aws_proxy, cleanups):
+    """Test API Gateway v2 stages and deployments with proxy."""
+    api_name_aws = f"test-http-api-deploy-{short_uid()}"
+
+    # start proxy - forwarding all API Gateway v2 requests
+    config = ProxyConfig(services={"apigatewayv2": {"resources": ".*"}})
+    start_aws_proxy(config)
+
+    # create clients
+    region_name = "us-east-1"
+    apigwv2_client = connect_to(region_name=region_name).apigatewayv2
+    apigwv2_client_aws = boto3.client("apigatewayv2", region_name=region_name)
+
+    # create HTTP API in AWS
+    create_response_aws = apigwv2_client_aws.create_api(
+        Name=api_name_aws, ProtocolType="HTTP"
+    )
+    api_id_aws = create_response_aws["ApiId"]
+    cleanups.append(lambda: apigwv2_client_aws.delete_api(ApiId=api_id_aws))
+
+    # create a simple integration and route (required for deployment)
+    integration_response = apigwv2_client_aws.create_integration(
+        ApiId=api_id_aws,
+        IntegrationType="HTTP_PROXY",
+        IntegrationMethod="GET",
+        IntegrationUri="https://httpbin.org/get",
+        PayloadFormatVersion="1.0",
+    )
+    integration_id = integration_response["IntegrationId"]
+
+    apigwv2_client_aws.create_route(
+        ApiId=api_id_aws,
+        RouteKey="GET /test",
+        Target=f"integrations/{integration_id}",
+    )
+
+    # create stage via LocalStack client (should proxy to AWS)
+    stage_name = "test"
+    apigwv2_client.create_stage(
+        ApiId=api_id_aws, StageName=stage_name, Description="Test stage"
+    )
+
+    # verify stage exists in AWS
+    stage_aws = apigwv2_client_aws.get_stage(ApiId=api_id_aws, StageName=stage_name)
+    assert stage_aws["StageName"] == stage_name
+    assert stage_aws["Description"] == "Test stage"
+
+    # get stage via LocalStack client
+    stage_local = apigwv2_client.get_stage(ApiId=api_id_aws, StageName=stage_name)
+    assert stage_local["StageName"] == stage_aws["StageName"]
+
+    # create deployment via AWS client
+    deployment_response = apigwv2_client_aws.create_deployment(
+        ApiId=api_id_aws, StageName=stage_name, Description="Test deployment"
+    )
+    deployment_id = deployment_response["DeploymentId"]
+
+    # verify deployment via LocalStack (proxied)
+    deployment_local = apigwv2_client.get_deployment(
+        ApiId=api_id_aws, DeploymentId=deployment_id
+    )
+    assert deployment_local["DeploymentId"] == deployment_id
+
+    # list stages via AWS client
+    stages_aws = apigwv2_client_aws.get_stages(ApiId=api_id_aws)
+    stage_names_aws = [s["StageName"] for s in stages_aws.get("Items", [])]
+    assert stage_name in stage_names_aws
+
+
+def test_apigatewayv2_read_only_mode(start_aws_proxy, cleanups):
+    """Test API Gateway v2 operations in read-only proxy mode."""
+    api_name_aws = f"test-http-api-readonly-{short_uid()}"
+
+    # create HTTP API in AWS first (before starting proxy)
+    region_name = "us-east-1"
+    apigwv2_client_aws = boto3.client("apigatewayv2", region_name=region_name)
+    create_response_aws = apigwv2_client_aws.create_api(
+        Name=api_name_aws, ProtocolType="HTTP"
+    )
+    api_id_aws = create_response_aws["ApiId"]
+    cleanups.append(lambda: apigwv2_client_aws.delete_api(ApiId=api_id_aws))
+
+    # start proxy in read-only mode
+    config = ProxyConfig(
+        services={"apigatewayv2": {"resources": ".*", "read_only": True}}
+    )
+    start_aws_proxy(config)
+
+    # create LocalStack client
+    apigwv2_client = connect_to(region_name=region_name).apigatewayv2
+
+    # read operations should work (proxied to AWS)
+    get_api_local = apigwv2_client.get_api(ApiId=api_id_aws)
+    assert get_api_local["Name"] == api_name_aws
+    assert get_api_local["ApiId"] == api_id_aws
+
+    # verify the API can also be read directly from AWS
+    get_api_aws = apigwv2_client_aws.get_api(ApiId=api_id_aws)
+    assert get_api_local["Name"] == get_api_aws["Name"]
+
+    # write operations should not be proxied in read-only mode
+    original_description = get_api_aws.get("Description", "")
+
+    # Attempt write operation - should either be blocked or not proxied
+    try:
+        apigwv2_client.update_api(
+            ApiId=api_id_aws, Description="Should not reach AWS"
+        )
+    except Exception as e:
+        logger.info(f"Read-only mode blocked write operation: {e}")
+
+    # Verify the API description was not changed in AWS
+    get_api_aws_after = apigwv2_client_aws.get_api(ApiId=api_id_aws)
+    assert get_api_aws_after.get("Description", "") == original_description
+
+
+def test_apigatewayv2_websocket_api(start_aws_proxy, cleanups):
+    """Test API Gateway v2 WebSocket API operations with proxy."""
+    api_name_aws = f"test-websocket-api-{short_uid()}"
+
+    # start proxy - forwarding all API Gateway v2 requests
+    config = ProxyConfig(services={"apigatewayv2": {"resources": ".*"}})
+    start_aws_proxy(config)
+
+    # create clients
+    region_name = "us-east-1"
+    apigwv2_client = connect_to(region_name=region_name).apigatewayv2
+    apigwv2_client_aws = boto3.client("apigatewayv2", region_name=region_name)
+
+    # create WebSocket API in AWS
+    create_response_aws = apigwv2_client_aws.create_api(
+        Name=api_name_aws,
+        ProtocolType="WEBSOCKET",
+        RouteSelectionExpression="$request.body.action",
+    )
+    api_id_aws = create_response_aws["ApiId"]
+    cleanups.append(lambda: apigwv2_client_aws.delete_api(ApiId=api_id_aws))
+
+    # assert that local call for this API is proxied
+    get_api_local = apigwv2_client.get_api(ApiId=api_id_aws)
+    get_api_aws = apigwv2_client_aws.get_api(ApiId=api_id_aws)
+    assert get_api_local["Name"] == get_api_aws["Name"] == api_name_aws
+    assert get_api_local["ProtocolType"] == "WEBSOCKET"
+
+    # create a route for WebSocket API via AWS client
+    route_response = apigwv2_client_aws.create_route(
+        ApiId=api_id_aws, RouteKey="$connect"
+    )
+    route_id = route_response["RouteId"]
+
+    # verify route via LocalStack (proxied)
+    route_local = apigwv2_client.get_route(ApiId=api_id_aws, RouteId=route_id)
+    assert route_local["RouteKey"] == "$connect"
+    assert route_local["RouteId"] == route_id
+
+    # list routes should include the $connect route
+    routes_local = apigwv2_client.get_routes(ApiId=api_id_aws)
+    route_keys = [r["RouteKey"] for r in routes_local.get("Items", [])]
+    assert "$connect" in route_keys