Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Techdebt: Replace sure with regular asserts in APIGatewayV2 tests (#6377)

Browse Source
This commit is contained in:
Bert Blommers 2023-06-08 11:40:10 +00:00 committed by GitHub
parent 681873d177
commit 12c82c3088
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 471 additions and 492 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

219
tests/test_apigatewayv2/test_apigatewayv2.py
View File

@ -1,7 +1,6 @@
"""Unit tests for apigatewayv2-supported APIs."""
import boto3
import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@ -17,9 +16,9 @@ def test_create_api_with_unknown_protocol_type():
with pytest.raises(ClientError) as exc:
client.create_api(Name="test-api", ProtocolType="?")
err = exc.value.response["Error"]
err["Code"].should.equal("BadRequestException")
err["Message"].should.equal(
"Invalid protocol specified. Must be one of [HTTP, WEBSOCKET]"
assert err["Code"] == "BadRequestException"
assert (
err["Message"] == "Invalid protocol specified. Must be one of [HTTP, WEBSOCKET]"
)
@ -28,20 +27,17 @@ def test_create_api_minimal():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
resp = client.create_api(Name="test-api", ProtocolType="HTTP")
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals(
"$request.header.x-api-key"
)
resp.should.have.key("CreatedDate")
resp.should.have.key("DisableExecuteApiEndpoint").equals(False)
resp.should.have.key("Name").equals("test-api")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals(
"$request.method $request.path"
assert "ApiId" in resp
assert (
resp["ApiEndpoint"]
== f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
assert resp["ApiKeySelectionExpression"] == "$request.header.x-api-key"
assert "CreatedDate" in resp
assert resp["DisableExecuteApiEndpoint"] is False
assert resp["Name"] == "test-api"
assert resp["ProtocolType"] == "HTTP"
assert resp["RouteSelectionExpression"] == "$request.method $request.path"
@mock_apigatewayv2
@ -66,29 +62,28 @@ def test_create_api():
Version="1.0",
)
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
assert "ApiId" in resp
assert (
resp["ApiEndpoint"]
== f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals("s3l3ction")
resp.should.have.key("CreatedDate")
resp.should.have.key("CorsConfiguration").equals(
{
"AllowCredentials": True,
"AllowHeaders": ["x-header1"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header1"],
"MaxAge": 2,
}
)
resp.should.have.key("Description").equals("my first api")
resp.should.have.key("DisableExecuteApiEndpoint").equals(True)
resp.should.have.key("DisableSchemaValidation").equals(True)
resp.should.have.key("Name").equals("test-api")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals("route_s3l3ction")
resp.should.have.key("Version").equals("1.0")
assert resp["ApiKeySelectionExpression"] == "s3l3ction"
assert "CreatedDate" in resp
assert resp["CorsConfiguration"] == {
"AllowCredentials": True,
"AllowHeaders": ["x-header1"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header1"],
"MaxAge": 2,
}
assert resp["Description"] == "my first api"
assert resp["DisableExecuteApiEndpoint"] is True
assert resp["DisableSchemaValidation"] is True
assert resp["Name"] == "test-api"
assert resp["ProtocolType"] == "HTTP"
assert resp["RouteSelectionExpression"] == "route_s3l3ction"
assert resp["Version"] == "1.0"
@mock_apigatewayv2
@ -100,7 +95,7 @@ def test_delete_api():
with pytest.raises(ClientError) as exc:
client.get_api(ApiId=api_id)
exc.value.response["Error"]["Code"].should.equal("NotFoundException")
assert exc.value.response["Error"]["Code"] == "NotFoundException"
@mock_apigatewayv2
@ -129,9 +124,9 @@ def test_delete_cors_configuration():
resp = client.get_api(ApiId=api_id)
resp.shouldnt.have.key("CorsConfiguration")
resp.should.have.key("Description").equals("my first api")
resp.should.have.key("Name").equals("test-api")
assert "CorsConfiguration" not in resp
assert resp["Description"] == "my first api"
assert resp["Name"] == "test-api"
@mock_apigatewayv2
@ -141,8 +136,8 @@ def test_get_api_unknown():
client.get_api(ApiId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid API identifier specified unknown")
assert err["Code"] == "NotFoundException"
assert err["Message"] == "Invalid API identifier specified unknown"
@mock_apigatewayv2
@ -152,34 +147,31 @@ def test_get_api():
resp = client.get_api(ApiId=api_id)
resp.should.have.key("ApiId").equals(api_id)
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.ap-southeast-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals(
"$request.header.x-api-key"
)
resp.should.have.key("CreatedDate")
resp.should.have.key("DisableExecuteApiEndpoint").equals(False)
resp.should.have.key("Name").equals("test-get-api")
resp.should.have.key("ProtocolType").equals("WEBSOCKET")
resp.should.have.key("RouteSelectionExpression").equals(
"$request.method $request.path"
assert resp["ApiId"] == api_id
assert (
resp["ApiEndpoint"]
== f"https://{resp['ApiId']}.execute-api.ap-southeast-1.amazonaws.com"
)
assert resp["ApiKeySelectionExpression"] == "$request.header.x-api-key"
assert "CreatedDate" in resp
assert resp["DisableExecuteApiEndpoint"] is False
assert resp["Name"] == "test-get-api"
assert resp["ProtocolType"] == "WEBSOCKET"
assert resp["RouteSelectionExpression"] == "$request.method $request.path"
@mock_apigatewayv2
def test_get_apis():
client = boto3.client("apigatewayv2", region_name="ap-southeast-1")
client.get_apis().should.have.key("Items").length_of(0)
assert len(client.get_apis()["Items"]) == 0
api_id_1 = client.create_api(Name="api1", ProtocolType="HTTP")["ApiId"]
api_id_2 = client.create_api(Name="api2", ProtocolType="WEBSOCKET")["ApiId"]
client.get_apis().should.have.key("Items").length_of(2)
assert len(client.get_apis()["Items"]) == 2
api_ids = [i["ApiId"] for i in client.get_apis()["Items"]]
api_ids.should.contain(api_id_1)
api_ids.should.contain(api_id_2)
assert api_id_1 in api_ids
assert api_id_2 in api_ids
@mock_apigatewayv2
@ -216,29 +208,28 @@ def test_update_api_minimal():
},
)
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
assert "ApiId" in resp
assert (
resp["ApiEndpoint"]
== f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals("s3l3ction")
resp.should.have.key("CreatedDate")
resp.should.have.key("CorsConfiguration").equals(
{
"AllowCredentials": False,
"AllowHeaders": ["x-header2"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header2"],
"MaxAge": 2,
}
)
resp.should.have.key("Description").equals("my first api")
resp.should.have.key("DisableExecuteApiEndpoint").equals(True)
resp.should.have.key("DisableSchemaValidation").equals(True)
resp.should.have.key("Name").equals("test-api")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals("route_s3l3ction")
resp.should.have.key("Version").equals("1.0")
assert resp["ApiKeySelectionExpression"] == "s3l3ction"
assert "CreatedDate" in resp
assert resp["CorsConfiguration"] == {
"AllowCredentials": False,
"AllowHeaders": ["x-header2"],
"AllowMethods": ["GET", "PUT"],
"AllowOrigins": ["google.com"],
"ExposeHeaders": ["x-header2"],
"MaxAge": 2,
}
assert resp["Description"] == "my first api"
assert resp["DisableExecuteApiEndpoint"] is True
assert resp["DisableSchemaValidation"] is True
assert resp["Name"] == "test-api"
assert resp["ProtocolType"] == "HTTP"
assert resp["RouteSelectionExpression"] == "route_s3l3ction"
assert resp["Version"] == "1.0"
@mock_apigatewayv2
@ -265,18 +256,19 @@ def test_update_api_empty_fields():
resp = client.update_api(ApiId=api_id, Description="", Name="updated", Version="")
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
assert "ApiId" in resp
assert (
resp["ApiEndpoint"]
== f"https://{resp['ApiId']}.execute-api.eu-west-1.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals("s3l3ction")
resp.should.have.key("Description").equals("")
resp.should.have.key("DisableExecuteApiEndpoint").equals(True)
resp.should.have.key("DisableSchemaValidation").equals(True)
resp.should.have.key("Name").equals("updated")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals("route_s3l3ction")
resp.should.have.key("Version").equals("")
assert resp["ApiKeySelectionExpression"] == "s3l3ction"
assert resp["Description"] == ""
assert resp["DisableExecuteApiEndpoint"] is True
assert resp["DisableSchemaValidation"] is True
assert resp["Name"] == "updated"
assert resp["ProtocolType"] == "HTTP"
assert resp["RouteSelectionExpression"] == "route_s3l3ction"
assert resp["Version"] == ""
@mock_apigatewayv2
@ -303,23 +295,22 @@ def test_update_api():
Version="1.1",
)
resp.should.have.key("ApiId")
resp.should.have.key("ApiEndpoint").equals(
f"https://{resp['ApiId']}.execute-api.us-east-2.amazonaws.com"
assert "ApiId" in resp
assert (
resp["ApiEndpoint"]
== f"https://{resp['ApiId']}.execute-api.us-east-2.amazonaws.com"
)
resp.should.have.key("ApiKeySelectionExpression").equals("api_key_s3l3ction")
resp.should.have.key("CorsConfiguration").equals(
{
"AllowCredentials": True,
"AllowHeaders": ["X-Amz-Target"],
"AllowMethods": ["GET"],
}
)
resp.should.have.key("CreatedDate")
resp.should.have.key("Description").equals("updated API")
resp.should.have.key("DisableSchemaValidation").equals(True)
resp.should.have.key("DisableExecuteApiEndpoint").equals(True)
resp.should.have.key("Name").equals("new name")
resp.should.have.key("ProtocolType").equals("HTTP")
resp.should.have.key("RouteSelectionExpression").equals("route_s3l3ction")
resp.should.have.key("Version").equals("1.1")
assert resp["ApiKeySelectionExpression"] == "api_key_s3l3ction"
assert resp["CorsConfiguration"] == {
"AllowCredentials": True,
"AllowHeaders": ["X-Amz-Target"],
"AllowMethods": ["GET"],
}
assert "CreatedDate" in resp
assert resp["Description"] == "updated API"
assert resp["DisableSchemaValidation"] is True
assert resp["DisableExecuteApiEndpoint"] is True
assert resp["Name"] == "new name"
assert resp["ProtocolType"] == "HTTP"
assert resp["RouteSelectionExpression"] == "route_s3l3ction"
assert resp["Version"] == "1.1"

99
tests/test_apigatewayv2/test_apigatewayv2_authorizers.py
View File

@ -18,9 +18,9 @@ def test_create_authorizer_minimum():
AuthorizerPayloadFormatVersion="2.0",
)
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("Name").equals("auth1")
assert "AuthorizerId" in resp
assert resp["AuthorizerType"] == "REQUEST"
assert resp["Name"] == "auth1"
@mock_apigatewayv2
@ -42,20 +42,18 @@ def test_create_authorizer():
Name="auth1",
)
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerCredentialsArn").equals("auth:creds:arn")
resp.should.have.key("AuthorizerPayloadFormatVersion").equals("2.0")
resp.should.have.key("AuthorizerResultTtlInSeconds").equals(3)
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("AuthorizerUri").equals("auth_uri")
resp.should.have.key("EnableSimpleResponses").equals(True)
resp.should.have.key("IdentitySource").equals(["$request.header.Authorization"])
resp.should.have.key("IdentityValidationExpression").equals("ive")
resp.should.have.key("JwtConfiguration").equals(
{"Audience": ["a1"], "Issuer": "moto.com"}
)
resp.should.have.key("Name").equals("auth1")
resp.should.have.key("AuthorizerPayloadFormatVersion").equals("2.0")
assert "AuthorizerId" in resp
assert resp["AuthorizerCredentialsArn"] == "auth:creds:arn"
assert resp["AuthorizerPayloadFormatVersion"] == "2.0"
assert resp["AuthorizerResultTtlInSeconds"] == 3
assert resp["AuthorizerType"] == "REQUEST"
assert resp["AuthorizerUri"] == "auth_uri"
assert resp["EnableSimpleResponses"] is True
assert resp["IdentitySource"] == ["$request.header.Authorization"]
assert resp["IdentityValidationExpression"] == "ive"
assert resp["JwtConfiguration"] == {"Audience": ["a1"], "Issuer": "moto.com"}
assert resp["Name"] == "auth1"
assert resp["AuthorizerPayloadFormatVersion"] == "2.0"
@mock_apigatewayv2
@ -73,9 +71,10 @@ def test_create_authorizer_without_payloadformatversion():
)
err = exc.value.response["Error"]
err["Code"].should.equal("BadRequestException")
err["Message"].should.equal(
"AuthorizerPayloadFormatVersion is a required parameter for REQUEST authorizer"
assert err["Code"] == "BadRequestException"
assert (
err["Message"]
== "AuthorizerPayloadFormatVersion is a required parameter for REQUEST authorizer"
)
@ -94,10 +93,10 @@ def test_get_authorizer():
resp = client.get_authorizer(ApiId=api_id, AuthorizerId=authorizer_id)
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("Name").equals("auth1")
resp.should.have.key("AuthorizerPayloadFormatVersion").equals("2.0")
assert "AuthorizerId" in resp
assert resp["AuthorizerType"] == "REQUEST"
assert resp["Name"] == "auth1"
assert resp["AuthorizerPayloadFormatVersion"] == "2.0"
@mock_apigatewayv2
@ -115,7 +114,7 @@ def test_delete_authorizer():
client.get_authorizer(ApiId=api_id, AuthorizerId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
assert err["Code"] == "NotFoundException"
@mock_apigatewayv2
@ -127,7 +126,7 @@ def test_get_authorizer_unknown():
client.get_authorizer(ApiId=api_id, AuthorizerId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
assert err["Code"] == "NotFoundException"
@mock_apigatewayv2
@ -151,19 +150,17 @@ def test_update_authorizer_single():
resp = client.update_authorizer(ApiId=api_id, AuthorizerId=auth_id, Name="auth2")
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerCredentialsArn").equals("auth:creds:arn")
resp.should.have.key("AuthorizerPayloadFormatVersion").equals("2.0")
resp.should.have.key("AuthorizerResultTtlInSeconds").equals(3)
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("AuthorizerUri").equals("auth_uri")
resp.should.have.key("EnableSimpleResponses").equals(True)
resp.should.have.key("IdentitySource").equals(["$request.header.Authorization"])
resp.should.have.key("IdentityValidationExpression").equals("ive")
resp.should.have.key("JwtConfiguration").equals(
{"Audience": ["a1"], "Issuer": "moto.com"}
)
resp.should.have.key("Name").equals("auth2")
assert "AuthorizerId" in resp
assert resp["AuthorizerCredentialsArn"] == "auth:creds:arn"
assert resp["AuthorizerPayloadFormatVersion"] == "2.0"
assert resp["AuthorizerResultTtlInSeconds"] == 3
assert resp["AuthorizerType"] == "REQUEST"
assert resp["AuthorizerUri"] == "auth_uri"
assert resp["EnableSimpleResponses"] is True
assert resp["IdentitySource"] == ["$request.header.Authorization"]
assert resp["IdentityValidationExpression"] == "ive"
assert resp["JwtConfiguration"] == {"Audience": ["a1"], "Issuer": "moto.com"}
assert resp["Name"] == "auth2"
@mock_apigatewayv2
@ -196,16 +193,14 @@ def test_update_authorizer_all_attributes():
resp = client.update_authorizer(ApiId=api_id, AuthorizerId=auth_id, Name="auth2")
resp.should.have.key("AuthorizerId")
resp.should.have.key("AuthorizerCredentialsArn").equals("")
resp.should.have.key("AuthorizerPayloadFormatVersion").equals("3.0")
resp.should.have.key("AuthorizerResultTtlInSeconds").equals(5)
resp.should.have.key("AuthorizerType").equals("REQUEST")
resp.should.have.key("AuthorizerUri").equals("auth_uri")
resp.should.have.key("EnableSimpleResponses").equals(False)
resp.should.have.key("IdentitySource").equals(["$request.header.Authentication"])
resp.should.have.key("IdentityValidationExpression").equals("ive2")
resp.should.have.key("JwtConfiguration").equals(
{"Audience": ["a2"], "Issuer": "moto.com"}
)
resp.should.have.key("Name").equals("auth2")
assert "AuthorizerId" in resp
assert resp["AuthorizerCredentialsArn"] == ""
assert resp["AuthorizerPayloadFormatVersion"] == "3.0"
assert resp["AuthorizerResultTtlInSeconds"] == 5
assert resp["AuthorizerType"] == "REQUEST"
assert resp["AuthorizerUri"] == "auth_uri"
assert resp["EnableSimpleResponses"] is False
assert resp["IdentitySource"] == ["$request.header.Authentication"]
assert resp["IdentityValidationExpression"] == "ive2"
assert resp["JwtConfiguration"] == {"Audience": ["a2"], "Issuer": "moto.com"}
assert resp["Name"] == "auth2"

36
tests/test_apigatewayv2/test_apigatewayv2_domains.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
import pytest
import botocore.exceptions
from moto import mock_apigatewayv2
@ -23,19 +22,19 @@ def test_create_domain_name():
# check post response has all keys
for key in expected_keys:
post_resp.should.have.key(key)
assert key in post_resp
# check get response has all keys
for key in expected_keys:
get_resp.should.have.key(key)
assert key in get_resp
# check that values are equal for post and get of same resource
for key in expected_keys:
post_resp.get(key).should.equal(get_resp.get(key))
assert post_resp.get(key) == get_resp.get(key)
# ensure known values are set correct in post response
post_resp.get("DomainName").should.equal(domain_name)
post_resp.get("Tags").should.equal(tags)
assert post_resp.get("DomainName") == domain_name
assert post_resp.get("Tags") == tags
@mock_apigatewayv2
@ -47,8 +46,8 @@ def test_create_domain_name_already_exists():
client.create_domain_name(DomainName="exists.io")
err = exc.value.response["Error"]
err["Code"].should.equal("ConflictException")
err["Message"].should.equal("The domain name resource already exists.")
assert err["Code"] == "ConflictException"
assert err["Message"] == "The domain name resource already exists."
@mock_apigatewayv2
@ -58,17 +57,17 @@ def test_get_domain_names():
prod_domain = client.create_domain_name(DomainName="prod.service.io")
# sanity check responses
dev_domain.should.have.key("DomainName").equals("dev.service.io")
prod_domain.should.have.key("DomainName").equals("prod.service.io")
assert dev_domain["DomainName"] == "dev.service.io"
assert prod_domain["DomainName"] == "prod.service.io"
# make comparable
del dev_domain["ResponseMetadata"]
del prod_domain["ResponseMetadata"]
get_resp = client.get_domain_names()
get_resp.should.have.key("Items")
get_resp.get("Items").should.contain(dev_domain)
get_resp.get("Items").should.contain(prod_domain)
assert "Items" in get_resp
assert dev_domain in get_resp.get("Items")
assert prod_domain in get_resp.get("Items")
@mock_apigatewayv2
@ -79,8 +78,8 @@ def test_delete_domain_name():
get_resp = client.get_domain_names()
del post_resp["ResponseMetadata"]
get_resp.should.have.key("Items")
get_resp.get("Items").should_not.contain(post_resp)
assert "Items" in get_resp
assert post_resp not in get_resp.get("Items")
@mock_apigatewayv2
@ -90,7 +89,8 @@ def test_delete_domain_name_dne():
client.delete_domain_name(DomainName="dne.io")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal(
"The domain name resource specified in the request was not found."
assert err["Code"] == "NotFoundException"
assert (
err["Message"]
== "The domain name resource specified in the request was not found."
)

52
tests/test_apigatewayv2/test_apigatewayv2_integrationresponses.py
View File

@ -14,7 +14,7 @@ def test_get_integration_responses_empty():
]
resp = client.get_integration_responses(ApiId=api_id, IntegrationId=int_id)
resp.should.have.key("Items").equals([])
assert resp["Items"] == []
@mock_apigatewayv2
@ -36,12 +36,12 @@ def test_create_integration_response():
TemplateSelectionExpression="tse",
)
int_res.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
int_res.should.have.key("IntegrationResponseId")
int_res.should.have.key("IntegrationResponseKey").equals("int_res_key")
int_res.should.have.key("ResponseParameters").equals({"x-header": "header-value"})
int_res.should.have.key("ResponseTemplates").equals({"t": "template"})
int_res.should.have.key("TemplateSelectionExpression").equals("tse")
assert int_res["ContentHandlingStrategy"] == "CONVERT_TO_BINARY"
assert "IntegrationResponseId" in int_res
assert int_res["IntegrationResponseKey"] == "int_res_key"
assert int_res["ResponseParameters"] == {"x-header": "header-value"}
assert int_res["ResponseTemplates"] == {"t": "template"}
assert int_res["TemplateSelectionExpression"] == "tse"
@mock_apigatewayv2
@ -66,12 +66,12 @@ def test_get_integration_response():
ApiId=api_id, IntegrationId=int_id, IntegrationResponseId=int_res_id
)
int_res.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
int_res.should.have.key("IntegrationResponseId")
int_res.should.have.key("IntegrationResponseKey").equals("int_res_key")
int_res.should.have.key("ResponseParameters").equals({"x-header": "header-value"})
int_res.should.have.key("ResponseTemplates").equals({"t": "template"})
int_res.should.have.key("TemplateSelectionExpression").equals("tse")
assert int_res["ContentHandlingStrategy"] == "CONVERT_TO_BINARY"
assert "IntegrationResponseId" in int_res
assert int_res["IntegrationResponseKey"] == "int_res_key"
assert int_res["ResponseParameters"] == {"x-header": "header-value"}
assert int_res["ResponseTemplates"] == {"t": "template"}
assert int_res["TemplateSelectionExpression"] == "tse"
@mock_apigatewayv2
@ -87,10 +87,8 @@ def test_get_integration_response_unknown():
ApiId=api_id, IntegrationId=int_id, IntegrationResponseId="unknown"
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal(
"Invalid IntegrationResponse identifier specified unknown"
)
assert err["Code"] == "NotFoundException"
assert err["Message"] == "Invalid IntegrationResponse identifier specified unknown"
@mock_apigatewayv2
@ -114,7 +112,7 @@ def test_delete_integration_response():
ApiId=api_id, IntegrationId=int_id, IntegrationResponseId=int_res_id
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
assert err["Code"] == "NotFoundException"
@mock_apigatewayv2
@ -136,9 +134,9 @@ def test_update_integration_response_single_attr():
ContentHandlingStrategy="CONVERT_TO_BINARY",
)
res.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
res.should.have.key("IntegrationResponseId")
res.should.have.key("IntegrationResponseKey").equals("int_res_key")
assert res["ContentHandlingStrategy"] == "CONVERT_TO_BINARY"
assert "IntegrationResponseId" in res
assert res["IntegrationResponseKey"] == "int_res_key"
@mock_apigatewayv2
@ -166,9 +164,9 @@ def test_update_integration_response_multiple_attrs():
TemplateSelectionExpression="tse",
)
res.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
res.should.have.key("IntegrationResponseId")
res.should.have.key("IntegrationResponseKey").equals("int_res_key2")
res.should.have.key("ResponseParameters").equals({"x-header": "header-value"})
res.should.have.key("ResponseTemplates").equals({"t": "template"})
res.should.have.key("TemplateSelectionExpression").equals("tse")
assert res["ContentHandlingStrategy"] == "CONVERT_TO_BINARY"
assert "IntegrationResponseId" in res
assert res["IntegrationResponseKey"] == "int_res_key2"
assert res["ResponseParameters"] == {"x-header": "header-value"}
assert res["ResponseTemplates"] == {"t": "template"}
assert res["TemplateSelectionExpression"] == "tse"

172
tests/test_apigatewayv2/test_apigatewayv2_integrations.py
View File

@ -11,7 +11,7 @@ def test_get_integrations_empty():
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.get_integrations(ApiId=api_id)
resp.should.have.key("Items").equals([])
assert resp["Items"] == []
@mock_apigatewayv2
@ -21,8 +21,8 @@ def test_create_integration_minimum():
resp = client.create_integration(ApiId=api_id, IntegrationType="HTTP")
resp.should.have.key("IntegrationId")
resp.should.have.key("IntegrationType").equals("HTTP")
assert "IntegrationId" in resp
assert resp["IntegrationType"] == "HTTP"
@mock_apigatewayv2
@ -34,10 +34,11 @@ def test_create_integration_for_internet_mock():
ApiId=api_id, ConnectionType="INTERNET", IntegrationType="MOCK"
)
resp.should.have.key("IntegrationId")
resp.should.have.key("IntegrationType").equals("MOCK")
resp.should.have.key("IntegrationResponseSelectionExpression").equals(
"${integration.response.statuscode}"
assert "IntegrationId" in resp
assert resp["IntegrationType"] == "MOCK"
assert (
resp["IntegrationResponseSelectionExpression"]
== "${integration.response.statuscode}"
)
@ -66,23 +67,23 @@ def test_create_integration_full():
TlsConfig={"ServerNameToVerify": "server"},
)
resp.should.have.key("IntegrationId")
resp.should.have.key("ConnectionId").equals("conn_id")
resp.should.have.key("ConnectionType").equals("INTERNET")
resp.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
resp.should.have.key("CredentialsArn").equals("cred:arn")
resp.should.have.key("Description").equals("my full integration")
resp.should.have.key("IntegrationMethod").equals("PUT")
resp.should.have.key("IntegrationType").equals("HTTP")
resp.should.have.key("IntegrationSubtype").equals("n/a")
resp.should.have.key("PassthroughBehavior").equals("WHEN_NO_MATCH")
resp.should.have.key("PayloadFormatVersion").equals("1.0")
resp.should.have.key("RequestParameters").equals({"r": "p"})
resp.should.have.key("RequestTemplates").equals({"r": "t"})
resp.should.have.key("ResponseParameters").equals({"res": {"par": "am"}})
resp.should.have.key("TemplateSelectionExpression").equals("tse")
resp.should.have.key("TimeoutInMillis").equals(123)
resp.should.have.key("TlsConfig").equals({"ServerNameToVerify": "server"})
assert "IntegrationId" in resp
assert resp["ConnectionId"] == "conn_id"
assert resp["ConnectionType"] == "INTERNET"
assert resp["ContentHandlingStrategy"] == "CONVERT_TO_BINARY"
assert resp["CredentialsArn"] == "cred:arn"
assert resp["Description"] == "my full integration"
assert resp["IntegrationMethod"] == "PUT"
assert resp["IntegrationType"] == "HTTP"
assert resp["IntegrationSubtype"] == "n/a"
assert resp["PassthroughBehavior"] == "WHEN_NO_MATCH"
assert resp["PayloadFormatVersion"] == "1.0"
assert resp["RequestParameters"] == {"r": "p"}
assert resp["RequestTemplates"] == {"r": "t"}
assert resp["ResponseParameters"] == {"res": {"par": "am"}}
assert resp["TemplateSelectionExpression"] == "tse"
assert resp["TimeoutInMillis"] == 123
assert resp["TlsConfig"] == {"ServerNameToVerify": "server"}
@mock_apigatewayv2
@ -96,8 +97,8 @@ def test_get_integration():
resp = client.get_integration(ApiId=api_id, IntegrationId=integration_id)
resp.should.have.key("IntegrationId")
resp.should.have.key("IntegrationType").equals("HTTP")
assert "IntegrationId" in resp
assert resp["IntegrationType"] == "HTTP"
@mock_apigatewayv2
@ -108,8 +109,8 @@ def test_get_integration_unknown():
with pytest.raises(ClientError) as exc:
client.get_integration(ApiId=api_id, IntegrationId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid Integration identifier specified unknown")
assert err["Code"] == "NotFoundException"
assert err["Message"] == "Invalid Integration identifier specified unknown"
@mock_apigatewayv2
@ -122,8 +123,8 @@ def test_get_integrations():
]
resp = client.get_integrations(ApiId=api_id)
resp.should.have.key("Items").length_of(1)
resp["Items"][0].should.have.key("IntegrationId").equals(integration_id)
assert len(resp["Items"]) == 1
assert resp["Items"][0]["IntegrationId"] == integration_id
@mock_apigatewayv2
@ -140,7 +141,7 @@ def test_delete_integration():
with pytest.raises(ClientError) as exc:
client.get_integration(ApiId=api_id, IntegrationId=integration_id)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
assert err["Code"] == "NotFoundException"
@mock_apigatewayv2
@ -172,23 +173,23 @@ def test_update_integration_single_attr():
ApiId=api_id, IntegrationId=int_id, Description="updated int"
)
resp.should.have.key("IntegrationId")
resp.should.have.key("ConnectionId").equals("conn_id")
resp.should.have.key("ConnectionType").equals("INTERNET")
resp.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_BINARY")
resp.should.have.key("CredentialsArn").equals("cred:arn")
resp.should.have.key("Description").equals("updated int")
resp.should.have.key("IntegrationMethod").equals("PUT")
resp.should.have.key("IntegrationType").equals("HTTP")
resp.should.have.key("IntegrationSubtype").equals("n/a")
resp.should.have.key("PassthroughBehavior").equals("WHEN_NO_MATCH")
resp.should.have.key("PayloadFormatVersion").equals("1.0")
resp.should.have.key("RequestParameters").equals({"r": "p"})
resp.should.have.key("RequestTemplates").equals({"r": "t"})
resp.should.have.key("ResponseParameters").equals({"res": {"par": "am"}})
resp.should.have.key("TemplateSelectionExpression").equals("tse")
resp.should.have.key("TimeoutInMillis").equals(123)
resp.should.have.key("TlsConfig").equals({"ServerNameToVerify": "server"})
assert "IntegrationId" in resp
assert resp["ConnectionId"] == "conn_id"
assert resp["ConnectionType"] == "INTERNET"
assert resp["ContentHandlingStrategy"] == "CONVERT_TO_BINARY"
assert resp["CredentialsArn"] == "cred:arn"
assert resp["Description"] == "updated int"
assert resp["IntegrationMethod"] == "PUT"
assert resp["IntegrationType"] == "HTTP"
assert resp["IntegrationSubtype"] == "n/a"
assert resp["PassthroughBehavior"] == "WHEN_NO_MATCH"
assert resp["PayloadFormatVersion"] == "1.0"
assert resp["RequestParameters"] == {"r": "p"}
assert resp["RequestTemplates"] == {"r": "t"}
assert resp["ResponseParameters"] == {"res": {"par": "am"}}
assert resp["TemplateSelectionExpression"] == "tse"
assert resp["TimeoutInMillis"] == 123
assert resp["TlsConfig"] == {"ServerNameToVerify": "server"}
@mock_apigatewayv2
@ -227,23 +228,23 @@ def test_update_integration_all_attrs():
PassthroughBehavior="NEVER",
)
resp.should.have.key("IntegrationId")
resp.should.have.key("ConnectionId").equals("conn_id")
resp.should.have.key("ConnectionType").equals("VPC_LINK")
resp.should.have.key("ContentHandlingStrategy").equals("CONVERT_TO_TEXT")
resp.should.have.key("CredentialsArn").equals("")
resp.should.have.key("Description").equals("my full integration")
resp.should.have.key("IntegrationMethod").equals("PATCH")
resp.should.have.key("IntegrationType").equals("AWS")
resp.should.have.key("IntegrationSubtype").equals("n/a")
resp.should.have.key("PassthroughBehavior").equals("NEVER")
resp.should.have.key("PayloadFormatVersion").equals("1.0")
resp.should.have.key("RequestParameters").equals({"r": "p"})
resp.should.have.key("RequestTemplates").equals({"r": "t"})
resp.should.have.key("ResponseParameters").equals({"res": {"par": "am"}})
resp.should.have.key("TemplateSelectionExpression").equals("tse")
resp.should.have.key("TimeoutInMillis").equals(123)
resp.should.have.key("TlsConfig").equals({"ServerNameToVerify": "server"})
assert "IntegrationId" in resp
assert resp["ConnectionId"] == "conn_id"
assert resp["ConnectionType"] == "VPC_LINK"
assert resp["ContentHandlingStrategy"] == "CONVERT_TO_TEXT"
assert resp["CredentialsArn"] == ""
assert resp["Description"] == "my full integration"
assert resp["IntegrationMethod"] == "PATCH"
assert resp["IntegrationType"] == "AWS"
assert resp["IntegrationSubtype"] == "n/a"
assert resp["PassthroughBehavior"] == "NEVER"
assert resp["PayloadFormatVersion"] == "1.0"
assert resp["RequestParameters"] == {"r": "p"}
assert resp["RequestTemplates"] == {"r": "t"}
assert resp["ResponseParameters"] == {"res": {"par": "am"}}
assert resp["TemplateSelectionExpression"] == "tse"
assert resp["TimeoutInMillis"] == 123
assert resp["TlsConfig"] == {"ServerNameToVerify": "server"}
@mock_apigatewayv2
@ -273,9 +274,10 @@ def test_update_integration_request_parameters():
int_id = resp["IntegrationId"]
# Having an empty value between quotes ("''") is valid
resp["RequestParameters"].should.equal(
{"append:header.header1": "$context.requestId", "remove:querystring.qs1": "''"}
)
assert resp["RequestParameters"] == {
"append:header.header1": "$context.requestId",
"remove:querystring.qs1": "''",
}
resp = client.update_integration(
ApiId=api_id,
@ -295,23 +297,19 @@ def test_update_integration_request_parameters():
},
)
resp.should.have.key("IntegrationId")
resp.should.have.key("IntegrationMethod").equals("ANY")
resp.should.have.key("IntegrationType").equals("HTTP_PROXY")
resp.should.have.key("PayloadFormatVersion").equals("1.0")
assert "IntegrationId" in resp
assert resp["IntegrationMethod"] == "ANY"
assert resp["IntegrationType"] == "HTTP_PROXY"
assert resp["PayloadFormatVersion"] == "1.0"
# Having no value ("") is not valid, so that param should not be persisted
resp.should.have.key("RequestParameters").equals(
{
"append:header.header1": "$context.accountId",
"overwrite:header.header2": "$stageVariables.environmentId",
}
)
resp.should.have.key("ResponseParameters").equals(
{
"404": {},
"500": {
"append:header.header1": "$context.requestId",
"overwrite:statuscode": "403",
},
}
)
assert resp["RequestParameters"] == {
"append:header.header1": "$context.accountId",
"overwrite:header.header2": "$stageVariables.environmentId",
}
assert resp["ResponseParameters"] == {
"404": {},
"500": {
"append:header.header1": "$context.requestId",
"overwrite:statuscode": "403",
},
}

62
tests/test_apigatewayv2/test_apigatewayv2_mappings.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
import pytest
import botocore.exceptions
from moto import mock_apigatewayv2
@ -25,21 +24,21 @@ def test_create_api_mapping():
# check post response has all expected keys
for key in expected_keys:
post_resp.should.have.key(key)
assert key in post_resp
# check get response has all expected keys
for key in expected_keys:
get_resp.should.have.key(key)
assert key in get_resp
# check that values are equal for post and get of same resource
for key in expected_keys:
post_resp.get(key).should.equal(get_resp.get(key))
assert post_resp.get(key) == get_resp.get(key)
# ensure known values are set correct in post response
post_resp.get("ApiId").should_not.equal(None)
post_resp.get("ApiMappingId").should_not.equal(None)
post_resp.get("ApiMappingKey").should.equal("v1/api")
post_resp.get("Stage").should.equal("$default")
assert post_resp.get("ApiId") is not None
assert post_resp.get("ApiMappingId") is not None
assert post_resp.get("ApiMappingKey") == "v1/api"
assert post_resp.get("Stage") == "$default"
@mock_apigatewayv2
@ -54,9 +53,10 @@ def test_create_api_mapping_missing_api():
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal(
"Invalid API identifier specified The resource specified in the request was not found."
assert err["Code"] == "NotFoundException"
assert (
err["Message"]
== "Invalid API identifier specified The resource specified in the request was not found."
)
@ -73,9 +73,10 @@ def test_create_api_mapping_missing_domain():
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal(
"The domain name resource specified in the request was not found."
assert err["Code"] == "NotFoundException"
assert (
err["Message"]
== "The domain name resource specified in the request was not found."
)
@ -100,8 +101,8 @@ def test_create_api_mapping_bad_mapping_keys():
)
err = exc.value.response["Error"]
err["Code"].should.equal("BadRequestException")
err["Message"].should.equal(message)
assert err["Code"] == "BadRequestException"
assert err["Message"] == message
@mock_apigatewayv2
@ -133,9 +134,9 @@ def test_get_api_mappings():
)
# sanity check responses
v1_mapping.should.have.key("ApiMappingKey").equals("v1/api")
v2_mapping.should.have.key("ApiMappingKey").equals("v2/api")
hr_mapping.should.have.key("ApiMappingKey").equals("hr/api")
assert v1_mapping["ApiMappingKey"] == "v1/api"
assert v2_mapping["ApiMappingKey"] == "v2/api"
assert hr_mapping["ApiMappingKey"] == "hr/api"
# make comparable
del v1_mapping["ResponseMetadata"]
@ -143,10 +144,10 @@ def test_get_api_mappings():
del hr_mapping["ResponseMetadata"]
get_resp = client.get_api_mappings(DomainName=included_domain["DomainName"])
get_resp.should.have.key("Items")
get_resp.get("Items").should.contain(v1_mapping)
get_resp.get("Items").should.contain(v2_mapping)
get_resp.get("Items").should_not.contain(hr_mapping)
assert "Items" in get_resp
assert v1_mapping in get_resp.get("Items")
assert v2_mapping in get_resp.get("Items")
assert hr_mapping not in get_resp.get("Items")
@mock_apigatewayv2
@ -165,16 +166,16 @@ def test_delete_api_mapping():
del v1_mapping["ResponseMetadata"]
get_resp = client.get_api_mappings(DomainName=api_domain["DomainName"])
get_resp.should.have.key("Items")
get_resp.get("Items").should.contain(v1_mapping)
assert "Items" in get_resp
assert v1_mapping in get_resp.get("Items")
client.delete_api_mapping(
DomainName=api_domain["DomainName"], ApiMappingId=v1_mapping["ApiMappingId"]
)
get_resp = client.get_api_mappings(DomainName=api_domain["DomainName"])
get_resp.should.have.key("Items")
get_resp.get("Items").should_not.contain(v1_mapping)
assert "Items" in get_resp
assert v1_mapping not in get_resp.get("Items")
@mock_apigatewayv2
@ -189,7 +190,8 @@ def test_delete_api_mapping_dne():
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal(
"The api mapping resource specified in the request was not found."
assert err["Code"] == "NotFoundException"
assert (
err["Message"]
== "The api mapping resource specified in the request was not found."
)

42
tests/test_apigatewayv2/test_apigatewayv2_models.py
View File

@ -14,11 +14,11 @@ def test_create_model():
ApiId=api_id, ContentType="app/xml", Description="desc", Name="nm", Schema="cs"
)
resp.should.have.key("ContentType").equals("app/xml")
resp.should.have.key("Description").equals("desc")
resp.should.have.key("ModelId")
resp.should.have.key("Name").equals("nm")
resp.should.have.key("Schema").equals("cs")
assert resp["ContentType"] == "app/xml"
assert resp["Description"] == "desc"
assert "ModelId" in resp
assert resp["Name"] == "nm"
assert resp["Schema"] == "cs"
@mock_apigatewayv2
@ -32,11 +32,11 @@ def test_get_model():
resp = client.get_model(ApiId=api_id, ModelId=model_id)
resp.should.have.key("ContentType").equals("app/xml")
resp.should.have.key("Description").equals("desc")
resp.should.have.key("ModelId")
resp.should.have.key("Name").equals("nm")
resp.should.have.key("Schema").equals("cs")
assert resp["ContentType"] == "app/xml"
assert resp["Description"] == "desc"
assert "ModelId" in resp
assert resp["Name"] == "nm"
assert resp["Schema"] == "cs"
@mock_apigatewayv2
@ -54,7 +54,7 @@ def test_delete_model():
client.get_model(ApiId=api_id, ModelId=model_id)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
assert err["Code"] == "NotFoundException"
@mock_apigatewayv2
@ -66,7 +66,7 @@ def test_get_model_unknown():
client.get_model(ApiId=api_id, ModelId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
assert err["Code"] == "NotFoundException"
@mock_apigatewayv2
@ -82,11 +82,11 @@ def test_update_model_single_attr():
resp = client.update_model(ApiId=api_id, ModelId=model_id, Schema="cs2")
resp.should.have.key("ContentType").equals("app/xml")
resp.should.have.key("Description").equals("desc")
resp.should.have.key("ModelId")
resp.should.have.key("Name").equals("nm")
resp.should.have.key("Schema").equals("cs2")
assert resp["ContentType"] == "app/xml"
assert resp["Description"] == "desc"
assert "ModelId" in resp
assert resp["Name"] == "nm"
assert resp["Schema"] == "cs2"
@mock_apigatewayv2
@ -109,7 +109,7 @@ def test_update_model_all_attrs():
Schema="cs2",
)
resp.should.have.key("ContentType").equals("app/html")
resp.should.have.key("Description").equals("html2.x")
resp.should.have.key("Name").equals("html-schema")
resp.should.have.key("Schema").equals("cs2")
assert resp["ContentType"] == "app/html"
assert resp["Description"] == "html2.x"
assert resp["Name"] == "html-schema"
assert resp["Schema"] == "cs2"

50
tests/test_apigatewayv2/test_apigatewayv2_reimport.py
View File

@ -1,6 +1,5 @@
import boto3
import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError
from moto import mock_apigatewayv2
@ -16,9 +15,9 @@ def test_reimport_api_standard_fields():
Body="---\nopenapi: 3.0.1\ninfo:\n title: tf-acc-test-2983214806752144669_DIFFERENT\n version: 2.0\nx-amazon-apigateway-cors:\n allow_methods:\n - delete\n allow_origins:\n - https://www.google.de\npaths:\n \"/test\":\n get:\n x-amazon-apigateway-integration:\n type: HTTP_PROXY\n httpMethod: GET\n payloadFormatVersion: '1.0'\n uri: https://www.google.de\n",
)
resp.should.have.key("CorsConfiguration").equals({})
resp.should.have.key("Name").equals("tf-acc-test-2983214806752144669_DIFFERENT")
resp.should.have.key("Version").equals("2.0")
assert resp["CorsConfiguration"] == {}
assert resp["Name"] == "tf-acc-test-2983214806752144669_DIFFERENT"
assert resp["Version"] == "2.0"
@mock_apigatewayv2
@ -33,14 +32,15 @@ def test_reimport_api_failonwarnings():
FailOnWarnings=True,
)
err = exc.value.response["Error"]
err["Code"].should.equal("BadRequestException")
err["Message"].should.equal(
"Warnings found during import:\n\tParse issue: attribute paths.'/update'(get).responses.200.content.schema.#/components/schemas/ModelThatDoesNotExist is missing"
assert err["Code"] == "BadRequestException"
assert (
err["Message"]
== "Warnings found during import:\n\tParse issue: attribute paths.'/update'(get).responses.200.content.schema.#/components/schemas/ModelThatDoesNotExist is missing"
)
# Title should not have been updated
resp = client.get_api(ApiId=api_id)
resp.should.have.key("Name").equals("test-import-api")
assert resp["Name"] == "test-import-api"
@mock_apigatewayv2
@ -56,7 +56,7 @@ def test_reimport_api_do_not_failonwarnings():
# Title should have been updated
resp = client.get_api(ApiId=api_id)
resp.should.have.key("Name").equals("Title test")
assert resp["Name"] == "Title test"
@mock_apigatewayv2
@ -71,24 +71,22 @@ def test_reimport_api_routes_and_integrations():
)
resp = client.get_integrations(ApiId=api_id)
resp.should.have.key("Items").length_of(1)
assert len(resp["Items"]) == 1
integration = resp["Items"][0]
integration.should.have.key("ConnectionType").equals("INTERNET")
integration.should.have.key("IntegrationId")
integration.should.have.key("IntegrationMethod").equals("GET")
integration.should.have.key("IntegrationType").equals("HTTP_PROXY")
integration.should.have.key("IntegrationUri").equals("https://www.google.de")
integration.should.have.key("PayloadFormatVersion").equals("1.0")
integration.should.have.key("TimeoutInMillis").equals(30000)
assert integration["ConnectionType"] == "INTERNET"
assert "IntegrationId" in integration
assert integration["IntegrationMethod"] == "GET"
assert integration["IntegrationType"] == "HTTP_PROXY"
assert integration["IntegrationUri"] == "https://www.google.de"
assert integration["PayloadFormatVersion"] == "1.0"
assert integration["TimeoutInMillis"] == 30000
resp = client.get_routes(ApiId=api_id)
resp.should.have.key("Items").length_of(1)
assert len(resp["Items"]) == 1
route = resp["Items"][0]
route.should.have.key("ApiKeyRequired").equals(False)
route.should.have.key("AuthorizationScopes").equals([])
route.should.have.key("RequestParameters").equals({})
route.should.have.key("RouteId")
route.should.have.key("RouteKey").equals("GET /test")
route.should.have.key("Target").should.equal(
f"integrations/{integration['IntegrationId']}"
)
assert route["ApiKeyRequired"] is False
assert route["AuthorizationScopes"] == []
assert route["RequestParameters"] == {}
assert "RouteId" in route
assert route["RouteKey"] == "GET /test"
assert route["Target"] == f"integrations/{integration['IntegrationId']}"

