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

Greengrass Implement resource_definition_version APIs (#5237)

Browse Source
This commit is contained in:
cm-iwata 2022-06-17 19:18:43 +09:00 committed by GitHub
parent 6a1a8ee1f8
commit 19bb5f96d2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 204 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
moto/greengrass/models.py
View File

@ -459,6 +459,32 @@ class GreengrassBackend(BaseBackend):
return resource_def_ver
def list_resource_definition_versions(self, resource_definition_id):
if resource_definition_id not in self.resource_definition_versions:
raise IdNotFoundException("That resources definition does not exist.")
return self.resource_definition_versions[resource_definition_id].values()
def get_resource_definition_version(
self, resource_definition_id, resource_definition_version_id
):
if resource_definition_id not in self.resource_definition_versions:
raise IdNotFoundException("That resources definition does not exist.")
if (
resource_definition_version_id
not in self.resource_definition_versions[resource_definition_id]
):
raise VersionNotFoundException(
f"Version {resource_definition_version_id} of Resource List Definition {resource_definition_id} does not exist."
)
return self.resource_definition_versions[resource_definition_id][
resource_definition_version_id
]
@staticmethod
def _validate_resources(resources):
for resource in resources:

33
moto/greengrass/responses.py
View File

@ -299,6 +299,9 @@ class GreengrassResponse(BaseResponse):
if self.method == "POST":
return self.create_resource_definition_version()
if self.method == "GET":
return self.list_resource_definition_versions()
def create_resource_definition_version(self):
resource_definition_id = self.path.split("/")[-2]
@ -309,6 +312,36 @@ class GreengrassResponse(BaseResponse):
)
return 201, {"status": 201}, json.dumps(res.to_dict())
def list_resource_definition_versions(self):
resource_device_definition_id = self.path.split("/")[-2]
res = self.greengrass_backend.list_resource_definition_versions(
resource_device_definition_id
)
return (
200,
{"status": 200},
json.dumps(
{"Versions": [resource_def_ver.to_dict() for resource_def_ver in res]}
),
)
def resource_definition_version(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.get_resource_definition_version()
def get_resource_definition_version(self):
resource_definition_id = self.path.split("/")[-3]
resource_definition_version_id = self.path.split("/")[-1]
res = self.greengrass_backend.get_resource_definition_version(
resource_definition_id=resource_definition_id,
resource_definition_version_id=resource_definition_version_id,
)
return 200, {"status": 200}, json.dumps(res.to_dict())
def function_definitions(self, request, full_url, headers):
self.setup_class(request, full_url, headers)

1
moto/greengrass/urls.py
View File

@ -24,4 +24,5 @@ url_paths = {
"{0}/greengrass/definition/resources$": response.resource_definitions,
"{0}/greengrass/definition/resources/(?P<definition_id>[^/]+)/?$": response.resource_definition,
"{0}/greengrass/definition/resources/(?P<definition_id>[^/]+)/versions$": response.resource_definition_versions,
"{0}/greengrass/definition/resources/(?P<definition_id>[^/]+)/versions/(?P<definition_version_id>[^/]+)/?$": response.resource_definition_version,
}

144
tests/test_greengrass/test_greengrass_resource.py
View File

@ -498,3 +498,147 @@ def test_update_resource_definition_with_invalid_id():
"That resources definition does not exist."
)
ex.value.response["Error"]["Code"].should.equal("IdNotFoundException")
@freezegun.freeze_time("2022-06-01 12:00:00")
@mock_greengrass
def test_list_resource_definition_versions():
client = boto3.client("greengrass", region_name="ap-northeast-1")
resources = [
{
"Id": "123",
"Name": "test_directory",
"ResourceDataContainer": {
"LocalDeviceResourceData": {
"SourcePath": "/dev/null",
}
},
}
]
initial_version = {"Resources": resources}
create_res = client.create_resource_definition(
InitialVersion=initial_version, Name="TestResource"
)
resource_def_id = create_res["Id"]
resource_def_vers_res = client.list_resource_definition_versions(
ResourceDefinitionId=resource_def_id
)
resource_def_vers_res.should.have.key("Versions")
resource_def_vers_res["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
device_def_ver = resource_def_vers_res["Versions"][0]
device_def_ver.should.have.key("Arn")
device_def_ver.should.have.key("CreationTimestamp")
device_def_ver.should.have.key("Id").equals(resource_def_id)
device_def_ver.should.have.key("Version")
if not TEST_SERVER_MODE:
device_def_ver["CreationTimestamp"].should.equal("2022-06-01T12:00:00.000Z")
@mock_greengrass
def test_list_resource_definition_versions_with_invalid_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
with pytest.raises(ClientError) as ex:
client.list_resource_definition_versions(
ResourceDefinitionId="fe2392e9-e67f-4308-af1b-ff94a128b231"
)
ex.value.response["Error"]["Message"].should.equal(
"That resources definition does not exist."
)
ex.value.response["Error"]["Code"].should.equal("IdNotFoundException")
@freezegun.freeze_time("2022-06-01 12:00:00")
@mock_greengrass
def test_get_resource_definition_version():
client = boto3.client("greengrass", region_name="ap-northeast-1")
resources = [
{
"Id": "123",
"Name": "test_directory",
"ResourceDataContainer": {
"LocalDeviceResourceData": {
"SourcePath": "/dev/null",
}
},
}
]
initial_version = {"Resources": resources}
create_res = client.create_resource_definition(
InitialVersion=initial_version, Name="TestResource"
)
resource_def_id = create_res["Id"]
resource_def_ver_id = create_res["LatestVersion"]
resource_def_ver_res = client.get_resource_definition_version(
ResourceDefinitionId=resource_def_id,
ResourceDefinitionVersionId=resource_def_ver_id,
)
resource_def_ver_res.should.have.key("Arn")
resource_def_ver_res.should.have.key("CreationTimestamp")
resource_def_ver_res.should.have.key("Definition").should.equal(initial_version)
resource_def_ver_res.should.have.key("Id").equals(resource_def_id)
resource_def_ver_res.should.have.key("Version")
resource_def_ver_res["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
if not TEST_SERVER_MODE:
resource_def_ver_res["CreationTimestamp"].should.equal(
"2022-06-01T12:00:00.000Z"
)
@mock_greengrass
def test_get_resource_definition_version_with_invalid_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
with pytest.raises(ClientError) as ex:
client.get_resource_definition_version(
ResourceDefinitionId="fe2392e9-e67f-4308-af1b-ff94a128b231",
ResourceDefinitionVersionId="cd2ea6dc-6634-4e89-8441-8003500435f9",
)
ex.value.response["Error"]["Message"].should.equal(
"That resources definition does not exist."
)
ex.value.response["Error"]["Code"].should.equal("IdNotFoundException")
@mock_greengrass
def test_get_resource_definition_version_with_invalid_version_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
resources = [
{
"Id": "123",
"Name": "test_directory",
"ResourceDataContainer": {
"LocalDeviceResourceData": {
"SourcePath": "/dev/null",
}
},
}
]
initial_version = {"Resources": resources}
create_res = client.create_resource_definition(
InitialVersion=initial_version, Name="TestResource"
)
resource_def_id = create_res["Id"]
invalid_resource_version_id = "6fbffc21-989e-4d29-a793-a42f450a78c6"
with pytest.raises(ClientError) as ex:
client.get_resource_definition_version(
ResourceDefinitionId=resource_def_id,
ResourceDefinitionVersionId=invalid_resource_version_id,
)
ex.value.response["Error"]["Message"].should.equal(
f"Version {invalid_resource_version_id} of Resource List Definition {resource_def_id} does not exist."
)
ex.value.response["Error"]["Code"].should.equal("VersionNotFoundException")