add proxy tests for the API Gateway service

Tests cover:
- REST API requests (create, get, update)
- Resources and methods (get_method, get_method_response, get_integration)
- Deployments and stages
- Read-only mode
- Selective resource matching and error handling

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

Author: whummer

aws-proxy/tests/proxy/test_apigateway.py

@@ -0,0 +1,295 @@
+# 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_apigateway_rest_api_requests(start_aws_proxy, cleanups):
+    api_name_aws = f"test-api-aws-{short_uid()}"
+
+    # start proxy - forwarding all API Gateway requests
+    config = ProxyConfig(services={"apigateway": {"resources": ".*"}})
+    start_aws_proxy(config)
+
+    # create clients
+    region_name = "us-east-1"
+    apigw_client = connect_to(region_name=region_name).apigateway
+    apigw_client_aws = boto3.client("apigateway", region_name=region_name)
+
+    # create REST API in AWS
+    create_response_aws = apigw_client_aws.create_rest_api(
+        name=api_name_aws, description="Test API for proxy testing"
+    )
+    api_id_aws = create_response_aws["id"]
+    cleanups.append(lambda: apigw_client_aws.delete_rest_api(restApiId=api_id_aws))
+
+    # assert that local call for this API is proxied
+    get_api_local = apigw_client.get_rest_api(restApiId=api_id_aws)
+    get_api_aws = apigw_client_aws.get_rest_api(restApiId=api_id_aws)
+    assert get_api_local["name"] == get_api_aws["name"] == api_name_aws
+    assert get_api_local["id"] == get_api_aws["id"] == api_id_aws
+
+    # negative test: verify that requesting a non-existent API fails
+    with pytest.raises(ClientError) as ctx:
+        apigw_client_aws.get_rest_api(restApiId="nonexistent123")
+    assert ctx.value.response["Error"]["Code"] == "NotFoundException"
+
+    # list APIs from AWS should include the created API
+    apis_aws = apigw_client_aws.get_rest_apis()
+    aws_api_ids = [api["id"] 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"
+    apigw_client.update_rest_api(
+        restApiId=api_id_aws,
+        patchOperations=[
+            {"op": "replace", "path": "/description", "value": updated_description}
+        ],
+    )
+
+    # verify update is reflected in AWS
+    get_api_aws = apigw_client_aws.get_rest_api(restApiId=api_id_aws)
+    assert get_api_aws["description"] == updated_description
+
+
+def test_apigateway_resources_and_methods(start_aws_proxy, cleanups):
+    api_name_aws = f"test-api-resources-{short_uid()}"
+
+    # start proxy - forwarding all API Gateway requests
+    config = ProxyConfig(services={"apigateway": {"resources": ".*"}})
+    start_aws_proxy(config)
+
+    # create clients
+    region_name = "us-east-1"
+    apigw_client = connect_to(region_name=region_name).apigateway
+    apigw_client_aws = boto3.client("apigateway", region_name=region_name)
+
+    # create REST API in AWS
+    create_response_aws = apigw_client_aws.create_rest_api(name=api_name_aws)
+    api_id_aws = create_response_aws["id"]
+    cleanups.append(lambda: apigw_client_aws.delete_rest_api(restApiId=api_id_aws))
+
+    # get root resource from AWS
+    resources_response = apigw_client_aws.get_resources(restApiId=api_id_aws)
+    root_resource_id = resources_response["items"][0]["id"]
+
+    # create a new resource via AWS client
+    resource_response = apigw_client_aws.create_resource(
+        restApiId=api_id_aws, parentId=root_resource_id, pathPart="users"
+    )
+    resource_id = resource_response["id"]
+
+    # create a method via AWS client
+    apigw_client_aws.put_method(
+        restApiId=api_id_aws,
+        resourceId=resource_id,
+        httpMethod="GET",
+        authorizationType="NONE",
+    )
+
+    # verify method exists via LocalStack client (should proxy to AWS)
+    method_local = apigw_client.get_method(
+        restApiId=api_id_aws, resourceId=resource_id, httpMethod="GET"
+    )
+    assert method_local["httpMethod"] == "GET"
+    assert method_local["authorizationType"] == "NONE"
+
+    # create method response via AWS client
+    apigw_client_aws.put_method_response(
+        restApiId=api_id_aws,
+        resourceId=resource_id,
+        httpMethod="GET",
+        statusCode="200",
+    )
+
+    # verify method response via LocalStack (proxied)
+    method_response_local = apigw_client.get_method_response(
+        restApiId=api_id_aws, resourceId=resource_id, httpMethod="GET", statusCode="200"
+    )
+    assert method_response_local["statusCode"] == "200"
+
+    # create integration via AWS client
+    apigw_client_aws.put_integration(
+        restApiId=api_id_aws,
+        resourceId=resource_id,
+        httpMethod="GET",
+        type="MOCK",
+        requestTemplates={"application/json": '{"statusCode": 200}'},
+    )
+
+    # verify integration via LocalStack (proxied)
+    integration_local = apigw_client.get_integration(
+        restApiId=api_id_aws, resourceId=resource_id, httpMethod="GET"
+    )
+    assert integration_local["type"] == "MOCK"
+
+    # delete method via AWS client
+    apigw_client_aws.delete_method(
+        restApiId=api_id_aws, resourceId=resource_id, httpMethod="GET"
+    )
+
+    # verify method is deleted via LocalStack (proxied)
+    with pytest.raises(ClientError) as ctx:
+        apigw_client.get_method(
+            restApiId=api_id_aws, resourceId=resource_id, httpMethod="GET"
+        )
+    assert ctx.value.response["Error"]["Code"] in ["NotFoundException", "404"]
+
+
+def test_apigateway_deployments(start_aws_proxy, cleanups):
+    api_name_aws = f"test-api-deploy-{short_uid()}"
+
+    # start proxy - forwarding all API Gateway requests
+    config = ProxyConfig(services={"apigateway": {"resources": ".*"}})
+    start_aws_proxy(config)
+
+    # create clients
+    region_name = "us-east-1"
+    apigw_client = connect_to(region_name=region_name).apigateway
+    apigw_client_aws = boto3.client("apigateway", region_name=region_name)
+
+    # create REST API in AWS
+    create_response_aws = apigw_client_aws.create_rest_api(name=api_name_aws)
+    api_id_aws = create_response_aws["id"]
+    cleanups.append(lambda: apigw_client_aws.delete_rest_api(restApiId=api_id_aws))
+
+    # get root resource and create a simple method
+    resources_response = apigw_client_aws.get_resources(restApiId=api_id_aws)
+    root_resource_id = resources_response["items"][0]["id"]
+
+    apigw_client_aws.put_method(
+        restApiId=api_id_aws,
+        resourceId=root_resource_id,
+        httpMethod="GET",
+        authorizationType="NONE",
+    )
+    apigw_client_aws.put_integration(
+        restApiId=api_id_aws,
+        resourceId=root_resource_id,
+        httpMethod="GET",
+        type="MOCK",
+    )
+
+    # create deployment via LocalStack client (should proxy to AWS)
+    stage_name = "test"
+    deployment_response = apigw_client.create_deployment(
+        restApiId=api_id_aws, stageName=stage_name, description="Test deployment"
+    )
+    deployment_id = deployment_response["id"]
+
+    # verify deployment exists in AWS
+    deployment_aws = apigw_client_aws.get_deployment(
+        restApiId=api_id_aws, deploymentId=deployment_id
+    )
+    assert deployment_aws["id"] == deployment_id
+    assert deployment_aws["description"] == "Test deployment"
+
+    # get stage via LocalStack client
+    stage_local = apigw_client.get_stage(restApiId=api_id_aws, stageName=stage_name)
+    stage_aws = apigw_client_aws.get_stage(restApiId=api_id_aws, stageName=stage_name)
+    assert stage_local["stageName"] == stage_aws["stageName"] == stage_name
+    assert stage_local["deploymentId"] == stage_aws["deploymentId"] == deployment_id
+
+    # list deployments via AWS client (verify deployment exists)
+    deployments_aws = apigw_client_aws.get_deployments(restApiId=api_id_aws)
+    aws_deployment_ids = [d["id"] for d in deployments_aws.get("items", [])]
+    assert deployment_id in aws_deployment_ids
+
+
+def test_apigateway_read_only_mode(start_aws_proxy, cleanups):
+    api_name_aws = f"test-api-readonly-{short_uid()}"
+
+    # create REST API in AWS first (before starting proxy)
+    region_name = "us-east-1"
+    apigw_client_aws = boto3.client("apigateway", region_name=region_name)
+    create_response_aws = apigw_client_aws.create_rest_api(name=api_name_aws)
+    api_id_aws = create_response_aws["id"]
+    cleanups.append(lambda: apigw_client_aws.delete_rest_api(restApiId=api_id_aws))
+
+    # start proxy in read-only mode
+    config = ProxyConfig(
+        services={"apigateway": {"resources": ".*", "read_only": True}}
+    )
+    start_aws_proxy(config)
+
+    # create LocalStack client
+    apigw_client = connect_to(region_name=region_name).apigateway
+
+    # read operations should work (proxied to AWS)
+    get_api_local = apigw_client.get_rest_api(restApiId=api_id_aws)
+    assert get_api_local["name"] == api_name_aws
+    assert get_api_local["id"] == api_id_aws
+
+    # verify the API can also be read directly from AWS
+    get_api_aws = apigw_client_aws.get_rest_api(restApiId=api_id_aws)
+    assert get_api_local["name"] == get_api_aws["name"]
+
+    # write operations should fail (not allowed in read-only mode)
+    # In read-only mode, the API exists in AWS but LocalStack should not
+    # allow write operations to be proxied
+    original_description = get_api_aws.get("description", "")
+
+    # Attempt write operation - should either be blocked or not proxied
+    try:
+        apigw_client.update_rest_api(
+            restApiId=api_id_aws,
+            patchOperations=[
+                {
+                    "op": "replace",
+                    "path": "/description",
+                    "value": "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 = apigw_client_aws.get_rest_api(restApiId=api_id_aws)
+    assert get_api_aws_after.get("description", "") == original_description
+
+
+def test_apigateway_selective_resource_matching(start_aws_proxy, cleanups):
+    api_name_aws = f"test-api-selective-{short_uid()}"
+
+    # start proxy - forwarding all API Gateway requests
+    config = ProxyConfig(services={"apigateway": {"resources": ".*"}})
+    start_aws_proxy(config)
+
+    # create clients
+    region_name = "us-east-1"
+    apigw_client = connect_to(region_name=region_name).apigateway
+    apigw_client_aws = boto3.client("apigateway", region_name=region_name)
+
+    # create API in AWS
+    create_response_aws = apigw_client_aws.create_rest_api(name=api_name_aws)
+    api_id_aws = create_response_aws["id"]
+    cleanups.append(lambda: apigw_client_aws.delete_rest_api(restApiId=api_id_aws))
+
+    # accessing the API via LocalStack should be proxied
+    get_api_local = apigw_client.get_rest_api(restApiId=api_id_aws)
+    assert get_api_local["name"] == api_name_aws
+
+    # verify the API details match between LocalStack and AWS
+    get_api_aws = apigw_client_aws.get_rest_api(restApiId=api_id_aws)
+    assert get_api_local["id"] == get_api_aws["id"]
+    assert get_api_local["name"] == get_api_aws["name"]
+
+    # negative test: verify that requesting a non-existent API ID fails
+    with pytest.raises(ClientError) as ctx:
+        apigw_client_aws.get_rest_api(restApiId="nonexistent123456")
+    assert ctx.value.response["Error"]["Code"] == "NotFoundException"
+
+    # verify LocalStack also returns error for non-existent API
+    with pytest.raises(ClientError) as ctx:
+        apigw_client.get_rest_api(restApiId="nonexistent123456")
+    assert ctx.value.response["Error"]["Code"] in ["NotFoundException", "404"]