add proxy tests for API Gateway v1/v2 (#119)

* 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>

* add proxy tests for the API Gateway v2 service

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

* clean up API Gateway v1/v2 proxy tests: remove logger, use pytest.raises

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

* bump version of aws-proxy extension

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: whummer

aws-proxy/README.md

@@ -126,6 +126,7 @@ If you wish to access the deprecated instructions, they can be found [here](http
 
 ## Change Log
 
+* `0.2.3`: Enhance proxy support and tests for several services (API Gateway v1/v2, CloudWatch, AppSync, Kinesis, KMS, SNS, Cognito-IDP)
 * `0.2.2`: Refactor UI to use WebAppExtension pattern
 * `0.2.1`: Restructure project to use pyproject.toml
 * `0.2.0`: Rename extension from `localstack-extension-aws-replicator` to `localstack-extension-aws-proxy`

aws-proxy/aws_proxy/client/auth_proxy.py

@@ -90,8 +90,10 @@ def proxy_request(self, request: Request, data: bytes) -> Response:
         if not parsed:
             return requests_response("", status_code=400)
         region_name, service_name = parsed
-        # Map AWS signing names to boto3 client names
-        service_name = SERVICE_NAME_MAPPING.get(service_name, service_name)
+
+        # Map service names based on request context
+        service_name = self._get_service_name(service_name, request.path)
+
         query_string = to_str(request.query_string or "")
 
         LOG.debug(
@@ -349,6 +351,18 @@ def _extract_region_and_service(self, headers) -> Optional[Tuple[str, str]]:
             return
         return parts[2], parts[3]
 
+    def _get_service_name(self, service_name: str, path: str) -> str:
+        """Map AWS signing service names to boto3 client names based on request context."""
+        # Map AWS signing names to boto3 client names
+        service_name = SERVICE_NAME_MAPPING.get(service_name, service_name)
+        # API Gateway v2 uses 'apigateway' as signing name but needs 'apigatewayv2' client
+        if service_name == "apigateway" and path.startswith("/v2/"):
+            return "apigatewayv2"
+        # CloudWatch uses 'monitoring' as signing name
+        if service_name == "monitoring":
+            return "cloudwatch"
+        return service_name
+
     @cache
     def _query_account_id_from_aws(self) -> str:
         session = boto3.Session()

aws-proxy/aws_proxy/server/aws_request_forwarder.py

@@ -283,6 +283,10 @@ def _is_read_request(self, context: RequestContext) -> bool:
             "BatchGetServiceLevelIndicatorReport",
         }:
             return True
+        if context.service.service_name == "apigatewayv2" and operation_name in {
+            "ExportApi",
+        }:
+            return True
         # TODO: add more rules
         return False
 

aws-proxy/pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "localstack-extension-aws-proxy"
 readme = "README.md"
-version = "0.2.2"
+version = "0.2.3"
 description = "LocalStack extension that proxies AWS resources into your LocalStack instance"
 authors = [
         { name = "LocalStack Team", email = "info@localstack.cloud"}

aws-proxy/tests/proxy/test_apigateway.py

@@ -0,0 +1,289 @@
+# Note: This file has been (partially or fully) generated by an AI agent.
+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
+
+
+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 be blocked in read-only mode
+    with pytest.raises(Exception):
+        apigw_client.update_rest_api(
+            restApiId=api_id_aws,
+            patchOperations=[
+                {
+                    "op": "replace",
+                    "path": "/description",
+                    "value": "Should not reach AWS",
+                }
+            ],
+        )
+
+    # 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"]