1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
import pytest
from botocore.exceptions import ClientError
from . import fixture_ec2, fixture_efs # noqa
@pytest.fixture(scope="function", name="file_system")
def fixture_file_system(efs):
create_fs_resp = efs.create_file_system(CreationToken="foobarbaz")
create_fs_resp.pop("ResponseMetadata")
yield create_fs_resp
@pytest.fixture(scope="function", name="subnet")
def fixture_subnet(ec2):
desc_sn_resp = ec2.describe_subnets()
subnet = desc_sn_resp["Subnets"][0]
yield subnet
def test_describe_mount_target_security_groups__unknown(efs):
with pytest.raises(ClientError) as exc_info:
efs.describe_mount_target_security_groups(MountTargetId="mt-asdf1234asdf")
err = exc_info.value.response["Error"]
assert err["Code"] == "MountTargetNotFound"
assert err["Message"] == "Mount target 'mt-asdf1234asdf' does not exist."
def test_describe_mount_target_security_groups(efs, ec2, file_system, subnet):
subnet_id = subnet["SubnetId"]
file_system_id = file_system["FileSystemId"]
desc_sg_resp = ec2.describe_security_groups()
security_group_id = desc_sg_resp["SecurityGroups"][0]["GroupId"]
# Create Mount Target
sample_input = {
"FileSystemId": file_system_id,
"SubnetId": subnet_id,
"SecurityGroups": [security_group_id],
}
create_mt_resp = efs.create_mount_target(**sample_input)
mount_target_id = create_mt_resp["MountTargetId"]
# Describe it's Security Groups
resp = efs.describe_mount_target_security_groups(MountTargetId=mount_target_id)
assert resp["SecurityGroups"] == [security_group_id]
def test_modify_mount_target_security_groups__unknown(efs):
with pytest.raises(ClientError) as exc_info:
efs.modify_mount_target_security_groups(
MountTargetId="mt-asdf1234asdf", SecurityGroups=[]
)
err = exc_info.value.response["Error"]
assert err["Code"] == "MountTargetNotFound"
assert err["Message"] == "Mount target 'mt-asdf1234asdf' does not exist."
def test_modify_mount_target_security_groups(efs, ec2, file_system, subnet):
subnet_id = subnet["SubnetId"]
file_system_id = file_system["FileSystemId"]
desc_sg_resp = ec2.describe_security_groups()["SecurityGroups"]
security_group_id = desc_sg_resp[0]["GroupId"]
# Create Mount Target
sample_input = {
"FileSystemId": file_system_id,
"SubnetId": subnet_id,
"SecurityGroups": [security_group_id],
}
create_mt_resp = efs.create_mount_target(**sample_input)
mount_target_id = create_mt_resp["MountTargetId"]
network_interface_id = create_mt_resp["NetworkInterfaceId"]
# Create alternative security groups
sg_id_2 = ec2.create_security_group(
VpcId=subnet["VpcId"], GroupName="sg-2", Description="SG-2"
)["GroupId"]
sg_id_3 = ec2.create_security_group(
VpcId=subnet["VpcId"], GroupName="sg-3", Description="SG-3"
)["GroupId"]
# Modify it's Security Groups
efs.modify_mount_target_security_groups(
MountTargetId=mount_target_id, SecurityGroups=[sg_id_2, sg_id_3]
)
# Describe it's Security Groups
resp = efs.describe_mount_target_security_groups(MountTargetId=mount_target_id)
assert resp["SecurityGroups"] == [sg_id_2, sg_id_3]
# Verify EC2 reflects this change
resp = ec2.describe_network_interfaces(NetworkInterfaceIds=[network_interface_id])
network_interface = resp["NetworkInterfaces"][0]
assert len(network_interface["Groups"]) == 2
assert {sg["GroupId"] for sg in network_interface["Groups"]} == {
sg_id_2,
sg_id_3,
}
|
