From 033095953f7b39d28007ae55218ea50fba6032a6 Mon Sep 17 00:00:00 2001 From: Waldemar Hummer Date: Sat, 24 Jan 2026 19:04:27 +0000 Subject: [PATCH 1/2] 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 --- aws-proxy/tests/proxy/test_apigateway.py | 295 +++++++++++++++++++++++ 1 file changed, 295 insertions(+) create mode 100644 aws-proxy/tests/proxy/test_apigateway.py diff --git a/aws-proxy/tests/proxy/test_apigateway.py b/aws-proxy/tests/proxy/test_apigateway.py new file mode 100644 index 0000000..879699f --- /dev/null +++ b/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"] From 111281ce13635e166b1028338acc60be9b3272b0 Mon Sep 17 00:00:00 2001 From: Waldemar Hummer Date: Mon, 2 Feb 2026 21:44:44 +0100 Subject: [PATCH 2/2] add proxy tests for the API Gateway v2 service Co-Authored-By: Claude Opus 4.5 --- aws-proxy/aws_proxy/client/auth_proxy.py | 14 + aws-proxy/tests/proxy/test_apigatewayv2.py | 281 +++++++++++++++++++++ 2 files changed, 295 insertions(+) create mode 100644 aws-proxy/tests/proxy/test_apigatewayv2.py diff --git a/aws-proxy/aws_proxy/client/auth_proxy.py b/aws-proxy/aws_proxy/client/auth_proxy.py index 5426029..459d727 100644 --- a/aws-proxy/aws_proxy/client/auth_proxy.py +++ b/aws-proxy/aws_proxy/client/auth_proxy.py @@ -86,6 +86,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 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( @@ -308,6 +312,16 @@ 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.""" + # 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() diff --git a/aws-proxy/tests/proxy/test_apigatewayv2.py b/aws-proxy/tests/proxy/test_apigatewayv2.py new file mode 100644 index 0000000..3d418b8 --- /dev/null +++ b/aws-proxy/tests/proxy/test_apigatewayv2.py @@ -0,0 +1,281 @@ +# 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