138
tests/test_apigatewayv2/test_apigatewayv2_routes.py
View File

@ -11,7 +11,7 @@ def test_get_routes_empty():
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.get_routes(ApiId=api_id)
resp.should.have.key("Items").equals([])
assert resp["Items"] == []
@mock_apigatewayv2
@ -20,18 +20,18 @@ def test_create_route_minimal():
api_id = client.create_api(Name="test-api", ProtocolType="HTTP")["ApiId"]
resp = client.create_route(ApiId=api_id, RouteKey="GET /")
resp.should.have.key("ApiKeyRequired").equals(False)
resp.should.have.key("AuthorizationType").equals("NONE")
resp.should.have.key("RouteId")
resp.should.have.key("RouteKey").equals("GET /")
assert resp["ApiKeyRequired"] is False
assert resp["AuthorizationType"] == "NONE"
assert "RouteId" in resp
assert resp["RouteKey"] == "GET /"
resp.shouldnt.have.key("AuthorizationScopes")
resp.shouldnt.have.key("AuthorizerId")
resp.shouldnt.have.key("ModelSelectionExpression")
resp.shouldnt.have.key("OperationName")
resp.shouldnt.have.key("RequestModels")
resp.shouldnt.have.key("RouteResponseSelectionExpression")
resp.shouldnt.have.key("Target")
assert "AuthorizationScopes" not in resp
assert "AuthorizerId" not in resp
assert "ModelSelectionExpression" not in resp
assert "OperationName" not in resp
assert "RequestModels" not in resp
assert "RouteResponseSelectionExpression" not in resp
assert "Target" not in resp
@mock_apigatewayv2
@ -57,19 +57,19 @@ def test_create_route_full():
Target="t",
)
resp.should.have.key("ApiKeyRequired").equals(True)
resp.should.have.key("AuthorizationType").equals("CUSTOM")
resp.should.have.key("AuthorizationScopes").equals(["scope1", "scope2"])
resp.should.have.key("AuthorizerId").equals("auth_id")
resp.should.have.key("RouteId")
resp.should.have.key("RouteKey").equals("GET /")
assert resp["ApiKeyRequired"] is True
assert resp["AuthorizationType"] == "CUSTOM"
assert resp["AuthorizationScopes"] == ["scope1", "scope2"]
assert resp["AuthorizerId"] == "auth_id"
assert "RouteId" in resp
assert resp["RouteKey"] == "GET /"
resp.should.have.key("ModelSelectionExpression").equals("mse")
resp.should.have.key("OperationName").equals("OP")
resp.should.have.key("RequestModels").equals({"req": "uest"})
resp.should.have.key("RequestParameters").equals({"action": {"Required": True}})
resp.should.have.key("RouteResponseSelectionExpression").equals("$default")
resp.should.have.key("Target").equals("t")
assert resp["ModelSelectionExpression"] == "mse"
assert resp["OperationName"] == "OP"
assert resp["RequestModels"] == {"req": "uest"}
assert resp["RequestParameters"] == {"action": {"Required": True}}
assert resp["RouteResponseSelectionExpression"] == "$default"
assert resp["Target"] == "t"
@mock_apigatewayv2
@ -81,7 +81,7 @@ def test_delete_route():
client.delete_route(ApiId=api_id, RouteId=route_id)
resp = client.get_routes(ApiId=api_id)
resp.should.have.key("Items").length_of(0)
assert len(resp["Items"]) == 0
@mock_apigatewayv2
@ -92,17 +92,17 @@ def test_get_route():
resp = client.get_route(ApiId=api_id, RouteId=route_id)
resp.should.have.key("ApiKeyRequired").equals(False)
resp.should.have.key("AuthorizationType").equals("NONE")
resp.should.have.key("RouteId")
resp.should.have.key("RouteKey").equals("GET /")
assert resp["ApiKeyRequired"] is False
assert resp["AuthorizationType"] == "NONE"
assert "RouteId" in resp
assert resp["RouteKey"] == "GET /"
resp.shouldnt.have.key("AuthorizationScopes")
resp.shouldnt.have.key("AuthorizerId")
resp.shouldnt.have.key("ModelSelectionExpression")
resp.shouldnt.have.key("OperationName")
resp.shouldnt.have.key("RouteResponseSelectionExpression")
resp.shouldnt.have.key("Target")
assert "AuthorizationScopes" not in resp
assert "AuthorizerId" not in resp
assert "ModelSelectionExpression" not in resp
assert "OperationName" not in resp
assert "RouteResponseSelectionExpression" not in resp
assert "Target" not in resp
@mock_apigatewayv2
@ -113,8 +113,8 @@ def test_get_route_unknown():
client.get_route(ApiId=api_id, RouteId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid Route identifier specified unknown")
assert err["Code"] == "NotFoundException"
assert err["Message"] == "Invalid Route identifier specified unknown"
@mock_apigatewayv2
@ -124,7 +124,7 @@ def test_get_routes():
client.create_route(ApiId=api_id, RouteKey="GET /")
resp = client.get_routes(ApiId=api_id)
resp.should.have.key("Items").length_of(1)
assert len(resp["Items"]) == 1
@mock_apigatewayv2
@ -135,10 +135,10 @@ def test_update_route_single_attribute():
resp = client.update_route(ApiId=api_id, RouteId=route_id, RouteKey="POST /")
resp.should.have.key("ApiKeyRequired").equals(False)
resp.should.have.key("AuthorizationType").equals("NONE")
resp.should.have.key("RouteId").equals(route_id)
resp.should.have.key("RouteKey").equals("POST /")
assert resp["ApiKeyRequired"] is False
assert resp["AuthorizationType"] == "NONE"
assert resp["RouteId"] == route_id
assert resp["RouteKey"] == "POST /"
@mock_apigatewayv2
@ -164,18 +164,18 @@ def test_update_route_all_attributes():
Target="t",
)
resp.should.have.key("ApiKeyRequired").equals(False)
resp.should.have.key("AuthorizationType").equals("JWT")
resp.should.have.key("AuthorizationScopes").equals(["scope"])
resp.should.have.key("AuthorizerId").equals("auth_id")
resp.should.have.key("RouteId")
resp.should.have.key("RouteKey").equals("GET /")
resp.should.have.key("ModelSelectionExpression").equals("mse")
resp.should.have.key("OperationName").equals("OP")
resp.should.have.key("RequestModels").equals({"req": "uest"})
resp.should.have.key("RequestParameters").equals({"action": {"Required": True}})
resp.should.have.key("RouteResponseSelectionExpression").equals("$default")
resp.should.have.key("Target").equals("t")
assert resp["ApiKeyRequired"] is False
assert resp["AuthorizationType"] == "JWT"
assert resp["AuthorizationScopes"] == ["scope"]
assert resp["AuthorizerId"] == "auth_id"
assert "RouteId" in resp
assert resp["RouteKey"] == "GET /"
assert resp["ModelSelectionExpression"] == "mse"
assert resp["OperationName"] == "OP"
assert resp["RequestModels"] == {"req": "uest"}
assert resp["RequestParameters"] == {"action": {"Required": True}}
assert resp["RouteResponseSelectionExpression"] == "$default"
assert resp["Target"] == "t"
@mock_apigatewayv2
@ -195,7 +195,7 @@ def test_delete_route_request_parameter():
request_params = client.get_route(ApiId=api_id, RouteId=route_id)[
"RequestParameters"
]
request_params.keys().should.have.length_of(3)
assert len(request_params.keys()) == 3
client.delete_route_request_parameter(
ApiId=api_id,
@ -206,9 +206,9 @@ def test_delete_route_request_parameter():
request_params = client.get_route(ApiId=api_id, RouteId=route_id)[
"RequestParameters"
]
request_params.keys().should.have.length_of(2)
request_params.should.have.key("action")
request_params.should.have.key("zparam")
assert len(request_params.keys()) == 2
assert "action" in request_params
assert "zparam" in request_params
@mock_apigatewayv2
@ -221,8 +221,8 @@ def test_create_route_response_minimal():
ApiId=api_id, RouteId=route_id, RouteResponseKey="$default"
)
resp.should.have.key("RouteResponseId")
resp.should.have.key("RouteResponseKey").equals("$default")
assert "RouteResponseId" in resp
assert resp["RouteResponseKey"] == "$default"
@mock_apigatewayv2
@ -239,12 +239,10 @@ def test_create_route_response():
RouteResponseKey="$default",
)
resp.should.have.key("RouteResponseId")
resp.should.have.key("RouteResponseKey").equals("$default")
resp.should.have.key("ModelSelectionExpression").equals("mse")
resp.should.have.key("ResponseModels").equals(
{"test": "tfacctest5832545056931060873"}
)
assert "RouteResponseId" in resp
assert resp["RouteResponseKey"] == "$default"
assert resp["ModelSelectionExpression"] == "mse"
assert resp["ResponseModels"] == {"test": "tfacctest5832545056931060873"}
@mock_apigatewayv2
@ -261,8 +259,8 @@ def test_get_route_response():
ApiId=api_id, RouteId=route_id, RouteResponseId=route_response_id
)
resp.should.have.key("RouteResponseId")
resp.should.have.key("RouteResponseKey").equals("$default")
assert "RouteResponseId" in resp
assert resp["RouteResponseKey"] == "$default"
@mock_apigatewayv2
@ -277,7 +275,7 @@ def test_get_route_response_unknown():
)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
assert err["Code"] == "NotFoundException"
@mock_apigatewayv2
@ -295,4 +293,4 @@ def test_delete_route_response_unknown():
client.get_route_response(ApiId=api_id, RouteId=r_id, RouteResponseId=rr_id)
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
assert err["Code"] == "NotFoundException"

