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

Techdebt: Replace sure with regular asserts in AppSync tests (#6381)

Browse Source
This commit is contained in:
Bert Blommers 2023-06-09 10:10:11 +00:00 committed by GitHub
parent 6e7edd5057
commit 56895eae17
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 102 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
tests/test_appsync/test_appsync.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_appsync
@ -15,17 +14,17 @@ def test_create_graphql_api():
client = boto3.client("appsync", region_name="ap-southeast-1")
resp = client.create_graphql_api(name="api1", authenticationType="API_KEY")
resp.should.have.key("graphqlApi")
assert "graphqlApi" in resp
api = resp["graphqlApi"]
api.should.have.key("name").equals("api1")
api.should.have.key("apiId")
api.should.have.key("authenticationType").equals("API_KEY")
api.should.have.key("arn").equals(
f"arn:aws:appsync:ap-southeast-1:{ACCOUNT_ID}:apis/{api['apiId']}"
assert api["name"] == "api1"
assert "apiId" in api
assert api["authenticationType"] == "API_KEY"
assert (
api["arn"] == f"arn:aws:appsync:ap-southeast-1:{ACCOUNT_ID}:apis/{api['apiId']}"
)
api.should.have.key("uris").equals({"GRAPHQL": "http://graphql.uri"})
api.should.have.key("xrayEnabled").equals(False)
assert api["uris"] == {"GRAPHQL": "http://graphql.uri"}
assert api["xrayEnabled"] is False
api.shouldnt.have.key("additionalAuthenticationProviders")
api.shouldnt.have.key("logConfig")
@ -44,23 +43,24 @@ def test_create_graphql_api_advanced():
xrayEnabled=True,
)
resp.should.have.key("graphqlApi")
assert "graphqlApi" in resp
api = resp["graphqlApi"]
api.should.have.key("name").equals("api1")
api.should.have.key("apiId")
api.should.have.key("authenticationType").equals("API_KEY")
api.should.have.key("arn").equals(
f"arn:aws:appsync:ap-southeast-1:{ACCOUNT_ID}:apis/{api['apiId']}"
assert api["name"] == "api1"
assert "apiId" in api
assert api["authenticationType"] == "API_KEY"
assert (
api["arn"] == f"arn:aws:appsync:ap-southeast-1:{ACCOUNT_ID}:apis/{api['apiId']}"
)
api.should.have.key("uris").equals({"GRAPHQL": "http://graphql.uri"})
api.should.have.key("additionalAuthenticationProviders").equals(
[{"authenticationType": "API_KEY"}]
)
api.should.have.key("logConfig").equals(
{"cloudWatchLogsRoleArn": "arn:aws:cloudwatch:role", "fieldLogLevel": "ERROR"}
)
api.should.have.key("xrayEnabled").equals(True)
assert api["uris"] == {"GRAPHQL": "http://graphql.uri"}
assert api["additionalAuthenticationProviders"] == [
{"authenticationType": "API_KEY"}
]
assert api["logConfig"] == {
"cloudWatchLogsRoleArn": "arn:aws:cloudwatch:role",
"fieldLogLevel": "ERROR",
}
assert api["xrayEnabled"] is True
@mock_appsync
@ -71,12 +71,12 @@ def test_get_graphql_api():
]["apiId"]
resp = client.get_graphql_api(apiId=api_id)
resp.should.have.key("graphqlApi")
assert "graphqlApi" in resp
api = resp["graphqlApi"]
api.should.have.key("name").equals("api1")
api.should.have.key("apiId")
api.should.have.key("authenticationType").equals("API_KEY")
assert api["name"] == "api1"
assert "apiId" in api
assert api["authenticationType"] == "API_KEY"
@mock_appsync
@ -104,22 +104,22 @@ def test_update_graphql_api():
graphql_api = client.get_graphql_api(apiId=api_id)["graphqlApi"]
graphql_api.should.have.key("name").equals("api2")
graphql_api.should.have.key("authenticationType").equals("AWS_IAM")
graphql_api.should.have.key("arn").equals(
f"arn:aws:appsync:ap-southeast-1:{ACCOUNT_ID}:apis/{graphql_api['apiId']}"
assert graphql_api["name"] == "api2"
assert graphql_api["authenticationType"] == "AWS_IAM"
assert (
graphql_api["arn"]
== f"arn:aws:appsync:ap-southeast-1:{ACCOUNT_ID}:apis/{graphql_api['apiId']}"
)
graphql_api.should.have.key("logConfig").equals(
{"cloudWatchLogsRoleArn": "arn:aws:cloudwatch:role", "fieldLogLevel": "ERROR"}
)
graphql_api.should.have.key("userPoolConfig").equals(
{
"awsRegion": "us-east-1",
"defaultAction": "DENY",
"userPoolId": "us-east-1_391729ed4a2d430a9d2abadecfc1ab86",
}
)
graphql_api.should.have.key("xrayEnabled").equals(True)
assert graphql_api["logConfig"] == {
"cloudWatchLogsRoleArn": "arn:aws:cloudwatch:role",
"fieldLogLevel": "ERROR",
}
assert graphql_api["userPoolConfig"] == {
"awsRegion": "us-east-1",
"defaultAction": "DENY",
"userPoolId": "us-east-1_391729ed4a2d430a9d2abadecfc1ab86",
}
assert graphql_api["xrayEnabled"] is True
@mock_appsync
@ -130,8 +130,8 @@ def test_get_graphql_api_unknown():
client.get_graphql_api(apiId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal("GraphQL API unknown not found.")
assert err["Code"] == "NotFoundException"
assert err["Message"] == "GraphQL API unknown not found."
@mock_appsync
@ -143,22 +143,22 @@ def test_delete_graphql_api():
]["apiId"]
resp = client.list_graphql_apis()
resp.should.have.key("graphqlApis").length_of(1)
assert len(resp["graphqlApis"]) == 1
client.delete_graphql_api(apiId=api_id)
resp = client.list_graphql_apis()
resp.should.have.key("graphqlApis").length_of(0)
assert len(resp["graphqlApis"]) == 0
@mock_appsync
def test_list_graphql_apis():
client = boto3.client("appsync", region_name="ap-southeast-1")
resp = client.list_graphql_apis()
resp.should.have.key("graphqlApis").equals([])
assert resp["graphqlApis"] == []
for _ in range(3):
client.create_graphql_api(name="api1", authenticationType="API_KEY")
resp = client.list_graphql_apis()
resp.should.have.key("graphqlApis").length_of(3)
assert len(resp["graphqlApis"]) == 3

36
tests/test_appsync/test_appsync_apikeys.py
View File

@ -14,13 +14,13 @@ def test_create_api_key_simple():
]["apiId"]
resp = client.create_api_key(apiId=api_id)
resp.should.have.key("apiKey")
assert "apiKey" in resp
api_key = resp["apiKey"]
api_key.should.have.key("id")
api_key.shouldnt.have.key("description")
api_key.should.have.key("expires")
api_key.should.have.key("deletes")
assert "id" in api_key
assert "description" not in api_key
assert "expires" in api_key
assert "deletes" in api_key
@mock_appsync
@ -36,13 +36,13 @@ def test_create_api_key():
apiId=api_id, description="my first api key", expires=tomorrow_in_secs
)
resp.should.have.key("apiKey")
assert "apiKey" in resp
api_key = resp["apiKey"]
api_key.should.have.key("id")
api_key.should.have.key("description").equals("my first api key")
api_key.should.have.key("expires").equals(tomorrow_in_secs)
api_key.should.have.key("deletes").equals(tomorrow_in_secs)
assert "id" in api_key
assert api_key["description"] == "my first api key"
assert api_key["expires"] == tomorrow_in_secs
assert api_key["deletes"] == tomorrow_in_secs
@mock_appsync
@ -57,14 +57,14 @@ def test_delete_api_key():
client.delete_api_key(apiId=api_id, id=api_key_id)
resp = client.list_api_keys(apiId=api_id)
resp.should.have.key("apiKeys").length_of(0)
assert len(resp["apiKeys"]) == 0
@mock_appsync
def test_list_api_keys_unknown_api():
client = boto3.client("appsync", region_name="ap-southeast-1")
resp = client.list_api_keys(apiId="unknown")
resp.should.have.key("apiKeys").equals([])
assert resp["apiKeys"] == []
@mock_appsync
@ -75,7 +75,7 @@ def test_list_api_keys_empty():
]["apiId"]
resp = client.list_api_keys(apiId=api_id)
resp.should.have.key("apiKeys").equals([])
assert resp["apiKeys"] == []
@mock_appsync
@ -87,7 +87,7 @@ def test_list_api_keys():
client.create_api_key(apiId=api_id)
client.create_api_key(apiId=api_id, description="my first api key")
resp = client.list_api_keys(apiId=api_id)
resp.should.have.key("apiKeys").length_of(2)
assert len(resp["apiKeys"]) == 2
@mock_appsync
@ -106,7 +106,7 @@ def test_update_api_key():
apiId=api_id, id=original["id"], description="my second api key"
)["apiKey"]
updated.should.have.key("id").equals(original["id"])
updated.should.have.key("description").equals("my second api key")
updated.should.have.key("expires").equals(original["expires"])
updated.should.have.key("deletes").equals(original["deletes"])
assert updated["id"] == original["id"]
assert updated["description"] == "my second api key"
assert updated["expires"] == original["expires"]
assert updated["deletes"] == original["deletes"]

