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

Tech debt - improve test coverage EC2/ELBv2/Events (#4181)

Browse Source
This commit is contained in:
Bert Blommers 2021-08-16 15:13:50 +01:00 committed by GitHub
parent 4df099c724
commit 547a1844ed
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 465 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
moto/ec2/models.py
View File

@ -4535,8 +4535,9 @@ class ManagedPrefixListBackend(object):
managed_pl = self.managed_prefix_lists.get(prefix_list_id)
managed_pl.prefix_list_name = prefix_list_name
if remove_entry or add_entry:
latest_version = managed_pl.entries.get(managed_pl.version)
entries = (
managed_pl.entries.get(current_version, managed_pl.version).copy()
managed_pl.entries.get(current_version, latest_version).copy()
if managed_pl.entries
else []
)

15
moto/events/models.py
View File

@ -267,14 +267,6 @@ class Rule(CloudFormationModel):
new_resource_name, cloudformation_json, region_name
)
@classmethod
def delete_from_cloudformation_json(
cls, resource_name, cloudformation_json, region_name
):
event_backend = events_backends[region_name]
event_name = resource_name
event_backend.delete_rule(name=event_name)
def describe(self):
attributes = {
"Arn": self.arn,
@ -578,13 +570,6 @@ class Archive(CloudFormationModel):
new_resource_name, cloudformation_json, region_name
)
@classmethod
def delete_from_cloudformation_json(
cls, resource_name, cloudformation_json, region_name
):
event_backend = events_backends[region_name]
event_backend.delete_archive(resource_name)
@unique
class ReplayState(Enum):

299
tests/test_ec2/test_prefix_lists.py Normal file
View File

