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

Greengrass Implement resource_definition APIs (#5235)

Browse Source
This commit is contained in:
cm-iwata 2022-06-16 21:01:39 +09:00 committed by GitHub
parent c3f06064ff
commit 6272d586ae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 307 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
moto/greengrass/exceptions.py
View File

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

24
moto/greengrass/models.py
View File

@ -7,6 +7,7 @@ from moto.core.utils import BackendDict, iso_8601_datetime_with_milliseconds
from .exceptions import (
GreengrassClientError,
IdNotFoundException,
InvalidInputException,
InvalidContainerDefinitionException,
VersionNotFoundException,
)
@ -354,6 +355,29 @@ class GreengrassBackend(BaseBackend):
return resource_def
def list_resource_definitions(self):
return self.resource_definitions
def get_resource_definition(self, resource_definition_id):
if resource_definition_id not in self.resource_definitions:
raise IdNotFoundException("That Resource List Definition does not exist.")
return self.resource_definitions[resource_definition_id]
def delete_resource_definition(self, resource_definition_id):
if resource_definition_id not in self.resource_definitions:
raise IdNotFoundException("That resources definition does not exist.")
del self.resource_definitions[resource_definition_id]
del self.resource_definition_versions[resource_definition_id]
def update_resource_definition(self, resource_definition_id, name):
if name == "":
raise InvalidInputException("Invalid resource name.")
if resource_definition_id not in self.resource_definitions:
raise IdNotFoundException("That resources definition does not exist.")
self.resource_definitions[resource_definition_id].name = name
def create_resource_definition_version(self, resource_definition_id, resources):
if resource_definition_id not in self.resource_definitions:

48
moto/greengrass/responses.py
View File

@ -236,6 +236,9 @@ class GreengrassResponse(BaseResponse):
if self.method == "POST":
return self.create_resource_definition()
if self.method == "GET":
return self.list_resource_definitions()
def create_resource_definition(self):
initial_version = self._get_param("InitialVersion")
@ -245,6 +248,51 @@ class GreengrassResponse(BaseResponse):
)
return 201, {"status": 201}, json.dumps(res.to_dict())
def list_resource_definitions(self):
res = self.greengrass_backend.list_resource_definitions()
return (
200,
{"status": 200},
json.dumps({"Definitions": [i.to_dict() for i in res.values()]}),
)
def resource_definition(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if self.method == "GET":
return self.get_resource_definition()
if self.method == "DELETE":
return self.delete_resource_definition()
if self.method == "PUT":
return self.update_resource_definition()
def get_resource_definition(self):
resource_definition_id = self.path.split("/")[-1]
res = self.greengrass_backend.get_resource_definition(
resource_definition_id=resource_definition_id
)
return 200, {"status": 200}, json.dumps(res.to_dict())
def delete_resource_definition(self):
resource_definition_id = self.path.split("/")[-1]
self.greengrass_backend.delete_resource_definition(
resource_definition_id=resource_definition_id
)
return 200, {"status": 200}, json.dumps({})
def update_resource_definition(self):
resource_definition_id = self.path.split("/")[-1]
name = self._get_param("Name")
self.greengrass_backend.update_resource_definition(
resource_definition_id=resource_definition_id, name=name
)
return 200, {"status": 200}, json.dumps({})
def resource_definition_versions(self, request, full_url, headers):
self.setup_class(request, full_url, headers)

1
moto/greengrass/urls.py
View File

@ -18,5 +18,6 @@ url_paths = {
"{0}/greengrass/definition/devices/(?P<definition_id>[^/]+)/versions$": response.device_definition_versions,
"{0}/greengrass/definition/devices/(?P<definition_id>[^/]+)/versions/(?P<definition_version_id>[^/]+)/?$": response.device_definition_version,
"{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,
}

228
tests/test_greengrass/test_greengrass_resource.py
View File

@ -270,3 +270,231 @@ def test_create_resources_definition_version_with_invalid_local_device_resource(
f", but got: {source_path}])"
)
ex.value.response["Error"]["Code"].should.equal("400")
@freezegun.freeze_time("2022-06-01 12:00:00")
@mock_greengrass
def test_list_resources():
client = boto3.client("greengrass", region_name="ap-northeast-1")
init_ver = {
"Resources": [
{
"Id": "123",
"Name": "test_directory",
"ResourceDataContainer": {
"LocalVolumeResourceData": {
"DestinationPath": "/test_dir",
"GroupOwnerSetting": {"AutoAddGroupOwner": True},
"SourcePath": "/home/ggc_user/test_dir",
}
},
}
]
}
resource_name = "MotoTestResource"
create_res = client.create_resource_definition(
InitialVersion=init_ver, Name=resource_name
)
res_def_id = create_res["Id"]
res_def_arn = create_res["Arn"]
res_ver = create_res["LatestVersion"]
res_ver_arn = create_res["LatestVersionArn"]
res = client.list_resource_definitions()
res["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
res.should.have.key("Definitions")
res["Definitions"].should.have.length_of(1)
definition = res["Definitions"][0]
definition.should.have.key("Id").equals(res_def_id)
definition.should.have.key("Arn").equals(res_def_arn)
definition.should.have.key("LatestVersion").equals(res_ver)
definition.should.have.key("LatestVersionArn").equals(res_ver_arn)
if not TEST_SERVER_MODE:
definition.should.have.key("CreationTimestamp").equals(
"2022-06-01T12:00:00.000Z"
)
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_resource_definition():
client = boto3.client("greengrass", region_name="ap-northeast-1")
init_ver = {
"Resources": [
{
"Id": "123",
"Name": "test_directory",
"ResourceDataContainer": {
"LocalVolumeResourceData": {
"DestinationPath": "/test_dir",
"GroupOwnerSetting": {"AutoAddGroupOwner": True},
"SourcePath": "/home/ggc_user/test_dir",
}
},
}
]
}
resource_name = "MotoTestResource"
create_res = client.create_resource_definition(
InitialVersion=init_ver, Name=resource_name
)
resource_def_id = create_res["Id"]
arn = create_res["Arn"]
latest_version = create_res["LatestVersion"]
latest_version_arn = create_res["LatestVersionArn"]
get_res = client.get_resource_definition(ResourceDefinitionId=resource_def_id)
get_res.should.have.key("Name").equals(resource_name)
get_res.should.have.key("Arn").equals(arn)
get_res.should.have.key("Id").equals(resource_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_get_resource_definition_with_invalid_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
with pytest.raises(ClientError) as ex:
client.get_resource_definition(
ResourceDefinitionId="76f22a66-176a-4474-b450-04099dc4b069"
)
ex.value.response["Error"]["Message"].should.equal(
"That Resource List Definition does not exist."
)
ex.value.response["Error"]["Code"].should.equal("IdNotFoundException")
@mock_greengrass
def test_delete_resource_definition():
client = boto3.client("greengrass", region_name="ap-northeast-1")
init_ver = {
"Resources": [
{
"Id": "123",
"Name": "test_directory",
"ResourceDataContainer": {
"LocalVolumeResourceData": {
"DestinationPath": "/test_dir",
"GroupOwnerSetting": {"AutoAddGroupOwner": True},
"SourcePath": "/home/ggc_user/test_dir",
}
},
}
]
}
create_res = client.create_resource_definition(
InitialVersion=init_ver, Name="TestResource"
)
resource_def_id = create_res["Id"]
del_res = client.delete_resource_definition(ResourceDefinitionId=resource_def_id)
del_res["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@mock_greengrass
def test_delete_resource_definition_with_invalid_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
with pytest.raises(ClientError) as ex:
client.delete_resource_definition(
ResourceDefinitionId="6fbffc21-989e-4d29-a793-a42f450a78c6"
)
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_update_resource_definition():
client = boto3.client("greengrass", region_name="ap-northeast-1")
init_ver = {
"Resources": [
{
"Id": "123",
"Name": "test_directory",
"ResourceDataContainer": {
"LocalVolumeResourceData": {
"DestinationPath": "/test_dir",
"GroupOwnerSetting": {"AutoAddGroupOwner": True},
"SourcePath": "/home/ggc_user/test_dir",
}
},
}
]
}
create_res = client.create_resource_definition(
InitialVersion=init_ver, Name="TestResource"
)
resource_def_id = create_res["Id"]
updated_resource_name = "UpdatedResource"
update_res = client.update_resource_definition(
ResourceDefinitionId=resource_def_id, Name=updated_resource_name
)
update_res["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
get_res = client.get_resource_definition(ResourceDefinitionId=resource_def_id)
get_res.should.have.key("Name").equals(updated_resource_name)
@mock_greengrass
def test_update_device_definition_with_empty_name():
client = boto3.client("greengrass", region_name="ap-northeast-1")
init_ver = {
"Resources": [
{
"Id": "123",
"Name": "test_directory",
"ResourceDataContainer": {
"LocalVolumeResourceData": {
"DestinationPath": "/test_dir",
"GroupOwnerSetting": {"AutoAddGroupOwner": True},
"SourcePath": "/home/ggc_user/test_dir",
}
},
}
]
}
create_res = client.create_resource_definition(
InitialVersion=init_ver, Name="TestResource"
)
resource_def_id = create_res["Id"]
with pytest.raises(ClientError) as ex:
client.update_resource_definition(ResourceDefinitionId=resource_def_id, Name="")
ex.value.response["Error"]["Message"].should.equal("Invalid resource name.")
ex.value.response["Error"]["Code"].should.equal("InvalidInputException")
@mock_greengrass
def test_update_resource_definition_with_invalid_id():
client = boto3.client("greengrass", region_name="ap-northeast-1")
with pytest.raises(ClientError) as ex:
client.update_resource_definition(
ResourceDefinitionId="6fbffc21-989e-4d29-a793-a42f450a78c6", Name="123"
)
ex.value.response["Error"]["Message"].should.equal(
"That resources definition does not exist."
)
ex.value.response["Error"]["Code"].should.equal("IdNotFoundException")