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

Techdebt: Replace sure with regular assertions in Neptune (#6663)

Browse Source
This commit is contained in:
Bert Blommers 2023-08-15 08:00:12 +00:00 committed by GitHub
parent dc32eb86a3
commit 8035269645
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 51 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
tests/test_neptune/test_cluster_options.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_neptune
@ -9,14 +8,14 @@ def test_db_cluster_options():
# We're not checking the exact data here, that is already done in TF
client = boto3.client("neptune", region_name="us-east-1")
response = client.describe_orderable_db_instance_options(Engine="neptune")
response["OrderableDBInstanceOptions"].should.have.length_of(286)
assert len(response["OrderableDBInstanceOptions"]) == 286
response = client.describe_orderable_db_instance_options(
Engine="neptune", EngineVersion="1.0.2.1"
)
response["OrderableDBInstanceOptions"].should.have.length_of(0)
assert len(response["OrderableDBInstanceOptions"]) == 0
response = client.describe_orderable_db_instance_options(
Engine="neptune", EngineVersion="1.0.3.0"
)
response["OrderableDBInstanceOptions"].should.have.length_of(12)
assert len(response["OrderableDBInstanceOptions"]) == 12

4
tests/test_neptune/test_cluster_tags.py
View File

@ -17,9 +17,9 @@ def test_add_tags_to_cluster():
)
tags = conn.list_tags_for_resource(ResourceName=cluster_arn)["TagList"]
tags.should.equal([{"Key": "k1", "Value": "v1"}, {"Key": "k2", "Value": "v2"}])
assert tags == [{"Key": "k1", "Value": "v1"}, {"Key": "k2", "Value": "v2"}]
conn.remove_tags_from_resource(ResourceName=cluster_arn, TagKeys=["k1"])
tags = conn.list_tags_for_resource(ResourceName=cluster_arn)["TagList"]
tags.should.equal([{"Key": "k2", "Value": "v2"}])
assert tags == [{"Key": "k2", "Value": "v2"}]

64
tests/test_neptune/test_clusters.py
View File

@ -1,7 +1,6 @@
"""Unit tests for neptune-supported APIs."""
import boto3
import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError
from moto import mock_neptune
@ -15,23 +14,21 @@ def test_create_db_cluster():
resp = client.create_db_cluster(DBClusterIdentifier="cluster-id", Engine="neptune")[
"DBCluster"
]
resp.should.have.key("DBClusterIdentifier").equals("cluster-id")
resp.should.have.key("DbClusterResourceId")
resp.should.have.key("DBClusterArn")
resp.should.have.key("Engine").equals("neptune")
resp.should.have.key("EngineVersion").equals("1.2.0.2")
resp.should.have.key("StorageEncrypted").equals(True)
resp.should.have.key("DBClusterParameterGroup").equals("")
resp.should.have.key("Endpoint")
resp.should.have.key("DbClusterResourceId").match("cluster-")
resp.should.have.key("AvailabilityZones").equals(
["us-east-2a", "us-east-2b", "us-east-2c"]
)
resp.shouldnt.have.key("ServerlessV2ScalingConfiguration")
assert resp["DBClusterIdentifier"] == "cluster-id"
assert "DbClusterResourceId" in resp
assert "DBClusterArn" in resp
assert resp["Engine"] == "neptune"
assert resp["EngineVersion"] == "1.2.0.2"
assert resp["StorageEncrypted"] is True
assert resp["DBClusterParameterGroup"] == ""
assert "Endpoint" in resp
assert "cluster-" in resp["DbClusterResourceId"]
assert resp["AvailabilityZones"] == ["us-east-2a", "us-east-2b", "us-east-2c"]
assert "ServerlessV2ScalingConfiguration" not in resp
# Double check this cluster is not available in another region
europe_client = boto3.client("neptune", region_name="eu-west-2")
europe_client.describe_db_clusters()["DBClusters"].should.have.length_of(0)
assert len(europe_client.describe_db_clusters()["DBClusters"]) == 0
@mock_neptune
@ -47,29 +44,30 @@ def test_create_db_cluster__with_additional_params():
ServerlessV2ScalingConfiguration={"MinCapacity": 1.0, "MaxCapacity": 2.0},
DatabaseName="sth",
)["DBCluster"]
resp.should.have.key("StorageEncrypted").equals(False)
resp.should.have.key("DBClusterParameterGroup").equals("myprm")
resp.should.have.key("EngineVersion").equals("1.1.0.1")
resp.should.have.key("KmsKeyId").equals("key")
resp.should.have.key("ServerlessV2ScalingConfiguration").equals(
{"MinCapacity": 1.0, "MaxCapacity": 2.0}
)
resp.should.have.key("DatabaseName").equals("sth")
assert resp["StorageEncrypted"] is False
assert resp["DBClusterParameterGroup"] == "myprm"
assert resp["EngineVersion"] == "1.1.0.1"
assert resp["KmsKeyId"] == "key"
assert resp["ServerlessV2ScalingConfiguration"] == {
"MinCapacity": 1.0,
"MaxCapacity": 2.0,
}
assert resp["DatabaseName"] == "sth"
@mock_neptune
def test_describe_db_clusters():
client = boto3.client("neptune", region_name="ap-southeast-1")
client.describe_db_clusters()["DBClusters"].should.equal([])
assert client.describe_db_clusters()["DBClusters"] == []
client.create_db_cluster(DBClusterIdentifier="cluster-id", Engine="neptune")
clusters = client.describe_db_clusters(DBClusterIdentifier="cluster-id")[
"DBClusters"
]
clusters.should.have.length_of(1)
clusters[0]["DBClusterIdentifier"].should.equal("cluster-id")
clusters[0].should.have.key("Engine").equals("neptune")
assert len(clusters) == 1
assert clusters[0]["DBClusterIdentifier"] == "cluster-id"
assert clusters[0]["Engine"] == "neptune"
@mock_neptune
@ -79,7 +77,7 @@ def test_delete_db_cluster():
client.create_db_cluster(DBClusterIdentifier="cluster-id", Engine="neptune")
client.delete_db_cluster(DBClusterIdentifier="cluster-id")
client.describe_db_clusters()["DBClusters"].should.equal([])
assert client.describe_db_clusters()["DBClusters"] == []
@mock_neptune
@ -89,7 +87,7 @@ def test_delete_unknown_db_cluster():
with pytest.raises(ClientError) as exc:
client.delete_db_cluster(DBClusterIdentifier="unknown-id")
err = exc.value.response["Error"]
err["Code"].should.equal("DBClusterNotFoundFault")
assert err["Code"] == "DBClusterNotFoundFault"
@mock_neptune
@ -102,9 +100,9 @@ def test_modify_db_cluster():
DBClusterParameterGroupName="myprm",
PreferredBackupWindow="window",
)["DBCluster"]
resp.should.have.key("DBClusterParameterGroup").equals("myprm")
resp.should.have.key("EngineVersion").equals("1.1.0.1")
resp.should.have.key("PreferredBackupWindow").equals("window")
assert resp["DBClusterParameterGroup"] == "myprm"
assert resp["EngineVersion"] == "1.1.0.1"
assert resp["PreferredBackupWindow"] == "window"
@mock_neptune
@ -115,4 +113,4 @@ def test_start_db_cluster():
]
cluster = client.start_db_cluster(DBClusterIdentifier="cluster-id")["DBCluster"]
cluster.should.have.key("Status").equals("started")
assert cluster["Status"] == "started"

