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

Greengrass Implement core_definition_version READ operation (#5195)

Browse Source
This commit is contained in:
cm-iwata 2022-06-08 06:17:02 +09:00 committed by GitHub
parent bc24e00e5c
commit c82b437195
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 201 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
moto/greengrass/exceptions.py
View File

@ -15,3 +15,9 @@ class InvalidContainerDefinitionException(GreengrassClientError):
def __init__(self, msg):
self.code = 400
super().__init__("InvalidContainerDefinitionException", msg)
class VersionNotFoundException(GreengrassClientError):
def __init__(self, msg):
self.code = 404
super().__init__("VersionNotFoundException", msg)

35
moto/greengrass/models.py
View File

@ -7,6 +7,7 @@ from moto.core.utils import BackendDict, iso_8601_datetime_with_milliseconds
from .exceptions import (
IdNotFoundException,
InvalidContainerDefinitionException,
VersionNotFoundException,
)
@ -45,8 +46,8 @@ class FakeCoreDefinitionVersion(BaseModel):
self.arn = f"arn:aws:greengrass:{region_name}:{get_account_id()}:greengrass/definition/cores/{self.core_definition_id}/versions/{self.version}"
self.created_at_datetime = datetime.utcnow()
def to_dict(self):
return {
def to_dict(self, include_detail=False):
obj = {
"Arn": self.arn,
"CreationTimestamp": iso_8601_datetime_with_milliseconds(
self.created_at_datetime
@ -55,6 +56,11 @@ class FakeCoreDefinitionVersion(BaseModel):
"Version": self.version,
}
if include_detail:
obj["Definition"] = self.definition
return obj
class GreengrassBackend(BaseBackend):
def __init__(self, region_name, account_id):
@ -124,5 +130,30 @@ class GreengrassBackend(BaseBackend):
return core_def_ver
def list_core_definition_versions(self, core_definition_id):
if core_definition_id not in self.core_definitions:
raise IdNotFoundException("That cores definition does not exist.")
return self.core_definition_versions[core_definition_id].values()
def get_core_definition_version(
self, core_definition_id, core_definition_version_id
):
if core_definition_id not in self.core_definitions:
raise IdNotFoundException("That cores definition does not exist.")
if (
core_definition_version_id
not in self.core_definition_versions[core_definition_id]
):
raise VersionNotFoundException(
f"Version {core_definition_version_id} of Core List Definition {core_definition_id} does not exist."
)
return self.core_definition_versions[core_definition_id][
core_definition_version_id
]
greengrass_backends = BackendDict(GreengrassBackend, "greengrass")

34
moto/greengrass/responses.py
View File

@ -72,8 +72,16 @@ class GreengrassResponse(BaseResponse):
)
return 200, {"status": 200}, json.dumps({})
def create_core_definition_version(self, request, full_url, headers):
def core_definition_versions(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.list_core_definition_versions()
if self.method == "POST":
return self.create_core_definition_version()
def create_core_definition_version(self):
core_definition_id = self.path.split("/")[-2]
cores = self._get_param("Cores")
@ -81,3 +89,27 @@ class GreengrassResponse(BaseResponse):
core_definition_id=core_definition_id, cores=cores
)
return 201, {"status": 201}, json.dumps(res.to_dict())
def list_core_definition_versions(self):
core_definition_id = self.path.split("/")[-2]
res = self.greengrass_backend.list_core_definition_versions(core_definition_id)
return (
200,
{"status": 200},
json.dumps({"Versions": [core_def_ver.to_dict() for core_def_ver in res]}),
)
def core_definition_version(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.get_core_definition_version()
def get_core_definition_version(self):
core_definition_id = self.path.split("/")[-3]
core_definition_version_id = self.path.split("/")[-1]
res = self.greengrass_backend.get_core_definition_version(
core_definition_id=core_definition_id,
core_definition_version_id=core_definition_version_id,
)
return 200, {"status": 200}, json.dumps(res.to_dict(include_detail=True))

3
moto/greengrass/urls.py
View File

@ -11,5 +11,6 @@ response = GreengrassResponse()
url_paths = {
"{0}/greengrass/definition/cores$": response.core_definitions,
"{0}/greengrass/definition/cores/(?P<definition_id>[^/]+)/?$": response.core_definition,
"{0}/greengrass/definition/cores/(?P<definition_id>[^/]+)/versions$": response.create_core_definition_version,
"{0}/greengrass/definition/cores/(?P<definition_id>[^/]+)/versions$": response.core_definition_versions,
"{0}/greengrass/definition/cores/(?P<definition_id>[^/]+)/versions/(?P<definition_version_id>[^/]+)/?$": response.core_definition_version,
}

127
tests/test_greengrass/test_greengrass_core.py
View File

@ -241,3 +241,130 @@ def test_create_core_definition_version():
core_def_ver_res["CreationTimestamp"].should.equal("2022-06-01T12:00:00.000Z")
core_def_ver_res.should.have.key("Id").equals(core_def_id)
core_def_ver_res.should.have.key("Version")
@freezegun.freeze_time("2022-06-01 12:00:00")
@mock_greengrass
def test_get_core_definition_version():
client = boto3.client("greengrass", region_name="ap-northeast-1")
initial_version = {
"Cores": [
{
"CertificateArn": f"arn:aws:iot:ap-northeast-1:{ACCOUNT_ID}:cert/36ed61be9c6271ae8da174e29d0e033c06af149d7b21672f3800fe322044554d",
"Id": "123456789",
"ThingArn": f"arn:aws:iot:ap-northeast-1:{ACCOUNT_ID}:thing/v1Thing",
}
]
}
core_def_res = client.create_core_definition(
InitialVersion=initial_version, Name="TestCore"
)
core_def_id = core_def_res["Id"]
core_def_ver_id = core_def_res["LatestVersion"]
core_def_ver_res = client.get_core_definition_version(
CoreDefinitionId=core_def_id, CoreDefinitionVersionId=core_def_ver_id
)
core_def_ver_res.should.have.key("Arn")
core_def_ver_res.should.have.key("CreationTimestamp")
core_def_ver_res.should.have.key("Definition").should.equal(initial_version)
if not TEST_SERVER_MODE:
core_def_ver_res["CreationTimestamp"].should.equal("2022-06-01T12:00:00.000Z")
core_def_ver_res.should.have.key("Id").equals(core_def_id)
core_def_ver_res.should.have.key("Version")
@mock_greengrass
def test_get_core_definition_version_with_invalid_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
with pytest.raises(ClientError) as ex:
client.get_core_definition_version(
CoreDefinitionId="fe2392e9-e67f-4308-af1b-ff94a128b231",
CoreDefinitionVersionId="cd2ea6dc-6634-4e89-8441-8003500435f9",
)
ex.value.response["Error"]["Message"].should.equal(
"That cores definition does not exist."
)
ex.value.response["Error"]["Code"].should.equal("IdNotFoundException")
@mock_greengrass
def test_get_core_definition_version_with_invalid_version_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
core_def_res = client.create_core_definition(
Name="TestCore",
InitialVersion={
"Cores": [
{
"CertificateArn": f"arn:aws:iot:ap-northeast-1:{ACCOUNT_ID}:cert/36ed61be9c6271ae8da174e29d0e033c06af149d7b21672f3800fe322044554d",
"Id": "123456789",
"ThingArn": f"arn:aws:iot:ap-northeast-1:{ACCOUNT_ID}:thing/v1Thing",
}
]
},
)
core_def_id = core_def_res["Id"]
invalid_version_id = "cd2ea6dc-6634-4e89-8441-8003500435f9"
with pytest.raises(ClientError) as ex:
client.get_core_definition_version(
CoreDefinitionId=core_def_id, CoreDefinitionVersionId=invalid_version_id
)
ex.value.response["Error"]["Message"].should.equal(
f"Version {invalid_version_id} of Core List Definition {core_def_id} does not exist."
)
ex.value.response["Error"]["Code"].should.equal("VersionNotFoundException")
@freezegun.freeze_time("2022-06-01 12:00:00")
@mock_greengrass
def test_list_core_definition_version():
client = boto3.client("greengrass", region_name="ap-northeast-1")
initial_version = {
"Cores": [
{
"CertificateArn": f"arn:aws:iot:ap-northeast-1:{ACCOUNT_ID}:cert/36ed61be9c6271ae8da174e29d0e033c06af149d7b21672f3800fe322044554d",
"Id": "123456789",
"ThingArn": f"arn:aws:iot:ap-northeast-1:{ACCOUNT_ID}:thing/v1Thing",
}
]
}
core_def_res = client.create_core_definition(
InitialVersion=initial_version, Name="TestCore"
)
core_def_id = core_def_res["Id"]
core_def_vers_res = client.list_core_definition_versions(
CoreDefinitionId=core_def_id
)
core_def_vers_res.should.have.key("Versions")
core_def_ver = core_def_vers_res["Versions"][0]
core_def_ver.should.have.key("Arn")
core_def_ver.should.have.key("CreationTimestamp")
if not TEST_SERVER_MODE:
core_def_ver["CreationTimestamp"].should.equal("2022-06-01T12:00:00.000Z")
core_def_ver.should.have.key("Id").equals(core_def_id)
core_def_ver.should.have.key("Version")
@mock_greengrass
def test_list_core_definition_version_with_invalid_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
with pytest.raises(ClientError) as ex:
client.list_core_definition_versions(
CoreDefinitionId="cd2ea6dc-6634-4e89-8441-8003500435f9"
)
ex.value.response["Error"]["Message"].should.equal(
"That cores definition does not exist."
)
ex.value.response["Error"]["Code"].should.equal("IdNotFoundException")