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

Greengrass Implement core_definition CRUD operation (#5193)

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

17
moto/greengrass/exceptions.py Normal file
View File

@ -0,0 +1,17 @@
from moto.core.exceptions import JsonRESTError
class GreengrassClientError(JsonRESTError):
code = 400
class IdNotFoundException(GreengrassClientError):
def __init__(self, msg):
self.code = 404
super().__init__("IdNotFoundException", msg)
class InvalidContainerDefinitionException(GreengrassClientError):
def __init__(self, msg):
self.code = 400
super().__init__("InvalidContainerDefinitionException", msg)

29
moto/greengrass/models.py
View File

@ -4,6 +4,10 @@ from datetime import datetime
from moto.core import BaseBackend, BaseModel, get_account_id
from moto.core.utils import BackendDict, iso_8601_datetime_with_milliseconds
from .exceptions import (
IdNotFoundException,
InvalidContainerDefinitionException,
)
class FakeCoreDefinition(BaseModel):
@ -78,6 +82,31 @@ class GreengrassBackend(BaseBackend):
)
return core_definition
def list_core_definitions(self):
return self.core_definitions.values()
def get_core_definition(self, core_definition_id):
if core_definition_id not in self.core_definitions:
raise IdNotFoundException("That Core List Definition does not exist")
return self.core_definitions[core_definition_id]
def delete_core_definition(self, core_definition_id):
if core_definition_id not in self.core_definitions:
raise IdNotFoundException("That cores definition does not exist.")
del self.core_definitions[core_definition_id]
del self.core_definition_versions[core_definition_id]
def update_core_definition(self, core_definition_id, name):
if name == "":
raise InvalidContainerDefinitionException(
"Input does not contain any attributes to be updated"
)
if core_definition_id not in self.core_definitions:
raise IdNotFoundException("That cores definition does not exist.")
self.core_definitions[core_definition_id].name = name
def create_core_definition_version(self, core_definition_id, cores):
definition = {"Cores": cores}

54
moto/greengrass/responses.py
View File