31
tests/test_neptune/test_global_clusters.py
View File

@ -1,12 +1,11 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_neptune
@mock_neptune
def test_describe():
client = boto3.client("neptune", "us-east-2")
client.describe_global_clusters()["GlobalClusters"].should.equal([])
assert client.describe_global_clusters()["GlobalClusters"] == []
@mock_neptune
@ -15,19 +14,19 @@ def test_create_global_cluster():
resp = client.create_global_cluster(
GlobalClusterIdentifier="g-id", Engine="neptune"
)["GlobalCluster"]
resp.should.have.key("GlobalClusterIdentifier").equals("g-id")
resp.should.have.key("GlobalClusterResourceId")
resp.should.have.key("GlobalClusterArn")
resp.should.have.key("Engine").equals("neptune")
resp.should.have.key("EngineVersion").equals("1.2.0.0")
resp.should.have.key("StorageEncrypted").equals(False)
resp.should.have.key("DeletionProtection").equals(False)
assert resp["GlobalClusterIdentifier"] == "g-id"
assert "GlobalClusterResourceId" in resp
assert "GlobalClusterArn" in resp
assert resp["Engine"] == "neptune"
assert resp["EngineVersion"] == "1.2.0.0"
assert resp["StorageEncrypted"] is False
assert resp["DeletionProtection"] is False
client.describe_global_clusters()["GlobalClusters"].should.have.length_of(1)
assert len(client.describe_global_clusters()["GlobalClusters"]) == 1
# As a global cluster, verify it can be retrieved everywhere
europe_client = boto3.client("neptune", "eu-north-1")
europe_client.describe_global_clusters()["GlobalClusters"].should.have.length_of(1)
assert len(europe_client.describe_global_clusters()["GlobalClusters"]) == 1
@mock_neptune
@ -40,10 +39,10 @@ def test_create_global_cluster_with_additional_params():
DeletionProtection=True,
StorageEncrypted=True,
)["GlobalCluster"]
resp.should.have.key("Engine").equals("neptune")
resp.should.have.key("EngineVersion").equals("1.0")
resp.should.have.key("StorageEncrypted").equals(True)
resp.should.have.key("DeletionProtection").equals(True)
assert resp["Engine"] == "neptune"
assert resp["EngineVersion"] == "1.0"
assert resp["StorageEncrypted"] is True
assert resp["DeletionProtection"] is True
@mock_neptune
@ -53,4 +52,4 @@ def test_delete_global_cluster():
client.delete_global_cluster(GlobalClusterIdentifier="g-id2")
client.describe_global_clusters()["GlobalClusters"].should.equal([])
assert client.describe_global_clusters()["GlobalClusters"] == []