#2521 - Implement API Gateway Stage deletion

This commit is contained in:
Bert Blommers 2019-11-04 09:12:24 +00:00
parent 67f9dd12da
commit 2d32ee18a6
4 changed files with 651 additions and 66 deletions

View File

@ -2,6 +2,89 @@ from __future__ import unicode_literals
from moto.core.exceptions import RESTError
class BadRequestException(RESTError):
pass
class AwsProxyNotAllowed(BadRequestException):
def __init__(self):
super(AwsProxyNotAllowed, self).__init__(
"BadRequestException",
"Integrations of type 'AWS_PROXY' currently only supports Lambda function and Firehose stream invocations.",
)
class CrossAccountNotAllowed(RESTError):
def __init__(self):
super(CrossAccountNotAllowed, self).__init__(
"AccessDeniedException", "Cross-account pass role is not allowed."
)
class RoleNotSpecified(BadRequestException):
def __init__(self):
super(RoleNotSpecified, self).__init__(
"BadRequestException", "Role ARN must be specified for AWS integrations"
)
class IntegrationMethodNotDefined(BadRequestException):
def __init__(self):
super(IntegrationMethodNotDefined, self).__init__(
"BadRequestException", "Enumeration value for HttpMethod must be non-empty"
)
class InvalidResourcePathException(BadRequestException):
def __init__(self):
super(InvalidResourcePathException, self).__init__(
"BadRequestException",
"Resource's path part only allow a-zA-Z0-9._- and curly braces at the beginning and the end.",
)
class InvalidHttpEndpoint(BadRequestException):
def __init__(self):
super(InvalidHttpEndpoint, self).__init__(
"BadRequestException", "Invalid HTTP endpoint specified for URI"
)
class InvalidArn(BadRequestException):
def __init__(self):
super(InvalidArn, self).__init__(
"BadRequestException", "Invalid ARN specified in the request"
)
class InvalidIntegrationArn(BadRequestException):
def __init__(self):
super(InvalidIntegrationArn, self).__init__(
"BadRequestException", "AWS ARN for integration must contain path or action"
)
class InvalidRequestInput(BadRequestException):
def __init__(self):
super(InvalidRequestInput, self).__init__(
"BadRequestException", "Invalid request input"
)
class NoIntegrationDefined(BadRequestException):
def __init__(self):
super(NoIntegrationDefined, self).__init__(
"BadRequestException", "No integration defined for method"
)
class NoMethodDefined(BadRequestException):
def __init__(self):
super(NoMethodDefined, self).__init__(
"BadRequestException", "The REST API doesn't contain any methods"
)
class StageNotFoundException(RESTError):
code = 404

View File

