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
|
import re
from unittest import SkipTest
import pytest
import moto.server as server
from moto import mock_aws, settings
FILE_SYSTEMS = "/2015-02-01/file-systems"
MOUNT_TARGETS = "/2015-02-01/mount-targets"
@pytest.fixture(scope="function", name="aws_credentials")
def fixture_aws_credentials(monkeypatch):
"""Mocked AWS Credentials for moto."""
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing")
monkeypatch.setenv("AWS_SESSION_TOKEN", "testing")
@pytest.fixture(scope="function", name="efs_client")
def fixture_efs_client(aws_credentials):
if not settings.TEST_DECORATOR_MODE:
raise SkipTest("Using server directly - no point in testing ServerMode")
with mock_aws():
yield server.create_backend_app("efs").test_client()
@pytest.fixture(scope="function", name="subnet_id")
def fixture_subnet_id(aws_credentials):
with mock_aws():
ec2_client = server.create_backend_app("ec2").test_client()
resp = ec2_client.get("/?Action=DescribeSubnets")
subnet_ids = re.findall("<subnetId>(.*?)</subnetId>", resp.data.decode("utf-8"))
yield subnet_ids[0]
@pytest.fixture(scope="function", name="file_system_id")
def fixture_file_system_id(efs_client):
resp = efs_client.post(
FILE_SYSTEMS, json={"CreationToken": "foobarbaz", "Backup": True}
)
yield resp.json["FileSystemId"]
"""
Test each method current supported to ensure it connects via the server.
NOTE: this does NOT test whether the endpoints are really functioning, just that they
connect and respond. Tests of functionality are contained in `test_file_system` and
`test_mount_target`.
"""
def test_efs_file_system_create(efs_client):
res = efs_client.post(FILE_SYSTEMS, json={"CreationToken": "2398asdfkajsdf"})
assert res.status_code == 201
def test_efs_file_system_describe(efs_client):
res = efs_client.get(FILE_SYSTEMS)
assert res.status_code == 200
def test_efs_file_system_delete(file_system_id, efs_client):
res = efs_client.delete(f"/2015-02-01/file-systems/{file_system_id}")
assert res.status_code == 204
def test_efs_mount_target_create(file_system_id, subnet_id, efs_client):
res = efs_client.post(
"/2015-02-01/mount-targets",
json={"FileSystemId": file_system_id, "SubnetId": subnet_id},
)
assert res.status_code == 200
def test_efs_mount_target_describe(file_system_id, efs_client):
res = efs_client.get(f"/2015-02-01/mount-targets?FileSystemId={file_system_id}")
assert res.status_code == 200
def test_efs_mount_target_delete(file_system_id, subnet_id, efs_client):
create_res = efs_client.post(
"/2015-02-01/mount-targets",
json={"FileSystemId": file_system_id, "SubnetId": subnet_id},
)
mt_id = create_res.json["MountTargetId"]
res = efs_client.delete(f"/2015-02-01/mount-targets/{mt_id}")
assert res.status_code == 204
def test_efs_describe_backup_policy(file_system_id, efs_client):
res = efs_client.get(f"/2015-02-01/file-systems/{file_system_id}/backup-policy")
assert res.status_code == 200
|
