Skip to content

Add proxy tests for the API Gateway service #119

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
whummer wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Add proxy tests for the API Gateway service #119

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
295 changes: 295 additions & 0 deletions aws-proxy/tests/proxy/test_apigateway.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"]