@ -0,0 +1,299 @@
import boto3
import sure # noqa
from moto import mock_ec2
from moto.core import ACCOUNT_ID
@mock_ec2
def test_create():
ec2 = boto3.client("ec2", region_name="us-west-1")
prefix_list = ec2.create_managed_prefix_list(
PrefixListName="examplelist", MaxEntries=2, AddressFamily="?"
)["PrefixList"]
prefix_list.should.have.key("PrefixListId").match("pl-[a-z0-9]+")
prefix_list.should.have.key("AddressFamily").equals("?")
prefix_list.should.have.key("State").equals("create-complete")
prefix_list.should.have.key("PrefixListArn").equals(
"arn:aws:ec2:us-west-1:{}:prefix-list/{}".format(
ACCOUNT_ID, prefix_list["PrefixListId"]
)
)
prefix_list.should.have.key("PrefixListName").equals("examplelist")
prefix_list.should.have.key("MaxEntries").equals(2)
prefix_list.should.have.key("Version").equals(1)
prefix_list.should.have.key("Tags").equals([])
prefix_list.should.have.key("OwnerId").equals(ACCOUNT_ID)
@mock_ec2
def test_create_with_tags():
ec2 = boto3.client("ec2", region_name="us-west-1")
prefix_list = ec2.create_managed_prefix_list(
PrefixListName="examplelist",
MaxEntries=2,
AddressFamily="?",
TagSpecifications=[
{"ResourceType": "", "Tags": [{"Key": "key1", "Value": "val1"}]}
],
)["PrefixList"]
prefix_list.should.have.key("PrefixListId").match("pl-[a-z0-9]+")
prefix_list.should.have.key("State").equals("create-complete")
prefix_list.should.have.key("Version").equals(1)
prefix_list.should.have.key("Tags").equals([{"Key": "key1", "Value": "val1"}])
@mock_ec2
def test_describe_managed_prefix_lists():
ec2 = boto3.client("ec2", region_name="us-west-1")
default_lists = ec2.describe_managed_prefix_lists()["PrefixLists"]
set([l["OwnerId"] for l in default_lists]).should.equal({"aws"})
ec2.create_managed_prefix_list(
PrefixListName="examplelist", MaxEntries=2, AddressFamily="?"
)
all_lists = ec2.describe_managed_prefix_lists()["PrefixLists"]
all_lists.should.have.length_of(len(default_lists) + 1)
set([l["OwnerId"] for l in all_lists]).should.equal({"aws", ACCOUNT_ID})
@mock_ec2
def test_describe_managed_prefix_lists_with_prefix():
ec2 = boto3.client("ec2", region_name="us-west-1")
default_lists = ec2.describe_managed_prefix_lists()["PrefixLists"]
set([l["OwnerId"] for l in default_lists]).should.equal({"aws"})
random_list_id = default_lists[0]["PrefixListId"]
lists_by_id = ec2.describe_managed_prefix_lists(PrefixListIds=[random_list_id])[
"PrefixLists"
]
lists_by_id.should.have.length_of(1)
lists_by_id[0]["OwnerId"].should.equal("aws")
@mock_ec2
def test_get_managed_prefix_list_entries():
ec2 = boto3.client("ec2", region_name="us-west-1")
resp = ec2.create_managed_prefix_list(
PrefixListName="examplelist",
MaxEntries=2,
AddressFamily="?",
Entries=[
{"Cidr": "10.0.0.1", "Description": "entry1"},
{"Cidr": "10.0.0.2", "Description": "entry2"},
],
)
ec2.create_managed_prefix_list(
PrefixListName="examplelist2",
MaxEntries=2,
AddressFamily="?",
Entries=[
{"Cidr": "10.0.0.3", "Description": "entry3"},
{"Cidr": "10.0.0.4", "Description": "entry4"},
],
)
prefix_list_id = resp["PrefixList"]["PrefixListId"]
entries = ec2.get_managed_prefix_list_entries(PrefixListId=prefix_list_id)[
"Entries"
]
entries.should.have.length_of(2)
entries.should.contain({"Cidr": "10.0.0.1", "Description": "entry1"})
entries.should.contain({"Cidr": "10.0.0.2", "Description": "entry2"})
@mock_ec2
def test_get_managed_prefix_list_entries_0_entries():
ec2 = boto3.client("ec2", region_name="us-west-1")
resp = ec2.create_managed_prefix_list(
PrefixListName="examplelist", MaxEntries=2, AddressFamily="?"
)
prefix_list_id = resp["PrefixList"]["PrefixListId"]
entries = ec2.get_managed_prefix_list_entries(PrefixListId=prefix_list_id)[
"Entries"
]
entries.should.equal([])
@mock_ec2
def test_delete_managed_prefix_list():
ec2 = boto3.client("ec2", region_name="us-west-1")
id1 = ec2.create_managed_prefix_list(
PrefixListName="examplelist1", MaxEntries=2, AddressFamily="?"
)["PrefixList"]["PrefixListId"]
id2 = ec2.create_managed_prefix_list(
PrefixListName="examplelist2", MaxEntries=2, AddressFamily="?"
)["PrefixList"]["PrefixListId"]
lists_by_id = ec2.describe_managed_prefix_lists(PrefixListIds=[id1, id2])[
"PrefixLists"
]
lists_by_id.should.have.length_of(2)
ec2.delete_managed_prefix_list(PrefixListId=id1)
lists_by_id = ec2.describe_managed_prefix_lists(PrefixListIds=[id1, id2])[
"PrefixLists"
]
lists_by_id.should.have.length_of(2)
set([l["State"] for l in lists_by_id]).should.equal(
{"create-complete", "delete-complete"}
)
@mock_ec2
def test_describe_prefix_lists():
ec2 = boto3.client("ec2", region_name="us-west-1")
default_lists = ec2.describe_prefix_lists()["PrefixLists"]
default_lists.should.have.length_of(2)
ec2.create_managed_prefix_list(
PrefixListName="examplelist", MaxEntries=2, AddressFamily="?"
)
all_lists = ec2.describe_prefix_lists()["PrefixLists"]
all_lists.should.have.length_of(2)
for l in all_lists:
l["PrefixListName"].should.contain("com.amazonaws")
@mock_ec2
def test_modify_manage_prefix_list():
ec2 = boto3.client("ec2", region_name="us-west-1")
resp = ec2.create_managed_prefix_list(
PrefixListName="examplelist",
MaxEntries=2,
AddressFamily="?",
Entries=[
{"Cidr": "10.0.0.1", "Description": "entry1"},
{"Cidr": "10.0.0.2", "Description": "entry2"},
],
)
prefix_list_id = resp["PrefixList"]["PrefixListId"]
prefix_list = ec2.modify_managed_prefix_list(
PrefixListId=prefix_list_id,
AddEntries=[{"Cidr": "10.0.0.3", "Description": "entry3"}],
RemoveEntries=[{"Cidr": "10.0.0.2"}],
)["PrefixList"]
prefix_list["PrefixListId"].should.equal(prefix_list_id)
prefix_list["State"].should.equal("modify-complete")
prefix_list["Version"].should.equal(2)
described = ec2.describe_managed_prefix_lists(PrefixListIds=[prefix_list_id])[
"PrefixLists"
][0]
described.should.equal(prefix_list)
entries = ec2.get_managed_prefix_list_entries(PrefixListId=prefix_list_id)[
"Entries"
]
entries.should.have.length_of(2)
entries.should.contain({"Cidr": "10.0.0.1", "Description": "entry1"})
entries.should.contain({"Cidr": "10.0.0.3", "Description": "entry3"})
entries.shouldnt.contain({"Cidr": "10.0.0.2", "Description": "entry2"})
@mock_ec2
def test_modify_manage_prefix_list_add_to_empty_list():
ec2 = boto3.client("ec2", region_name="us-west-1")
resp = ec2.create_managed_prefix_list(
PrefixListName="examplelist", MaxEntries=2, AddressFamily="?"
)
prefix_list_id = resp["PrefixList"]["PrefixListId"]
prefix_list = ec2.modify_managed_prefix_list(
PrefixListId=prefix_list_id,
AddEntries=[{"Cidr": "10.0.0.3", "Description": "entry3"}],
RemoveEntries=[{"Cidr": "10.0.0.2"}],
)["PrefixList"]
prefix_list["PrefixListId"].should.equal(prefix_list_id)
prefix_list["State"].should.equal("modify-complete")
prefix_list["Version"].should.equal(2)
described = ec2.describe_managed_prefix_lists(PrefixListIds=[prefix_list_id])[
"PrefixLists"
][0]
described.should.equal(prefix_list)
entries = ec2.get_managed_prefix_list_entries(PrefixListId=prefix_list_id)[
"Entries"
]
entries.should.have.length_of(1)
entries.should.contain({"Cidr": "10.0.0.3", "Description": "entry3"})
entries.shouldnt.contain({"Cidr": "10.0.0.2", "Description": "entry2"})
@mock_ec2
def test_modify_manage_prefix_list_name_only():
ec2 = boto3.client("ec2", region_name="us-west-1")
resp = ec2.create_managed_prefix_list(
PrefixListName="examplelist", MaxEntries=2, AddressFamily="?"
)
prefix_list_id = resp["PrefixList"]["PrefixListId"]
prefix_list = ec2.modify_managed_prefix_list(
PrefixListId=prefix_list_id, PrefixListName="new name"
)["PrefixList"]
prefix_list["PrefixListId"].should.equal(prefix_list_id)
prefix_list["PrefixListName"].should.equal("new name")
prefix_list["State"].should.equal("modify-complete")
prefix_list["Version"].should.equal(1)
described = ec2.describe_managed_prefix_lists(PrefixListIds=[prefix_list_id])[
"PrefixLists"
][0]
described.should.equal(prefix_list)
@mock_ec2
def test_modify_manage_prefix_list_specifying_version():
ec2 = boto3.client("ec2", region_name="us-west-1")
resp = ec2.create_managed_prefix_list(
PrefixListName="examplelist",
MaxEntries=2,
AddressFamily="?",
Entries=[
{"Cidr": "10.0.0.1", "Description": "entry1"},
{"Cidr": "10.0.0.2", "Description": "entry2"},
],
)
prefix_list_id = resp["PrefixList"]["PrefixListId"]
prefix_list = ec2.modify_managed_prefix_list(
PrefixListId=prefix_list_id,
AddEntries=[{"Cidr": "10.0.0.3", "Description": "entry3"}],
RemoveEntries=[{"Cidr": "10.0.0.2"}],
)["PrefixList"]
prefix_list["Version"].should.equal(2)
prefix_list = ec2.modify_managed_prefix_list(
PrefixListId=prefix_list_id,
CurrentVersion=2,
RemoveEntries=[{"Cidr": "10.0.0.1"}],
)["PrefixList"]
prefix_list["Version"].should.equal(3)
described = ec2.describe_managed_prefix_lists(PrefixListIds=[prefix_list_id])[
"PrefixLists"
][0]
described.should.equal(prefix_list)
entries = ec2.get_managed_prefix_list_entries(PrefixListId=prefix_list_id)[
"Entries"
]
entries.should.have.length_of(1)
entries.should.contain({"Cidr": "10.0.0.3", "Description": "entry3"})

