From 972fd75ce042fe756188c62bef75cc3d9ec38168 Mon Sep 17 00:00:00 2001 From: Waldemar Hummer Date: Sat, 24 Jan 2026 19:04:27 +0000 Subject: [PATCH] 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"]