8
tests/test_apigatewayv2/test_apigatewayv2_tags.py
View File

@ -10,7 +10,7 @@ def test_create_api_with_tags():
Name="test-api", ProtocolType="HTTP", Tags={"key1": "value1", "key2": "value2"}
)
resp.should.have.key("Tags").equals({"key1": "value1", "key2": "value2"})
assert resp["Tags"] == {"key1": "value1", "key2": "value2"}
@mock_apigatewayv2
@ -25,7 +25,7 @@ def test_tag_resource():
resp = client.get_api(ApiId=api_id)
resp.should.have.key("Tags").equals({"key1": "value1", "key2": "value2"})
assert resp["Tags"] == {"key1": "value1", "key2": "value2"}
@mock_apigatewayv2
@ -40,7 +40,7 @@ def test_get_tags():
resp = client.get_tags(ResourceArn=resource_arn)
resp.should.have.key("Tags").equals({"key1": "value1", "key2": "value2"})
assert resp["Tags"] == {"key1": "value1", "key2": "value2"}
@mock_apigatewayv2
@ -59,4 +59,4 @@ def test_untag_resource():
resp = client.get_tags(ResourceArn=resource_arn)
resp.should.have.key("Tags").equals({"key1": "value1", "key3": "value3"})
assert resp["Tags"] == {"key1": "value1", "key3": "value3"}

