test_apigateway.py

# 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"]