Techdebt: Replace sure with regular asserts in AMP tests (#6364)

This commit is contained in:
Bert Blommers 2023-06-04 09:50:10 +00:00 committed by GitHub
parent 6cd9aae100
commit 073b494113
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 84 additions and 84 deletions

View File

@ -13,21 +13,21 @@ class TestAmpLoggingConfig(unittest.TestCase):
def test_describe_logging(self): def test_describe_logging(self):
resp = self.client.describe_logging_configuration(workspaceId=self.workspace_id) resp = self.client.describe_logging_configuration(workspaceId=self.workspace_id)
resp.should.have.key("loggingConfiguration").equals({}) assert resp["loggingConfiguration"] == {}
def test_create_logging(self): def test_create_logging(self):
resp = self.client.create_logging_configuration( resp = self.client.create_logging_configuration(
workspaceId=self.workspace_id, logGroupArn="log/arn" workspaceId=self.workspace_id, logGroupArn="log/arn"
) )
resp.should.have.key("status").equals({"statusCode": "ACTIVE"}) assert resp["status"] == {"statusCode": "ACTIVE"}
resp = self.client.describe_logging_configuration( resp = self.client.describe_logging_configuration(
workspaceId=self.workspace_id workspaceId=self.workspace_id
)["loggingConfiguration"] )["loggingConfiguration"]
resp.should.have.key("createdAt") assert "createdAt" in resp
resp.should.have.key("logGroupArn").equals("log/arn") assert resp["logGroupArn"] == "log/arn"
resp.should.have.key("status").equals({"statusCode": "ACTIVE"}) assert resp["status"] == {"statusCode": "ACTIVE"}
resp.should.have.key("workspace").equals(self.workspace_id) assert resp["workspace"] == self.workspace_id
def test_update_logging(self): def test_update_logging(self):
self.client.create_logging_configuration( self.client.create_logging_configuration(
@ -37,13 +37,13 @@ class TestAmpLoggingConfig(unittest.TestCase):
resp = self.client.update_logging_configuration( resp = self.client.update_logging_configuration(
workspaceId=self.workspace_id, logGroupArn="log/arn2" workspaceId=self.workspace_id, logGroupArn="log/arn2"
) )
resp.should.have.key("status").equals({"statusCode": "ACTIVE"}) assert resp["status"] == {"statusCode": "ACTIVE"}
resp = self.client.describe_logging_configuration( resp = self.client.describe_logging_configuration(
workspaceId=self.workspace_id workspaceId=self.workspace_id
)["loggingConfiguration"] )["loggingConfiguration"]
resp.should.have.key("modifiedAt") assert "modifiedAt" in resp
resp.should.have.key("logGroupArn").equals("log/arn2") assert resp["logGroupArn"] == "log/arn2"
def test_delete_logging(self): def test_delete_logging(self):
resp = self.client.create_logging_configuration( resp = self.client.create_logging_configuration(
@ -53,4 +53,4 @@ class TestAmpLoggingConfig(unittest.TestCase):
self.client.delete_logging_configuration(workspaceId=self.workspace_id) self.client.delete_logging_configuration(workspaceId=self.workspace_id)
resp = self.client.describe_logging_configuration(workspaceId=self.workspace_id) resp = self.client.describe_logging_configuration(workspaceId=self.workspace_id)
resp.should.have.key("loggingConfiguration").equals({}) assert resp["loggingConfiguration"] == {}

View File

@ -1,7 +1,6 @@
"""Unit tests for amp-supported APIs.""" """Unit tests for amp-supported APIs."""
import boto3 import boto3
import pytest import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from moto import mock_amp from moto import mock_amp
@ -18,9 +17,9 @@ def test_create_rule_groups_namespace():
data=b"asdf", name="my first rule group", workspaceId=workspace_id data=b"asdf", name="my first rule group", workspaceId=workspace_id
) )
resp.should.have.key("arn") assert "arn" in resp
resp.should.have.key("name").equals("my first rule group") assert resp["name"] == "my first rule group"
resp.should.have.key("status") assert "status" in resp
@mock_amp @mock_amp
@ -36,8 +35,8 @@ def test_delete_rule_groups_namespace():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
client.describe_rule_groups_namespace(name="myname", workspaceId=workspace_id) client.describe_rule_groups_namespace(name="myname", workspaceId=workspace_id)
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ResourceNotFoundException") assert err["Code"] == "ResourceNotFoundException"
err["Message"].should.equal("RuleGroupNamespace not found") assert err["Message"] == "RuleGroupNamespace not found"
@mock_amp @mock_amp
@ -52,15 +51,15 @@ def test_describe_rule_groups_namespace():
resp = client.describe_rule_groups_namespace( resp = client.describe_rule_groups_namespace(
name="myname", workspaceId=workspace_id name="myname", workspaceId=workspace_id
) )
resp.should.have.key("ruleGroupsNamespace") assert "ruleGroupsNamespace" in resp
ns = resp["ruleGroupsNamespace"] ns = resp["ruleGroupsNamespace"]
ns.should.have.key("arn") assert "arn" in ns
ns.should.have.key("createdAt") assert "createdAt" in ns
ns.should.have.key("data").equals(b"asdf") assert ns["data"] == b"asdf"
ns.should.have.key("modifiedAt") assert "modifiedAt" in ns
ns.should.have.key("name").equals("myname") assert ns["name"] == "myname"
ns.should.have.key("status") assert "status" in ns
@mock_amp @mock_amp
@ -79,12 +78,12 @@ def test_put_rule_groups_namespace():
resp = client.describe_rule_groups_namespace( resp = client.describe_rule_groups_namespace(
name="myname", workspaceId=workspace_id name="myname", workspaceId=workspace_id
) )
resp.should.have.key("ruleGroupsNamespace") assert "ruleGroupsNamespace" in resp
ns = resp["ruleGroupsNamespace"] ns = resp["ruleGroupsNamespace"]
ns.should.have.key("arn") assert "arn" in ns
ns.should.have.key("createdAt") assert "createdAt" in ns
ns.should.have.key("data").equals(b"updated") assert ns["data"] == b"updated"
@mock_amp @mock_amp
@ -97,18 +96,18 @@ def test_list_rule_groups_namespaces():
) )
resp = client.list_rule_groups_namespaces(workspaceId=w_id) resp = client.list_rule_groups_namespaces(workspaceId=w_id)
resp.should.have.key("ruleGroupsNamespaces").length_of(15) assert len(resp["ruleGroupsNamespaces"]) == 15
resp.shouldnt.have.key("nextToken") assert "nextToken" not in resp
resp = client.list_rule_groups_namespaces(workspaceId=w_id, name="ns1") resp = client.list_rule_groups_namespaces(workspaceId=w_id, name="ns1")
resp.should.have.key("ruleGroupsNamespaces").length_of(6) assert len(resp["ruleGroupsNamespaces"]) == 6
names = [ns["name"] for ns in resp["ruleGroupsNamespaces"]] names = [ns["name"] for ns in resp["ruleGroupsNamespaces"]]
set(names).should.equal({"ns10", "ns13", "ns1", "ns12", "ns11", "ns14"}) assert set(names) == {"ns10", "ns13", "ns1", "ns12", "ns11", "ns14"}
resp = client.list_rule_groups_namespaces(workspaceId=w_id, name="ns10") resp = client.list_rule_groups_namespaces(workspaceId=w_id, name="ns10")
resp.should.have.key("ruleGroupsNamespaces").length_of(1) assert len(resp["ruleGroupsNamespaces"]) == 1
names = [ns["name"] for ns in resp["ruleGroupsNamespaces"]] names = [ns["name"] for ns in resp["ruleGroupsNamespaces"]]
set(names).should.equal({"ns10"}) assert set(names) == {"ns10"}
@mock_amp @mock_amp
@ -122,26 +121,26 @@ def test_list_rule_groups_namespaces__paginated():
# default pagesize is 100 # default pagesize is 100
page1 = client.list_rule_groups_namespaces(workspaceId=w_id) page1 = client.list_rule_groups_namespaces(workspaceId=w_id)
page1.should.have.key("ruleGroupsNamespaces").length_of(100) assert len(page1["ruleGroupsNamespaces"]) == 100
page1.should.have.key("nextToken") assert "nextToken" in page1
# We can ask for a smaller pagesize # We can ask for a smaller pagesize
page2 = client.list_rule_groups_namespaces( page2 = client.list_rule_groups_namespaces(
workspaceId=w_id, maxResults=15, nextToken=page1["nextToken"] workspaceId=w_id, maxResults=15, nextToken=page1["nextToken"]
) )
page2.should.have.key("ruleGroupsNamespaces").length_of(15) assert len(page2["ruleGroupsNamespaces"]) == 15
page2.should.have.key("nextToken") assert "nextToken" in page2
page3 = client.list_rule_groups_namespaces( page3 = client.list_rule_groups_namespaces(
workspaceId=w_id, maxResults=15, nextToken=page2["nextToken"] workspaceId=w_id, maxResults=15, nextToken=page2["nextToken"]
) )
page3.should.have.key("ruleGroupsNamespaces").length_of(10) assert len(page3["ruleGroupsNamespaces"]) == 10
page3.shouldnt.have.key("nextToken") assert "nextToken" not in page3
# We could request all of them in one go # We could request all of them in one go
full_page = client.list_rule_groups_namespaces(workspaceId=w_id, maxResults=150) full_page = client.list_rule_groups_namespaces(workspaceId=w_id, maxResults=150)
full_page.should.have.key("ruleGroupsNamespaces").length_of(125) assert len(full_page["ruleGroupsNamespaces"]) == 125
full_page.shouldnt.have.key("nextToken") assert "nextToken" not in full_page
@mock_amp @mock_amp
@ -157,17 +156,18 @@ def test_tag_resource():
client.tag_resource(resourceArn=arn, tags={"t1": "v1", "t2": "v2"}) client.tag_resource(resourceArn=arn, tags={"t1": "v1", "t2": "v2"})
client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal( assert get_tags(arn, client) == {"t": "v", "t1": "v1", "t2": "v2"}
{"t": "v", "t1": "v1", "t2": "v2"} ns = client.describe_rule_groups_namespace(workspaceId=w_id, name="ns")[
)
client.describe_rule_groups_namespace(workspaceId=w_id, name="ns")[
"ruleGroupsNamespace" "ruleGroupsNamespace"
]["tags"].should.equal({"t": "v", "t1": "v1", "t2": "v2"}) ]
assert ns["tags"] == {"t": "v", "t1": "v1", "t2": "v2"}
client.untag_resource(resourceArn=arn, tagKeys=["t1"]) client.untag_resource(resourceArn=arn, tagKeys=["t1"])
client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal( assert get_tags(arn, client) == {"t": "v", "t2": "v2"}
{"t": "v", "t2": "v2"}
)
client.untag_resource(resourceArn=arn, tagKeys=["t", "t2"]) client.untag_resource(resourceArn=arn, tagKeys=["t", "t2"])
client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal({}) assert get_tags(arn, client) == {}
def get_tags(arn, client):
return client.list_tags_for_resource(resourceArn=arn)["tags"]

View File

@ -1,6 +1,5 @@
import boto3 import boto3
import pytest import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from moto import mock_amp from moto import mock_amp
@ -15,9 +14,9 @@ def test_create_workspace():
client = boto3.client("amp", region_name="ap-southeast-1") client = boto3.client("amp", region_name="ap-southeast-1")
resp = client.create_workspace(alias="test", clientToken="mytoken") resp = client.create_workspace(alias="test", clientToken="mytoken")
resp.should.have.key("arn") assert "arn" in resp
resp.should.have.key("status").equals({"statusCode": "ACTIVE"}) assert resp["status"] == {"statusCode": "ACTIVE"}
resp.should.have.key("workspaceId") assert "workspaceId" in resp
@mock_amp @mock_amp
@ -28,15 +27,15 @@ def test_describe_workspace():
] ]
resp = client.describe_workspace(workspaceId=workspace_id) resp = client.describe_workspace(workspaceId=workspace_id)
resp.should.have.key("workspace") assert "workspace" in resp
workspace = resp["workspace"] workspace = resp["workspace"]
workspace.should.have.key("alias") assert "alias" in workspace
workspace.should.have.key("arn") assert "arn" in workspace
workspace.should.have.key("createdAt") assert "createdAt" in workspace
workspace.should.have.key("prometheusEndpoint") assert "prometheusEndpoint" in workspace
workspace.should.have.key("status").equals({"statusCode": "ACTIVE"}) assert workspace["status"] == {"statusCode": "ACTIVE"}
workspace.should.have.key("workspaceId").equals(workspace_id) assert workspace["workspaceId"] == workspace_id
@mock_amp @mock_amp
@ -48,12 +47,12 @@ def test_list_workspaces():
spaces = client.list_workspaces(maxResults=1000)["workspaces"] spaces = client.list_workspaces(maxResults=1000)["workspaces"]
assert len(spaces) >= 2 assert len(spaces) >= 2
[sp.get("alias") for sp in spaces].should.contain("test") assert "test" in [sp.get("alias") for sp in spaces]
[sp.get("alias") for sp in spaces].should.contain(my_alias) assert my_alias in [sp.get("alias") for sp in spaces]
resp = client.list_workspaces(alias=my_alias) resp = client.list_workspaces(alias=my_alias)
resp.should.have.key("workspaces").length_of(1) assert len(resp["workspaces"]) == 1
resp["workspaces"][0].should.have.key("alias").equals(my_alias) assert resp["workspaces"][0]["alias"] == my_alias
@mock_amp @mock_amp
@ -64,13 +63,13 @@ def test_list_workspaces__paginated():
# default pagesize is 100 # default pagesize is 100
page1 = client.list_workspaces() page1 = client.list_workspaces()
page1.should.have.key("workspaces").length_of(100) assert len(page1["workspaces"]) == 100
page1.should.have.key("nextToken") assert "nextToken" in page1
# We can ask for a smaller pagesize # We can ask for a smaller pagesize
page2 = client.list_workspaces(maxResults=15, nextToken=page1["nextToken"]) page2 = client.list_workspaces(maxResults=15, nextToken=page1["nextToken"])
page2.should.have.key("workspaces").length_of(15) assert len(page2["workspaces"]) == 15
page2.should.have.key("nextToken") assert "nextToken" in page2
# We could request all of them in one go # We could request all of them in one go
all_workspaces = client.list_workspaces(maxResults=1000)["workspaces"] all_workspaces = client.list_workspaces(maxResults=1000)["workspaces"]
@ -86,8 +85,7 @@ def test_list_tags_for_resource():
alias="test", clientToken="mytoken", tags={"t1": "v1", "t2": "v2"} alias="test", clientToken="mytoken", tags={"t1": "v1", "t2": "v2"}
)["arn"] )["arn"]
resp = client.list_tags_for_resource(resourceArn=arn) assert get_tags(arn, client) == {"t1": "v1", "t2": "v2"}
resp.should.have.key("tags").equals({"t1": "v1", "t2": "v2"})
@mock_amp @mock_amp
@ -97,12 +95,12 @@ def test_update_workspace_alias():
workspace_id = client.create_workspace(alias="initial")["workspaceId"] workspace_id = client.create_workspace(alias="initial")["workspaceId"]
w = client.describe_workspace(workspaceId=workspace_id)["workspace"] w = client.describe_workspace(workspaceId=workspace_id)["workspace"]
w.should.have.key("alias").equals("initial") assert w["alias"] == "initial"
client.update_workspace_alias(alias="updated", workspaceId=workspace_id) client.update_workspace_alias(alias="updated", workspaceId=workspace_id)
w = client.describe_workspace(workspaceId=workspace_id)["workspace"] w = client.describe_workspace(workspaceId=workspace_id)["workspace"]
w.should.have.key("alias").equals("updated") assert w["alias"] == "updated"
@mock_amp @mock_amp
@ -118,8 +116,8 @@ def test_delete_workspace():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
client.describe_workspace(workspaceId=workspace_id) client.describe_workspace(workspaceId=workspace_id)
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ResourceNotFoundException") assert err["Code"] == "ResourceNotFoundException"
err["Message"].should.equal("Workspace not found") assert err["Message"] == "Workspace not found"
@mock_amp @mock_amp
@ -132,17 +130,19 @@ def test_tag_resource():
client.tag_resource(resourceArn=arn, tags={"t1": "v1", "t2": "v2"}) client.tag_resource(resourceArn=arn, tags={"t1": "v1", "t2": "v2"})
client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal( expected = {"t": "v", "t1": "v1", "t2": "v2"}
{"t": "v", "t1": "v1", "t2": "v2"} assert get_tags(arn, client) == expected
assert (
client.describe_workspace(workspaceId=workspace_id)["workspace"]["tags"]
== expected
) )
client.describe_workspace(workspaceId=workspace_id)["workspace"][
"tags"
].should.equal({"t": "v", "t1": "v1", "t2": "v2"})
client.untag_resource(resourceArn=arn, tagKeys=["t1"]) client.untag_resource(resourceArn=arn, tagKeys=["t1"])
client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal( assert get_tags(arn, client) == {"t": "v", "t2": "v2"}
{"t": "v", "t2": "v2"}
)
client.untag_resource(resourceArn=arn, tagKeys=["t", "t2"]) client.untag_resource(resourceArn=arn, tagKeys=["t", "t2"])
client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal({}) assert get_tags(arn, client) == {}
def get_tags(arn, client):
return client.list_tags_for_resource(resourceArn=arn)["tags"]