56
tests/test_appsync/test_appsync_schema.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
import json
from botocore.exceptions import ClientError
import pytest
@ -62,7 +61,7 @@ def test_start_schema_creation():
resp = client.start_schema_creation(apiId=api_id, definition=b"sth")
resp.should.have.key("status").equals("PROCESSING")
assert resp["status"] == "PROCESSING"
@mock_appsync
@ -75,8 +74,8 @@ def test_get_schema_creation_status():
client.start_schema_creation(apiId=api_id, definition=schema.encode("utf-8"))
resp = client.get_schema_creation_status(apiId=api_id)
resp.should.have.key("status").equals("SUCCESS")
resp.shouldnt.have.key("details")
assert resp["status"] == "SUCCESS"
assert "details" not in resp
@mock_appsync
@ -89,8 +88,8 @@ def test_get_schema_creation_status_invalid():
client.start_schema_creation(apiId=api_id, definition=b"sth")
resp = client.get_schema_creation_status(apiId=api_id)
resp.should.have.key("status").equals("FAILED")
resp.should.have.key("details").match("Syntax Error")
assert resp["status"] == "FAILED"
assert "Syntax Error" in resp["details"]
@mock_appsync
@ -104,17 +103,17 @@ def test_get_type_from_schema():
client.start_schema_creation(apiId=api_id, definition=schema.encode("utf-8"))
resp = client.get_type(apiId=api_id, typeName="Post", format="SDL")
resp.should.have.key("type")
assert "type" in resp
graphql_type = resp["type"]
graphql_type.should.have.key("name").equals("Post")
graphql_type.should.have.key("description").equals("My custom post type")
graphql_type.should.have.key("arn").equals("arn:aws:appsync:graphql_type/Post")
graphql_type.should.have.key("definition").equals("NotYetImplemented")
graphql_type.should.have.key("format").equals("SDL")
assert graphql_type["name"] == "Post"
assert graphql_type["description"] == "My custom post type"
assert graphql_type["arn"] == "arn:aws:appsync:graphql_type/Post"
assert graphql_type["definition"] == "NotYetImplemented"
assert graphql_type["format"] == "SDL"
query_type = client.get_type(apiId=api_id, typeName="Query", format="SDL")["type"]
query_type.should.have.key("name").equals("Query")
query_type.shouldnt.have.key("description")
assert query_type["name"] == "Query"
assert "description" not in query_type
@mock_appsync
@ -128,9 +127,9 @@ def test_get_introspection_schema_raise_gql_schema_error_if_no_schema():
with pytest.raises(ClientError) as exc:
client.get_introspection_schema(apiId=api_id, format="SDL")
err = exc.value.response["Error"]
err["Code"].should.equal("GraphQLSchemaException")
assert err["Code"] == "GraphQLSchemaException"
# AWS API appears to return InvalidSyntaxError if no schema exists
err["Message"].should.equal("InvalidSyntaxError")
assert err["Message"] == "InvalidSyntaxError"
@mock_appsync
@ -145,13 +144,12 @@ def test_get_introspection_schema_sdl():
resp = client.get_introspection_schema(apiId=api_id, format="SDL")
schema_sdl = resp["schema"].read().decode("utf-8")
schema_sdl.should.contain("putPost(")
schema_sdl.should.contain("singlePost(id: ID!): Post")
assert "putPost(" in schema_sdl
assert "singlePost(id: ID!): Post" in schema_sdl
@mock_appsync
def test_get_introspection_schema_json():
client = boto3.client("appsync", region_name="us-east-2")
api_id = client.create_graphql_api(name="api1", authenticationType="API_KEY")[
@ -162,12 +160,12 @@ def test_get_introspection_schema_json():
resp = client.get_introspection_schema(apiId=api_id, format="JSON")
schema_json = json.loads(resp["schema"].read().decode("utf-8"))
schema_json.should.have.key("__schema")
schema_json["__schema"].should.have.key("queryType")
schema_json["__schema"].should.have.key("mutationType")
schema_json["__schema"].should.have.key("subscriptionType")
schema_json["__schema"].should.have.key("types")
schema_json["__schema"].should.have.key("directives")
assert "__schema" in schema_json
assert "queryType" in schema_json["__schema"]
assert "mutationType" in schema_json["__schema"]
assert "subscriptionType" in schema_json["__schema"]
assert "types" in schema_json["__schema"]
assert "directives" in schema_json["__schema"]
@mock_appsync
@ -184,8 +182,8 @@ def test_get_introspection_schema_bad_format():
client.get_introspection_schema(apiId=api_id, format="NotAFormat")
err = exc.value.response["Error"]
err["Code"].should.equal("BadRequestException")
err["Message"].should.equal("Invalid format NotAFormat given")
assert err["Code"] == "BadRequestException"
assert err["Message"] == "Invalid format NotAFormat given"
@mock_appsync
@ -206,7 +204,7 @@ def test_get_introspection_schema_include_directives_true():
schema_sdl = resp["schema"].read().decode("utf-8")
schema_sdl.should.contain("@aws_subscribe")
assert "@aws_subscribe" in schema_sdl
@mock_appsync
@ -227,4 +225,4 @@ def test_get_introspection_schema_include_directives_false():
schema_sdl = resp["schema"].read().decode("utf-8")
schema_sdl.shouldnt.contain("@aws_subscribe")
assert "@aws_subscribe" not in schema_sdl

15
tests/test_appsync/test_appsync_tags.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_appsync
@ -11,11 +10,11 @@ def test_create_graphql_api_with_tags():
name="api1", authenticationType="API_KEY", tags={"key": "val", "key2": "val2"}
)["graphqlApi"]
api.should.have.key("tags").equals({"key": "val", "key2": "val2"})
assert api["tags"] == {"key": "val", "key2": "val2"}
api = client.get_graphql_api(apiId=api["apiId"])["graphqlApi"]
api.should.have.key("tags").equals({"key": "val", "key2": "val2"})
assert api["tags"] == {"key": "val", "key2": "val2"}
@mock_appsync
@ -28,7 +27,7 @@ def test_tag_resource():
client.tag_resource(resourceArn=(api["arn"]), tags={"key1": "val1"})
api = client.get_graphql_api(apiId=api["apiId"])["graphqlApi"]
api.should.have.key("tags").equals({"key1": "val1"})
assert api["tags"] == {"key1": "val1"}
@mock_appsync
@ -45,7 +44,7 @@ def test_tag_resource_with_existing_tags():
)
api = client.get_graphql_api(apiId=api["apiId"])["graphqlApi"]
api.should.have.key("tags").equals({"key2": "new value", "key3": "val3"})
assert api["tags"] == {"key2": "new value", "key3": "val3"}
@mock_appsync
@ -58,7 +57,7 @@ def test_untag_resource():
client.untag_resource(resourceArn=api["arn"], tagKeys=["key"])
api = client.get_graphql_api(apiId=api["apiId"])["graphqlApi"]
api.should.have.key("tags").equals({"key2": "val2"})
assert api["tags"] == {"key2": "val2"}
@mock_appsync
@ -71,7 +70,7 @@ def test_untag_resource_all():
client.untag_resource(resourceArn=api["arn"], tagKeys=["key", "key2"])
api = client.get_graphql_api(apiId=api["apiId"])["graphqlApi"]
api.should.have.key("tags").equals({})
assert api["tags"] == {}
@mock_appsync
@ -83,4 +82,4 @@ def test_list_tags_for_resource():
resp = client.list_tags_for_resource(resourceArn=api["arn"])
resp.should.have.key("tags").equals({"key": "val", "key2": "val2"})
assert resp["tags"] == {"key": "val", "key2": "val2"}

5
tests/test_appsync/test_server.py
View File

@ -1,5 +1,4 @@
import json
import sure # noqa # pylint: disable=unused-import
import moto.server as server
@ -11,5 +10,5 @@ def test_appsync_list_tags_for_resource():
resp = test_client.get(
"/v1/tags/arn%3Aaws%3Aappsync%3Aus-east-1%3A123456789012%3Aapis%2Ff405dd93-855e-451d-ab00-7325b8e439c6?tagKeys=Description"
)
resp.status_code.should.equal(200)
json.loads(resp.data).should.equals({"tags": {}})
assert resp.status_code == 200
assert json.loads(resp.data) == {"tags": {}}