@ -3,15 +3,32 @@ from __future__ import unicode_literals
import random
import string
import re
import requests
import time
from boto3.session import Session
from urlparse import urlparse
import responses
from moto.core import BaseBackend, BaseModel
from .utils import create_id
from moto.core.utils import path_url
from .exceptions import StageNotFoundException, ApiKeyNotFoundException
from moto.sts.models import ACCOUNT_ID
from .exceptions import (
ApiKeyNotFoundException,
AwsProxyNotAllowed,
CrossAccountNotAllowed,
IntegrationMethodNotDefined,
InvalidArn,
InvalidIntegrationArn,
InvalidHttpEndpoint,
InvalidResourcePathException,
InvalidRequestInput,
StageNotFoundException,
RoleNotSpecified,
NoIntegrationDefined,
NoMethodDefined,
)
STAGE_URL = "https://{api_id}.execute-api.{region_name}.amazonaws.com/{stage_name}"
@ -534,6 +551,8 @@ class APIGatewayBackend(BaseBackend):
return resource
def create_resource(self, function_id, parent_resource_id, path_part):
if not re.match("^\\{?[a-zA-Z0-9._-]+\\}?$", path_part):
raise InvalidResourcePathException()
api = self.get_rest_api(function_id)
child = api.add_child(path=path_part, parent_id=parent_resource_id)
return child
@ -594,6 +613,10 @@ class APIGatewayBackend(BaseBackend):
stage = api.stages[stage_name] = Stage()
return stage.apply_operations(patch_operations)
def delete_stage(self, function_id, stage_name):
api = self.get_rest_api(function_id)
del api.stages[stage_name]
def get_method_response(self, function_id, resource_id, method_type, response_code):
method = self.get_method(function_id, resource_id, method_type)
method_response = method.get_response(response_code)
@ -620,9 +643,40 @@ class APIGatewayBackend(BaseBackend):
method_type,
integration_type,
uri,
integration_method=None,
credentials=None,
request_templates=None,
):
resource = self.get_resource(function_id, resource_id)
if credentials and not re.match(
"^arn:aws:iam::" + str(ACCOUNT_ID), credentials
):
raise CrossAccountNotAllowed()
if not integration_method and integration_type in [
"HTTP",
"HTTP_PROXY",
"AWS",
"AWS_PROXY",
]:
raise IntegrationMethodNotDefined()
if integration_type in ["AWS_PROXY"] and re.match(
"^arn:aws:apigateway:[a-zA-Z0-9-]+:s3", uri
):
raise AwsProxyNotAllowed()
if (
integration_type in ["AWS"]
and re.match("^arn:aws:apigateway:[a-zA-Z0-9-]+:s3", uri)
and not credentials
):
raise RoleNotSpecified()
if integration_type in ["HTTP", "HTTP_PROXY"] and not self._uri_validator(uri):
raise InvalidHttpEndpoint()
if integration_type in ["AWS", "AWS_PROXY"] and not re.match("^arn:aws:", uri):
raise InvalidArn()
if integration_type in ["AWS", "AWS_PROXY"] and not re.match(
"^arn:aws:apigateway:[a-zA-Z0-9-]+:[a-zA-Z0-9-]+:(path|action)/", uri
):
raise InvalidIntegrationArn()
integration = resource.add_integration(
method_type, integration_type, uri, request_templates=request_templates
)
@ -637,8 +691,16 @@ class APIGatewayBackend(BaseBackend):
return resource.delete_integration(method_type)
def create_integration_response(
self, function_id, resource_id, method_type, status_code, selection_pattern
self,
function_id,
resource_id,
method_type,
status_code,
selection_pattern,
response_templates,
):
if response_templates is None:
raise InvalidRequestInput()
integration = self.get_integration(function_id, resource_id, method_type)
integration_response = integration.create_integration_response(
status_code, selection_pattern
@ -665,6 +727,18 @@ class APIGatewayBackend(BaseBackend):
if stage_variables is None:
stage_variables = {}
api = self.get_rest_api(function_id)
methods = [
res.resource_methods.values()[0] for res in self.list_resources(function_id)
]
if not any(methods):
raise NoMethodDefined()
method_integrations = [
method["methodIntegration"]
for method in methods
if "methodIntegration" in method
]
if not all(method_integrations):
raise NoIntegrationDefined()
deployment = api.create_deployment(name, description, stage_variables)
return deployment
@ -753,6 +827,13 @@ class APIGatewayBackend(BaseBackend):
self.usage_plan_keys[usage_plan_id].pop(key_id)
return {}
def _uri_validator(self, uri):
try:
result = urlparse(uri)
return all([result.scheme, result.netloc, result.path])
except Exception:
return False
apigateway_backends = {}
for region_name in Session().get_available_regions("apigateway"):

View File

@ -4,12 +4,24 @@ import json
from moto.core.responses import BaseResponse
from .models import apigateway_backends
from .exceptions import StageNotFoundException, ApiKeyNotFoundException
from .exceptions import (
ApiKeyNotFoundException,
BadRequestException,
CrossAccountNotAllowed,
StageNotFoundException,
)
class APIGatewayResponse(BaseResponse):
def error(self, type_, message, status=400):
return (
status,
self.response_headers,
json.dumps({"__type": type_, "message": message}),
)
def _get_param(self, key):
return json.loads(self.body).get(key)
return json.loads(self.body).get(key) if self.body else None
def _get_param_with_default_value(self, key, default):
jsonbody = json.loads(self.body)
@ -63,14 +75,21 @@ class APIGatewayResponse(BaseResponse):
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
resource_id = self.path.split("/")[-1]
if self.method == "GET":
resource = self.backend.get_resource(function_id, resource_id)
elif self.method == "POST":
path_part = self._get_param("pathPart")
resource = self.backend.create_resource(function_id, resource_id, path_part)
elif self.method == "DELETE":
resource = self.backend.delete_resource(function_id, resource_id)
return 200, {}, json.dumps(resource.to_dict())
try:
if self.method == "GET":
resource = self.backend.get_resource(function_id, resource_id)
elif self.method == "POST":
path_part = self._get_param("pathPart")
resource = self.backend.create_resource(
function_id, resource_id, path_part
)
elif self.method == "DELETE":
resource = self.backend.delete_resource(function_id, resource_id)
return 200, {}, json.dumps(resource.to_dict())
except BadRequestException as e:
return self.error(
"com.amazonaws.dynamodb.v20111205#BadRequestException", e.message
)
def resource_methods(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
@ -165,6 +184,9 @@ class APIGatewayResponse(BaseResponse):
stage_response = self.backend.update_stage(
function_id, stage_name, patch_operations
)
elif self.method == "DELETE":
self.backend.delete_stage(function_id, stage_name)
return 202, {}, "{}"
return 200, {}, json.dumps(stage_response)
def integrations(self, request, full_url, headers):
@ -174,27 +196,40 @@ class APIGatewayResponse(BaseResponse):
resource_id = url_path_parts[4]
method_type = url_path_parts[6]
if self.method == "GET":
integration_response = self.backend.get_integration(
function_id, resource_id, method_type
try:
if self.method == "GET":
integration_response = self.backend.get_integration(
function_id, resource_id, method_type
)
elif self.method == "PUT":
integration_type = self._get_param("type")
uri = self._get_param("uri")
integration_http_method = self._get_param("httpMethod")
creds = self._get_param("credentials")
request_templates = self._get_param("requestTemplates")
integration_response = self.backend.create_integration(
function_id,
resource_id,
method_type,
integration_type,
uri,
credentials=creds,
integration_method=integration_http_method,
request_templates=request_templates,
)
elif self.method == "DELETE":
integration_response = self.backend.delete_integration(
function_id, resource_id, method_type
)
return 200, {}, json.dumps(integration_response)
except BadRequestException as e:
return self.error(
"com.amazonaws.dynamodb.v20111205#BadRequestException", e.message
)
elif self.method == "PUT":
integration_type = self._get_param("type")
uri = self._get_param("uri")
request_templates = self._get_param("requestTemplates")
integration_response = self.backend.create_integration(
function_id,
resource_id,
method_type,
integration_type,
uri,
request_templates=request_templates,
except CrossAccountNotAllowed as e:
return self.error(
"com.amazonaws.dynamodb.v20111205#AccessDeniedException", e.message
)
elif self.method == "DELETE":
integration_response = self.backend.delete_integration(
function_id, resource_id, method_type
)
return 200, {}, json.dumps(integration_response)
def integration_responses(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
@ -204,36 +239,52 @@ class APIGatewayResponse(BaseResponse):
method_type = url_path_parts[6]
status_code = url_path_parts[9]
if self.method == "GET":
integration_response = self.backend.get_integration_response(
function_id, resource_id, method_type, status_code
try:
if self.method == "GET":
integration_response = self.backend.get_integration_response(
function_id, resource_id, method_type, status_code
)
elif self.method == "PUT":
selection_pattern = self._get_param("selectionPattern")
response_templates = self._get_param("responseTemplates")
integration_response = self.backend.create_integration_response(
function_id,
resource_id,
method_type,
status_code,
selection_pattern,
response_templates,
)
elif self.method == "DELETE":
integration_response = self.backend.delete_integration_response(
function_id, resource_id, method_type, status_code
)
return 200, {}, json.dumps(integration_response)
except BadRequestException as e:
return self.error(
"com.amazonaws.dynamodb.v20111205#BadRequestException", e.message
)
elif self.method == "PUT":
selection_pattern = self._get_param("selectionPattern")
integration_response = self.backend.create_integration_response(
function_id, resource_id, method_type, status_code, selection_pattern
)
elif self.method == "DELETE":
integration_response = self.backend.delete_integration_response(
function_id, resource_id, method_type, status_code
)
return 200, {}, json.dumps(integration_response)
def deployments(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
if self.method == "GET":
deployments = self.backend.get_deployments(function_id)
return 200, {}, json.dumps({"item": deployments})
elif self.method == "POST":
name = self._get_param("stageName")
description = self._get_param_with_default_value("description", "")
stage_variables = self._get_param_with_default_value("variables", {})
deployment = self.backend.create_deployment(
function_id, name, description, stage_variables
try:
if self.method == "GET":
deployments = self.backend.get_deployments(function_id)
return 200, {}, json.dumps({"item": deployments})
elif self.method == "POST":
name = self._get_param("stageName")
description = self._get_param_with_default_value("description", "")
stage_variables = self._get_param_with_default_value("variables", {})
deployment = self.backend.create_deployment(
function_id, name, description, stage_variables
)
return 200, {}, json.dumps(deployment)
except BadRequestException as e:
return self.error(
"com.amazonaws.dynamodb.v20111205#BadRequestException", e.message
)
return 200, {}, json.dumps(deployment)
def individual_deployment(self, request, full_url, headers):
self.setup_class(request, full_url, headers)

View File

@ -9,6 +9,7 @@ from botocore.exceptions import ClientError
import responses
from moto import mock_apigateway, settings
from nose.tools import assert_raises
@freeze_time("2015-01-01")
@ -45,6 +46,32 @@ def test_list_and_delete_apis():
len(response["items"]).should.equal(1)
@mock_apigateway
def test_create_resource__validate_name():
client = boto3.client("apigateway", region_name="us-west-2")
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources["items"] if resource["path"] == "/"][
0
]["id"]
invalid_names = ["/users", "users/", "users/{user_id}", "us{er"]
valid_names = ["users", "{user_id}", "user_09", "good-dog"]
# All invalid names should throw an exception
for name in invalid_names:
with assert_raises(ClientError) as ex:
client.create_resource(restApiId=api_id, parentId=root_id, pathPart=name)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"Resource's path part only allow a-zA-Z0-9._- and curly braces at the beginning and the end."
)
# All valid names should go through
for name in valid_names:
client.create_resource(restApiId=api_id, parentId=root_id, pathPart=name)
@mock_apigateway
def test_create_resource():
client = boto3.client("apigateway", region_name="us-west-2")
@ -69,9 +96,7 @@ def test_create_resource():
}
)
response = client.create_resource(
restApiId=api_id, parentId=root_id, pathPart="/users"
)
client.create_resource(restApiId=api_id, parentId=root_id, pathPart="users")
resources = client.get_resources(restApiId=api_id)["items"]
len(resources).should.equal(2)
@ -79,9 +104,7 @@ def test_create_resource():
0
]
response = client.delete_resource(
restApiId=api_id, resourceId=non_root_resource["id"]
)
client.delete_resource(restApiId=api_id, resourceId=non_root_resource["id"])
len(client.get_resources(restApiId=api_id)["items"]).should.equal(1)
@ -223,6 +246,7 @@ def test_integrations():
httpMethod="GET",
type="HTTP",
uri="http://httpbin.org/robots.txt",
integrationHttpMethod="POST",
)
# this is hard to match against, so remove it
response["ResponseMetadata"].pop("HTTPHeaders", None)
@ -308,6 +332,7 @@ def test_integrations():
type="HTTP",
uri=test_uri,
requestTemplates=templates,
integrationHttpMethod="POST",
)
# this is hard to match against, so remove it
response["ResponseMetadata"].pop("HTTPHeaders", None)
@ -340,12 +365,13 @@ def test_integration_response():
restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"
)
response = client.put_integration(
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type="HTTP",
uri="http://httpbin.org/robots.txt",
integrationHttpMethod="POST",
)
response = client.put_integration_response(
@ -354,6 +380,7 @@ def test_integration_response():
httpMethod="GET",
statusCode="200",
selectionPattern="foobar",
responseTemplates={},
)
# this is hard to match against, so remove it
response["ResponseMetadata"].pop("HTTPHeaders", None)
@ -410,6 +437,7 @@ def test_update_stage_configuration():
stage_name = "staging"
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
create_method_integration(client, api_id)
response = client.create_deployment(
restApiId=api_id, stageName=stage_name, description="1.0.1"
@ -534,7 +562,8 @@ def test_create_stage():
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
response = client.create_deployment(restApiId=api_id, stageName=stage_name)
create_method_integration(client, api_id)
response = client.create_deployment(restApiId=api_id, stageName=stage_name,)
deployment_id = response["id"]
response = client.get_deployment(restApiId=api_id, deploymentId=deployment_id)
@ -690,12 +719,323 @@ def test_create_stage():
stage["cacheClusterSize"].should.equal("1.6")
@mock_apigateway
def test_create_deployment_requires_REST_methods():
client = boto3.client("apigateway", region_name="us-west-2")
stage_name = "staging"
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
with assert_raises(ClientError) as ex:
client.create_deployment(restApiId=api_id, stageName=stage_name)["id"]
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"The REST API doesn't contain any methods"
)
@mock_apigateway
def test_create_deployment_requires_REST_method_integrations():
client = boto3.client("apigateway", region_name="us-west-2")
stage_name = "staging"
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources["items"] if resource["path"] == "/"][
0
]["id"]
client.put_method(
restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="NONE"
)
with assert_raises(ClientError) as ex:
client.create_deployment(restApiId=api_id, stageName=stage_name)["id"]
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"No integration defined for method"
)
@mock_apigateway
def test_create_simple_deployment_with_get_method():
client = boto3.client("apigateway", region_name="us-west-2")
stage_name = "staging"
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
create_method_integration(client, api_id)
deployment = client.create_deployment(restApiId=api_id, stageName=stage_name)
assert "id" in deployment
@mock_apigateway
def test_create_simple_deployment_with_post_method():
client = boto3.client("apigateway", region_name="us-west-2")
stage_name = "staging"
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
create_method_integration(client, api_id, httpMethod="POST")
deployment = client.create_deployment(restApiId=api_id, stageName=stage_name)
assert "id" in deployment
@mock_apigateway
# https://github.com/aws/aws-sdk-js/issues/2588
def test_put_integration_response_requires_responseTemplate():
client = boto3.client("apigateway", region_name="us-west-2")
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources["items"] if resource["path"] == "/"][
0
]["id"]
client.put_method(
restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="NONE"
)
client.put_method_response(
restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"
)
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type="HTTP",
uri="http://httpbin.org/robots.txt",
integrationHttpMethod="POST",
)
with assert_raises(ClientError) as ex:
client.put_integration_response(
restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"
)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal("Invalid request input")
# Works fine if responseTemplate is defined
client.put_integration_response(
restApiId=api_id,
resourceId=resource["id"],
httpMethod="GET",
statusCode="200",
responseTemplates={},
)
@mock_apigateway
def test_put_integration_validation():
client = boto3.client("apigateway", region_name="us-west-2")
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources["items"] if resource["path"] == "/"][
0
]["id"]
client.put_method(
restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="NONE"
)
client.put_method_response(
restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"
)
http_types = ["HTTP", "HTTP_PROXY"]
aws_types = ["AWS", "AWS_PROXY"]
types_requiring_integration_method = http_types + aws_types
types_not_requiring_integration_method = ["MOCK"]
for type in types_requiring_integration_method:
# Ensure that integrations of these types fail if no integrationHttpMethod is provided
with assert_raises(ClientError) as ex:
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type=type,
uri="http://httpbin.org/robots.txt",
)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"Enumeration value for HttpMethod must be non-empty"
)
for type in types_not_requiring_integration_method:
# Ensure that integrations of these types do not need the integrationHttpMethod
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type=type,
uri="http://httpbin.org/robots.txt",
)
for type in http_types:
# Ensure that it works fine when providing the integrationHttpMethod-argument
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type=type,
uri="http://httpbin.org/robots.txt",
integrationHttpMethod="POST",
)
for type in ["AWS"]:
# Ensure that it works fine when providing the integrationHttpMethod + credentials
client.put_integration(
restApiId=api_id,
resourceId=root_id,
credentials="arn:aws:iam::123456789012:role/service-role/testfunction-role-oe783psq",
httpMethod="GET",
type=type,
uri="arn:aws:apigateway:us-west-2:s3:path/b/k",
integrationHttpMethod="POST",
)
for type in aws_types:
# Ensure that credentials are not required when URI points to a Lambda stream
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type=type,
uri="arn:aws:apigateway:eu-west-1:lambda:path/2015-03-31/functions/arn:aws:lambda:eu-west-1:012345678901:function:MyLambda/invocations",
integrationHttpMethod="POST",
)
for type in ["AWS_PROXY"]:
# Ensure that aws_proxy does not support S3
with assert_raises(ClientError) as ex:
client.put_integration(
restApiId=api_id,
resourceId=root_id,
credentials="arn:aws:iam::123456789012:role/service-role/testfunction-role-oe783psq",
httpMethod="GET",
type=type,
uri="arn:aws:apigateway:us-west-2:s3:path/b/k",
integrationHttpMethod="POST",
)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"Integrations of type 'AWS_PROXY' currently only supports Lambda function and Firehose stream invocations."
)
for type in aws_types:
# Ensure that the Role ARN is for the current account
with assert_raises(ClientError) as ex:
client.put_integration(
restApiId=api_id,
resourceId=root_id,
credentials="arn:aws:iam::000000000000:role/service-role/testrole",
httpMethod="GET",
type=type,
uri="arn:aws:apigateway:us-west-2:s3:path/b/k",
integrationHttpMethod="POST",
)
ex.exception.response["Error"]["Code"].should.equal("AccessDeniedException")
ex.exception.response["Error"]["Message"].should.equal(
"Cross-account pass role is not allowed."
)
for type in ["AWS"]:
# Ensure that the Role ARN is specified for aws integrations
with assert_raises(ClientError) as ex:
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type=type,
uri="arn:aws:apigateway:us-west-2:s3:path/b/k",
integrationHttpMethod="POST",
)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"Role ARN must be specified for AWS integrations"
)
for type in http_types:
# Ensure that the URI is valid HTTP
with assert_raises(ClientError) as ex:
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type=type,
uri="non-valid-http",
integrationHttpMethod="POST",
)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"Invalid HTTP endpoint specified for URI"
)
for type in aws_types:
# Ensure that the URI is an ARN
with assert_raises(ClientError) as ex:
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type=type,
uri="non-valid-arn",
integrationHttpMethod="POST",
)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"Invalid ARN specified in the request"
)
for type in aws_types:
# Ensure that the URI is a valid ARN
with assert_raises(ClientError) as ex:
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type=type,
uri="arn:aws:iam::0000000000:role/service-role/asdf",
integrationHttpMethod="POST",
)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal(
"AWS ARN for integration must contain path or action"
)
@mock_apigateway
def test_delete_stage():
client = boto3.client("apigateway", region_name="us-west-2")
stage_name = "staging"
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
create_method_integration(client, api_id)
deployment_id1 = client.create_deployment(restApiId=api_id, stageName=stage_name)[
"id"
]
deployment_id2 = client.create_deployment(restApiId=api_id, stageName=stage_name)[
"id"
]
new_stage_name = "current"
client.create_stage(
restApiId=api_id, stageName=new_stage_name, deploymentId=deployment_id1
)
new_stage_name_with_vars = "stage_with_vars"
client.create_stage(
restApiId=api_id,
stageName=new_stage_name_with_vars,
deploymentId=deployment_id2,
variables={"env": "dev"},
)
stages = client.get_stages(restApiId=api_id)["item"]
[stage["stageName"] for stage in stages].should.equal(
[new_stage_name, new_stage_name_with_vars, stage_name]
)
# delete stage
response = client.delete_stage(restApiId=api_id, stageName=new_stage_name_with_vars)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(202)
# verify other stage still exists
stages = client.get_stages(restApiId=api_id)["item"]
[stage["stageName"] for stage in stages].should.equal([new_stage_name, stage_name])
@mock_apigateway
def test_deployment():
client = boto3.client("apigateway", region_name="us-west-2")
stage_name = "staging"
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
create_method_integration(client, api_id)
response = client.create_deployment(restApiId=api_id, stageName=stage_name)
deployment_id = response["id"]
@ -719,7 +1059,7 @@ def test_deployment():
response["items"][0].pop("createdDate")
response["items"].should.equal([{"id": deployment_id, "description": ""}])
response = client.delete_deployment(restApiId=api_id, deploymentId=deployment_id)
client.delete_deployment(restApiId=api_id, deploymentId=deployment_id)
response = client.get_deployments(restApiId=api_id)
len(response["items"]).should.equal(0)
@ -730,7 +1070,7 @@ def test_deployment():
stage["stageName"].should.equal(stage_name)
stage["deploymentId"].should.equal(deployment_id)
stage = client.update_stage(
client.update_stage(
restApiId=api_id,
stageName=stage_name,
patchOperations=[
@ -774,6 +1114,7 @@ def test_http_proxying_integration():
httpMethod="GET",
type="HTTP",
uri="http://httpbin.org/robots.txt",
integrationHttpMethod="POST",
)
stage_name = "staging"
@ -888,7 +1229,6 @@ def test_usage_plans():
@mock_apigateway
def test_usage_plan_keys():
region_name = "us-west-2"
usage_plan_id = "test_usage_plan_id"
client = boto3.client("apigateway", region_name=region_name)
usage_plan_id = "test"
@ -932,7 +1272,6 @@ def test_usage_plan_keys():
@mock_apigateway
def test_create_usage_plan_key_non_existent_api_key():
region_name = "us-west-2"
usage_plan_id = "test_usage_plan_id"
client = boto3.client("apigateway", region_name=region_name)
usage_plan_id = "test"
@ -976,3 +1315,34 @@ def test_get_usage_plans_using_key_id():
len(only_plans_with_key["items"]).should.equal(1)
only_plans_with_key["items"][0]["name"].should.equal(attached_plan["name"])
only_plans_with_key["items"][0]["id"].should.equal(attached_plan["id"])
def create_method_integration(client, api_id, httpMethod="GET"):
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources["items"] if resource["path"] == "/"][
0
]["id"]
client.put_method(
restApiId=api_id,
resourceId=root_id,
httpMethod=httpMethod,
authorizationType="NONE",
)
client.put_method_response(
restApiId=api_id, resourceId=root_id, httpMethod=httpMethod, statusCode="200"
)
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod=httpMethod,
type="HTTP",
uri="http://httpbin.org/robots.txt",
integrationHttpMethod="POST",
)
client.put_integration_response(
restApiId=api_id,
resourceId=root_id,
httpMethod=httpMethod,
statusCode="200",
responseTemplates={},
)