31
tests/test_ec2/test_vpcs.py
View File

@ -945,3 +945,34 @@ def test_describe_vpc_end_points():
)
except ClientError as err:
assert err.response["Error"]["Code"] == "InvalidVpcEndpointId.NotFound"
@mock_ec2
def test_delete_vpc_end_points():
ec2 = boto3.client("ec2", region_name="us-west-1")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
route_table = ec2.create_route_table(VpcId=vpc["Vpc"]["VpcId"])
vpc_end_point1 = ec2.create_vpc_endpoint(
VpcId=vpc["Vpc"]["VpcId"],
ServiceName="com.amazonaws.us-east-1.s3",
RouteTableIds=[route_table["RouteTable"]["RouteTableId"]],
VpcEndpointType="gateway",
)["VpcEndpoint"]
vpc_end_point2 = ec2.create_vpc_endpoint(
VpcId=vpc["Vpc"]["VpcId"],
ServiceName="com.amazonaws.us-east-2.s3",
RouteTableIds=[route_table["RouteTable"]["RouteTableId"]],
VpcEndpointType="gateway",
)
vpc_endpoints = ec2.describe_vpc_endpoints()["VpcEndpoints"]
vpc_endpoints.should.have.length_of(2)
ec2.delete_vpc_endpoints(VpcEndpointIds=[vpc_end_point1["VpcEndpointId"]])
vpc_endpoints = ec2.describe_vpc_endpoints()["VpcEndpoints"]
vpc_endpoints.should.have.length_of(2)
states = set([vpce["State"] for vpce in vpc_endpoints])
states.should.equal({"available", "deleted"})