@ -11,8 +11,26 @@ class GreengrassResponse(BaseResponse):
def greengrass_backend(self):
return greengrass_backends[self.region]
def create_core_definition(self, request, full_url, headers):
def core_definitions(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.list_core_definitions()
if self.method == "POST":
return self.create_core_definition()
def list_core_definitions(self):
res = self.greengrass_backend.list_core_definitions()
return (
200,
{"status": 200},
json.dumps(
{"Definitions": [core_definition.to_dict() for core_definition in res]}
),
)
def create_core_definition(self):
name = self._get_param("Name")
initial_version = self._get_param("InitialVersion")
res = self.greengrass_backend.create_core_definition(
@ -20,6 +38,40 @@ class GreengrassResponse(BaseResponse):
)
return 201, {"status": 201}, json.dumps(res.to_dict())
def core_definition(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.get_core_definition()
if self.method == "DELETE":
return self.delete_core_definition()
if self.method == "PUT":
return self.update_core_definition()
def get_core_definition(self):
core_definition_id = self.path.split("/")[-1]
res = self.greengrass_backend.get_core_definition(
core_definition_id=core_definition_id
)
return 200, {"status": 200}, json.dumps(res.to_dict())
def delete_core_definition(self):
core_definition_id = self.path.split("/")[-1]
self.greengrass_backend.delete_core_definition(
core_definition_id=core_definition_id
)
return 200, {"status": 200}, json.dumps({})
def update_core_definition(self):
core_definition_id = self.path.split("/")[-1]
name = self._get_param("Name")
self.greengrass_backend.update_core_definition(
core_definition_id=core_definition_id, name=name
)
return 200, {"status": 200}, json.dumps({})
def create_core_definition_version(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
core_definition_id = self.path.split("/")[-2]

3
moto/greengrass/urls.py
View File

@ -9,6 +9,7 @@ response = GreengrassResponse()
url_paths = {
"{0}/greengrass/definition/cores$": response.create_core_definition,
"{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,
}

169
tests/test_greengrass/test_greengrass_core.py
View File

@ -1,5 +1,7 @@
import boto3
from botocore.client import ClientError
import freezegun
import pytest
from moto import mock_greengrass
from moto.core import get_account_id
@ -35,6 +37,173 @@ def test_create_core_definition():
res["ResponseMetadata"]["HTTPStatusCode"].should.equal(201)
@freezegun.freeze_time("2022-06-01 12:00:00")
@mock_greengrass
def test_list_core_definitions():
client = boto3.client("greengrass", region_name="ap-northeast-1")
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/CoreThing",
}
]
initial_version = {"Cores": cores}
core_name = "TestCore"
client.create_core_definition(InitialVersion=initial_version, Name=core_name)
res = client.list_core_definitions()
res.should.have.key("Definitions")
core_definition = res["Definitions"][0]
core_definition.should.have.key("Name").equals(core_name)
core_definition.should.have.key("Arn")
core_definition.should.have.key("Id")
core_definition.should.have.key("LatestVersion")
core_definition.should.have.key("LatestVersionArn")
if not TEST_SERVER_MODE:
core_definition.should.have.key("CreationTimestamp").equal(
"2022-06-01T12:00:00.000Z"
)
core_definition.should.have.key("LastUpdatedTimestamp").equals(
"2022-06-01T12:00:00.000Z"
)
@freezegun.freeze_time("2022-06-01 12:00:00")
@mock_greengrass
def test_get_core_definition():
client = boto3.client("greengrass", region_name="ap-northeast-1")
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/CoreThing",
}
]
initial_version = {"Cores": cores}
core_name = "TestCore"
create_res = client.create_core_definition(
InitialVersion=initial_version, Name=core_name
)
core_def_id = create_res["Id"]
arn = create_res["Arn"]
latest_version = create_res["LatestVersion"]
latest_version_arn = create_res["LatestVersionArn"]
get_res = client.get_core_definition(CoreDefinitionId=core_def_id)
get_res.should.have.key("Name").equals(core_name)
get_res.should.have.key("Arn").equals(arn)
get_res.should.have.key("Id").equals(core_def_id)
get_res.should.have.key("LatestVersion").equals(latest_version)
get_res.should.have.key("LatestVersionArn").equals(latest_version_arn)
if not TEST_SERVER_MODE:
get_res.should.have.key("CreationTimestamp").equal("2022-06-01T12:00:00.000Z")
get_res.should.have.key("LastUpdatedTimestamp").equals(
"2022-06-01T12:00:00.000Z"
)
@mock_greengrass
def test_delete_core_definition():
client = boto3.client("greengrass", region_name="ap-northeast-1")
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/CoreThing",
}
]
initial_version = {"Cores": cores}
create_res = client.create_core_definition(
InitialVersion=initial_version, Name="TestCore"
)
core_def_id = create_res["Id"]
client.get_core_definition(CoreDefinitionId=core_def_id)
client.delete_core_definition(CoreDefinitionId=core_def_id)
with pytest.raises(ClientError) as ex:
client.delete_core_definition(CoreDefinitionId=core_def_id)
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_update_core_definition():
client = boto3.client("greengrass", region_name="ap-northeast-1")
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/CoreThing",
}
]
initial_version = {"Cores": cores}
create_res = client.create_core_definition(
InitialVersion=initial_version, Name="TestCore"
)
core_def_id = create_res["Id"]
updated_core_name = "UpdatedCore"
client.update_core_definition(CoreDefinitionId=core_def_id, Name="UpdatedCore")
get_res = client.get_core_definition(CoreDefinitionId=core_def_id)
get_res.should.have.key("Name").equals(updated_core_name)
@mock_greengrass
def test_update_core_definition_with_empty_name():
client = boto3.client("greengrass", region_name="ap-northeast-1")
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/CoreThing",
}
]
initial_version = {"Cores": cores}
create_res = client.create_core_definition(
InitialVersion=initial_version, Name="TestCore"
)
core_def_id = create_res["Id"]
with pytest.raises(ClientError) as ex:
client.update_core_definition(CoreDefinitionId=core_def_id, Name="")
ex.value.response["Error"]["Message"].should.equal(
"Input does not contain any attributes to be updated"
)
ex.value.response["Error"]["Code"].should.equal(
"InvalidContainerDefinitionException"
)
@mock_greengrass
def test_update_core_definition_with_invalid_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
with pytest.raises(ClientError) as ex:
client.update_core_definition(
CoreDefinitionId="6fbffc21-989e-4d29-a793-a42f450a78c6", Name="abc"
)
ex.value.response["Error"]["Message"].should.equal(
"That cores 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_create_core_definition_version():