80
tests/test_apigatewayv2/test_apigatewayv2_vpclinks.py
View File

@ -10,7 +10,7 @@ def test_get_vpc_links_empty():
client = boto3.client("apigatewayv2", region_name="eu-west-1")
resp = client.get_vpc_links()
resp.should.have.key("Items").equals([])
assert resp["Items"] == []
@mock_apigatewayv2
@ -24,14 +24,14 @@ def test_create_vpc_links():
Tags={"key1": "value1"},
)
resp.should.have.key("CreatedDate")
resp.should.have.key("Name").equals("vpcl")
resp.should.have.key("SecurityGroupIds").equals(["sg1", "sg2"])
resp.should.have.key("SubnetIds").equals(["sid1", "sid2"])
resp.should.have.key("Tags").equals({"key1": "value1"})
resp.should.have.key("VpcLinkId")
resp.should.have.key("VpcLinkStatus").equals("AVAILABLE")
resp.should.have.key("VpcLinkVersion").equals("V2")
assert "CreatedDate" in resp
assert resp["Name"] == "vpcl"
assert resp["SecurityGroupIds"] == ["sg1", "sg2"]
assert resp["SubnetIds"] == ["sid1", "sid2"]
assert resp["Tags"] == {"key1": "value1"}
assert "VpcLinkId" in resp
assert resp["VpcLinkStatus"] == "AVAILABLE"
assert resp["VpcLinkVersion"] == "V2"
@mock_apigatewayv2
@ -47,14 +47,14 @@ def test_get_vpc_link():
resp = client.get_vpc_link(VpcLinkId=vpc_link_id)
resp.should.have.key("CreatedDate")
resp.should.have.key("Name").equals("vpcl")
resp.should.have.key("SecurityGroupIds").equals(["sg1", "sg2"])
resp.should.have.key("SubnetIds").equals(["sid1", "sid2"])
resp.should.have.key("Tags").equals({"key1": "value1"})
resp.should.have.key("VpcLinkId")
resp.should.have.key("VpcLinkStatus").equals("AVAILABLE")
resp.should.have.key("VpcLinkVersion").equals("V2")
assert "CreatedDate" in resp
assert resp["Name"] == "vpcl"
assert resp["SecurityGroupIds"] == ["sg1", "sg2"]
assert resp["SubnetIds"] == ["sid1", "sid2"]
assert resp["Tags"] == {"key1": "value1"}
assert "VpcLinkId" in resp
assert resp["VpcLinkStatus"] == "AVAILABLE"
assert resp["VpcLinkVersion"] == "V2"
@mock_apigatewayv2
@ -64,8 +64,8 @@ def test_get_vpc_link_unknown():
with pytest.raises(ClientError) as exc:
client.get_vpc_link(VpcLinkId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("Invalid VpcLink identifier specified unknown")
assert err["Code"] == "NotFoundException"
assert err["Message"] == "Invalid VpcLink identifier specified unknown"
@mock_apigatewayv2
@ -80,8 +80,8 @@ def test_get_vpc_links():
)["VpcLinkId"]
links = client.get_vpc_links()["Items"]
links.should.have.length_of(1)
links[0]["VpcLinkId"].should.equal(vpc_link_id)
assert len(links) == 1
assert links[0]["VpcLinkId"] == vpc_link_id
client.create_vpc_link(
Name="vpcl",
@ -91,7 +91,7 @@ def test_get_vpc_links():
)
links = client.get_vpc_links()["Items"]
links.should.have.length_of(2)
assert len(links) == 2
@mock_apigatewayv2
@ -106,12 +106,12 @@ def test_delete_vpc_link():
)["VpcLinkId"]
links = client.get_vpc_links()["Items"]
links.should.have.length_of(1)
assert len(links) == 1
client.delete_vpc_link(VpcLinkId=vpc_link_id)
links = client.get_vpc_links()["Items"]
links.should.have.length_of(0)
assert len(links) == 0
@mock_apigatewayv2
@ -126,14 +126,14 @@ def test_update_vpc_link():
resp = client.update_vpc_link(VpcLinkId=vpc_link_id, Name="vpcl2")
resp.should.have.key("CreatedDate")
resp.should.have.key("Name").equals("vpcl2")
resp.should.have.key("SecurityGroupIds").equals(["sg1", "sg2"])
resp.should.have.key("SubnetIds").equals(["sid1", "sid2"])
resp.should.have.key("Tags").equals({"key1": "value1"})
resp.should.have.key("VpcLinkId")
resp.should.have.key("VpcLinkStatus").equals("AVAILABLE")
resp.should.have.key("VpcLinkVersion").equals("V2")
assert "CreatedDate" in resp
assert resp["Name"] == "vpcl2"
assert resp["SecurityGroupIds"] == ["sg1", "sg2"]
assert resp["SubnetIds"] == ["sid1", "sid2"]
assert resp["Tags"] == {"key1": "value1"}
assert "VpcLinkId" in resp
assert resp["VpcLinkStatus"] == "AVAILABLE"
assert resp["VpcLinkVersion"] == "V2"
@mock_apigatewayv2
@ -152,11 +152,11 @@ def test_untag_vpc_link():
resp = client.get_vpc_link(VpcLinkId=vpc_link_id)
resp.should.have.key("CreatedDate")
resp.should.have.key("Name").equals("vpcl")
resp.should.have.key("SecurityGroupIds").equals(["sg1", "sg2"])
resp.should.have.key("SubnetIds").equals(["sid1", "sid2"])
resp.should.have.key("Tags").equals({"key2": "val2"})
resp.should.have.key("VpcLinkId")
resp.should.have.key("VpcLinkStatus").equals("AVAILABLE")
resp.should.have.key("VpcLinkVersion").equals("V2")
assert "CreatedDate" in resp
assert resp["Name"] == "vpcl"
assert resp["SecurityGroupIds"] == ["sg1", "sg2"]
assert resp["SubnetIds"] == ["sid1", "sid2"]
assert resp["Tags"] == {"key2": "val2"}
assert "VpcLinkId" in resp
assert resp["VpcLinkStatus"] == "AVAILABLE"
assert resp["VpcLinkVersion"] == "V2"

5
tests/test_apigatewayv2/test_server.py
View File

@ -1,5 +1,4 @@
import json
import sure # noqa # pylint: disable=unused-import
import moto.server as server
@ -9,5 +8,5 @@ def test_apigatewayv2_list_apis():
test_client = backend.test_client()
resp = test_client.get("/v2/apis")
resp.status_code.should.equal(200)
json.loads(resp.data).should.equal({"items": []})
assert resp.status_code == 200
assert json.loads(resp.data) == {"items": []}