10
tests/test_elbv2/test_elbv2.py
View File

@ -1597,19 +1597,23 @@ def test_set_subnets():
len(resp["LoadBalancers"][0]["AvailabilityZones"]).should.equal(3)
# Only 1 AZ
with pytest.raises(ClientError):
with pytest.raises(ClientError) as ex:
client.set_subnets(LoadBalancerArn=arn, Subnets=[subnet1.id])
err = ex.value.response["Error"]
err["Message"].should.equal("More than 1 availability zone must be specified")
# Multiple subnets in same AZ
with pytest.raises(ClientError):
with pytest.raises(ClientError) as ex:
client.set_subnets(
LoadBalancerArn=arn, Subnets=[subnet1.id, subnet2.id, subnet2.id]
)
err = ex.value.response["Error"]
err["Message"].should.equal("The specified subnet does not exist.")
@mock_elbv2
@mock_ec2
def test_set_subnets():
def test_modify_load_balancer_attributes_idle_timeout():
client = boto3.client("elbv2", region_name="us-east-1")
ec2 = boto3.resource("ec2", region_name="us-east-1")

131
tests/test_events/test_events.py
View File

@ -16,7 +16,11 @@ from moto.events import mock_events
RULES = [
{"Name": "test1", "ScheduleExpression": "rate(5 minutes)"},
{"Name": "test2", "ScheduleExpression": "rate(1 minute)"},
{
"Name": "test2",
"ScheduleExpression": "rate(1 minute)",
"Tags": [{"Key": "tagk1", "Value": "tagv1"}],
},
{"Name": "test3", "EventPattern": '{"source": ["test-source"]}'},
]
@ -66,6 +70,7 @@ def generate_environment():
Name=rule["Name"],
ScheduleExpression=rule.get("ScheduleExpression", ""),
EventPattern=rule.get("EventPattern", ""),
Tags=rule.get("Tags", []),
)
targets = []
@ -203,11 +208,21 @@ def test_enable_disable_rule():
assert rule["State"] == "ENABLED"
# Test invalid name
try:
with pytest.raises(ClientError) as ex:
client.enable_rule(Name="junk")
except ClientError as ce:
assert ce.response["Error"]["Code"] == "ResourceNotFoundException"
err = ex.value.response["Error"]
err["Code"] == "ResourceNotFoundException"
@mock_events
def test_disable_unknown_rule():
client = generate_environment()
with pytest.raises(ClientError) as ex:
client.disable_rule(Name="unknown")
err = ex.value.response["Error"]
err["Message"].should.equal("Rule unknown does not exist.")
@mock_events
@ -452,6 +467,30 @@ def test_permissions():
assert resp_policy["Statement"][0]["Sid"] == "Account1"
@mock_events
def test_permission_policy():
client = boto3.client("events", "eu-central-1")
policy = {
"Statement": [
{
"Sid": "asdf",
"Action": "events:PutEvents",
"Principal": "111111111111",
"StatementId": "Account1",
"Effect": "n/a",
"Resource": "n/a",
}
]
}
client.put_permission(Policy=json.dumps(policy))
resp = client.describe_event_bus()
resp_policy = json.loads(resp["Policy"])
resp_policy["Statement"].should.have.length_of(1)
resp_policy["Statement"][0]["Sid"].should.equal("asdf")
@mock_events
def test_put_permission_errors():
client = boto3.client("events", "us-east-1")
@ -839,10 +878,38 @@ def test_delete_event_bus_errors():
)
@mock_events
def test_create_rule_with_tags():
client = generate_environment()
rule_name = "test2"
rule_arn = client.describe_rule(Name=rule_name).get("Arn")
actual = client.list_tags_for_resource(ResourceARN=rule_arn)["Tags"]
actual.should.equal([{"Key": "tagk1", "Value": "tagv1"}])
@mock_events
def test_delete_rule_with_tags():
client = generate_environment()
rule_name = "test2"
rule_arn = client.describe_rule(Name=rule_name).get("Arn")
client.delete_rule(Name=rule_name)
with pytest.raises(ClientError) as ex:
client.list_tags_for_resource(ResourceARN=rule_arn)
err = ex.value.response["Error"]
err["Message"].should.equal("Rule test2 does not exist on EventBus default.")
with pytest.raises(ClientError) as ex:
client.describe_rule(Name=rule_name)
err = ex.value.response["Error"]
err["Message"].should.equal("Rule test2 does not exist.")
@mock_events
def test_rule_tagging_happy():
client = generate_environment()
rule_name = get_random_rule()["Name"]
rule_name = "test1"
rule_arn = client.describe_rule(Name=rule_name).get("Arn")
tags = [{"Key": "key1", "Value": "value1"}, {"Key": "key2", "Value": "value2"}]
@ -2195,6 +2262,40 @@ def test_create_and_describe_connection():
description.should.have.key("CreationTime")
@mock_events
def test_create_and_update_connection():
client = boto3.client("events", "eu-central-1")
client.create_connection(
Name="test",
Description="test description",
AuthorizationType="API_KEY",
AuthParameters={
"ApiKeyAuthParameters": {"ApiKeyName": "test", "ApiKeyValue": "test"}
},
)
client.update_connection(Name="test", Description="updated desc")
description = client.describe_connection(Name="test")
description["Name"].should.equal("test")
description["Description"].should.equal("updated desc")
description["AuthorizationType"].should.equal("API_KEY")
description["ConnectionState"].should.equal("AUTHORIZED")
description.should.have.key("CreationTime")
@mock_events
def test_update_unknown_connection():
client = boto3.client("events", "eu-north-1")
with pytest.raises(ClientError) as ex:
client.update_connection(Name="unknown")
err = ex.value.response["Error"]
err["Message"].should.equal("Connection 'unknown' does not exist.")
@mock_events
def test_delete_connection():
client = boto3.client("events", "eu-central-1")
@ -2335,6 +2436,26 @@ def test_delete_api_destination():
client.list_api_destinations()["ApiDestinations"].should.have.length_of(0)
@mock_events
def test_describe_unknown_api_destination():
client = boto3.client("events", "eu-central-1")
with pytest.raises(ClientError) as ex:
client.describe_api_destination(Name="unknown")
err = ex.value.response["Error"]
err["Message"].should.equal("An api-destination 'unknown' does not exist.")
@mock_events
def test_delete_unknown_api_destination():
client = boto3.client("events", "eu-central-1")
with pytest.raises(ClientError) as ex:
client.delete_api_destination(Name="unknown")
err = ex.value.response["Error"]
err["Message"].should.equal("An api-destination 'unknown' does not exist.")
# Scenarios for describe_connection
# Scenario 01: Success
# Scenario 02: Failure - Connection not present