diff --git a/moto/core/responses.py b/moto/core/responses.py index 5951ab170a5..83770e0288e 100644 --- a/moto/core/responses.py +++ b/moto/core/responses.py @@ -491,7 +491,7 @@ def uri_to_regexp(self, uri: str) -> str: def _convert(elem: str, is_last: bool) -> str: if not re.match("^{.*}$", elem): - return elem + return elem.replace("$", r"\$") name = ( elem.replace("{", "") .replace("}", "") diff --git a/moto/greengrass/responses.py b/moto/greengrass/responses.py index b9e99769f18..5e94a32a12a 100644 --- a/moto/greengrass/responses.py +++ b/moto/greengrass/responses.py @@ -1,5 +1,4 @@ import json -from typing import Any from moto.core.common_types import TYPE_RESPONSE from moto.core.responses import BaseResponse @@ -16,25 +15,10 @@ def __init__(self) -> None: def greengrass_backend(self) -> GreengrassBackend: return greengrass_backends[self.current_account][self.region] - def core_definitions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - 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) -> TYPE_RESPONSE: + def list_core_definitions(self) -> str: res = self.greengrass_backend.list_core_definitions() - return ( - 200, - {"status": 200}, - json.dumps( - {"Definitions": [core_definition.to_dict() for core_definition in res]} - ), + return json.dumps( + {"Definitions": [core_definition.to_dict() for core_definition in res]} ) def create_core_definition(self) -> TYPE_RESPONSE: @@ -45,52 +29,27 @@ def create_core_definition(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def core_definition( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - 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) -> TYPE_RESPONSE: + def get_core_definition(self) -> str: 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()) + return json.dumps(res.to_dict()) - def delete_core_definition(self) -> TYPE_RESPONSE: + def delete_core_definition(self) -> str: 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({}) + return json.dumps({}) - def update_core_definition(self) -> TYPE_RESPONSE: + def update_core_definition(self) -> str: 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 core_definition_versions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - 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() + return "{}" def create_core_definition_version(self) -> TYPE_RESPONSE: core_definition_id = self.path.split("/")[-2] @@ -101,42 +60,21 @@ def create_core_definition_version(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_core_definition_versions(self) -> TYPE_RESPONSE: + def list_core_definition_versions(self) -> str: 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]}), + return json.dumps( + {"Versions": [core_def_ver.to_dict() for core_def_ver in res]} ) - def core_definition_version( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_core_definition_version() - - def get_core_definition_version(self) -> TYPE_RESPONSE: + def get_core_definition_version(self) -> str: 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)) - - def device_definitions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_device_definition() - - if self.method == "GET": - return self.list_device_definition() + return json.dumps(res.to_dict(include_detail=True)) def create_device_definition(self) -> TYPE_RESPONSE: name = self._get_param("Name") @@ -146,31 +84,12 @@ def create_device_definition(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_device_definition(self) -> TYPE_RESPONSE: + def list_device_definitions(self) -> str: res = self.greengrass_backend.list_device_definitions() - return ( - 200, - {"status": 200}, - json.dumps( - { - "Definitions": [ - device_definition.to_dict() for device_definition in res - ] - } - ), + return json.dumps( + {"Definitions": [device_definition.to_dict() for device_definition in res]} ) - def device_definition_versions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_device_definition_version() - - if self.method == "GET": - return self.list_device_definition_versions() - def create_device_definition_version(self) -> TYPE_RESPONSE: device_definition_id = self.path.split("/")[-2] devices = self._get_param("Devices") @@ -180,82 +99,45 @@ def create_device_definition_version(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_device_definition_versions(self) -> TYPE_RESPONSE: + def list_device_definition_versions(self) -> str: device_definition_id = self.path.split("/")[-2] res = self.greengrass_backend.list_device_definition_versions( device_definition_id ) - return ( - 200, - {"status": 200}, - json.dumps( - {"Versions": [device_def_ver.to_dict() for device_def_ver in res]} - ), + return json.dumps( + {"Versions": [device_def_ver.to_dict() for device_def_ver in res]} ) - def device_definition( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_device_definition() - - if self.method == "DELETE": - return self.delete_device_definition() - - if self.method == "PUT": - return self.update_device_definition() - - def get_device_definition(self) -> TYPE_RESPONSE: + def get_device_definition(self) -> str: device_definition_id = self.path.split("/")[-1] res = self.greengrass_backend.get_device_definition( device_definition_id=device_definition_id ) - return 200, {"status": 200}, json.dumps(res.to_dict()) + return json.dumps(res.to_dict()) - def delete_device_definition(self) -> TYPE_RESPONSE: + def delete_device_definition(self) -> str: device_definition_id = self.path.split("/")[-1] self.greengrass_backend.delete_device_definition( device_definition_id=device_definition_id ) - return 200, {"status": 200}, json.dumps({}) + return json.dumps({}) - def update_device_definition(self) -> TYPE_RESPONSE: + def update_device_definition(self) -> str: device_definition_id = self.path.split("/")[-1] name = self._get_param("Name") self.greengrass_backend.update_device_definition( device_definition_id=device_definition_id, name=name ) - return 200, {"status": 200}, json.dumps({}) - - def device_definition_version( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_device_definition_version() + return "{}" - def get_device_definition_version(self) -> TYPE_RESPONSE: + def get_device_definition_version(self) -> str: device_definition_id = self.path.split("/")[-3] device_definition_version_id = self.path.split("/")[-1] res = self.greengrass_backend.get_device_definition_version( device_definition_id=device_definition_id, device_definition_version_id=device_definition_version_id, ) - return 200, {"status": 200}, json.dumps(res.to_dict(include_detail=True)) - - def resource_definitions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_resource_definition() - - if self.method == "GET": - return self.list_resource_definitions() + return json.dumps(res.to_dict(include_detail=True)) def create_resource_definition(self) -> TYPE_RESPONSE: initial_version = self._get_param("InitialVersion") @@ -265,60 +147,31 @@ def create_resource_definition(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_resource_definitions(self) -> TYPE_RESPONSE: + def list_resource_definitions(self) -> str: res = self.greengrass_backend.list_resource_definitions() - return ( - 200, - {"status": 200}, - json.dumps({"Definitions": [i.to_dict() for i in res]}), - ) - - def resource_definition( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_resource_definition() + return json.dumps({"Definitions": [i.to_dict() for i in res]}) - if self.method == "DELETE": - return self.delete_resource_definition() - - if self.method == "PUT": - return self.update_resource_definition() - - def get_resource_definition(self) -> TYPE_RESPONSE: + def get_resource_definition(self) -> str: 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()) + return json.dumps(res.to_dict()) - def delete_resource_definition(self) -> TYPE_RESPONSE: + def delete_resource_definition(self) -> str: 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({}) + return "{}" - def update_resource_definition(self) -> TYPE_RESPONSE: + def update_resource_definition(self) -> str: 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( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_resource_definition_version() - - if self.method == "GET": - return self.list_resource_definition_versions() + return "{}" def create_resource_definition_version(self) -> TYPE_RESPONSE: resource_definition_id = self.path.split("/")[-2] @@ -329,47 +182,24 @@ def create_resource_definition_version(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_resource_definition_versions(self) -> TYPE_RESPONSE: + def list_resource_definition_versions(self) -> str: 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]} - ), + return json.dumps( + {"Versions": [resource_def_ver.to_dict() for resource_def_ver in res]} ) - def resource_definition_version( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_resource_definition_version() - - def get_resource_definition_version(self) -> TYPE_RESPONSE: + def get_resource_definition_version(self) -> str: 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( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_function_definition() - - if self.method == "GET": - return self.list_function_definitions() + return json.dumps(res.to_dict()) def create_function_definition(self) -> TYPE_RESPONSE: initial_version = self._get_param("InitialVersion") @@ -379,62 +209,33 @@ def create_function_definition(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_function_definitions(self) -> TYPE_RESPONSE: + def list_function_definitions(self) -> str: res = self.greengrass_backend.list_function_definitions() - return ( - 200, - {"status": 200}, - json.dumps( - {"Definitions": [func_definition.to_dict() for func_definition in res]} - ), + return json.dumps( + {"Definitions": [func_definition.to_dict() for func_definition in res]} ) - def function_definition( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_function_definition() - - if self.method == "DELETE": - return self.delete_function_definition() - - if self.method == "PUT": - return self.update_function_definition() - - def get_function_definition(self) -> TYPE_RESPONSE: + def get_function_definition(self) -> str: function_definition_id = self.path.split("/")[-1] res = self.greengrass_backend.get_function_definition( function_definition_id=function_definition_id, ) - return 200, {"status": 200}, json.dumps(res.to_dict()) + return json.dumps(res.to_dict()) - def delete_function_definition(self) -> TYPE_RESPONSE: + def delete_function_definition(self) -> str: function_definition_id = self.path.split("/")[-1] self.greengrass_backend.delete_function_definition( function_definition_id=function_definition_id, ) - return 200, {"status": 200}, json.dumps({}) + return "{}" - def update_function_definition(self) -> TYPE_RESPONSE: + def update_function_definition(self) -> str: function_definition_id = self.path.split("/")[-1] name = self._get_param("Name") self.greengrass_backend.update_function_definition( function_definition_id=function_definition_id, name=name ) - return 200, {"status": 200}, json.dumps({}) - - def function_definition_versions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_function_definition_version() - - if self.method == "GET": - return self.list_function_definition_versions() + return "{}" def create_function_definition_version(self) -> TYPE_RESPONSE: default_config = self._get_param("DefaultConfig") @@ -448,41 +249,22 @@ def create_function_definition_version(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_function_definition_versions(self) -> TYPE_RESPONSE: + def list_function_definition_versions(self) -> str: function_definition_id = self.path.split("/")[-2] res = self.greengrass_backend.list_function_definition_versions( function_definition_id=function_definition_id ) versions = [i.to_dict() for i in res.values()] - return 200, {"status": 200}, json.dumps({"Versions": versions}) - - def function_definition_version( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) + return json.dumps({"Versions": versions}) - if self.method == "GET": - return self.get_function_definition_version() - - def get_function_definition_version(self) -> TYPE_RESPONSE: + def get_function_definition_version(self) -> str: function_definition_id = self.path.split("/")[-3] function_definition_version_id = self.path.split("/")[-1] res = self.greengrass_backend.get_function_definition_version( function_definition_id=function_definition_id, function_definition_version_id=function_definition_version_id, ) - return 200, {"status": 200}, json.dumps(res.to_dict()) - - def subscription_definitions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_subscription_definition() - - if self.method == "GET": - return self.list_subscription_definitions() + return json.dumps(res.to_dict()) def create_subscription_definition(self) -> TYPE_RESPONSE: initial_version = self._get_param("InitialVersion") @@ -492,67 +274,37 @@ def create_subscription_definition(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_subscription_definitions(self) -> TYPE_RESPONSE: + def list_subscription_definitions(self) -> str: res = self.greengrass_backend.list_subscription_definitions() - return ( - 200, - {"status": 200}, - json.dumps( - { - "Definitions": [ - subscription_definition.to_dict() - for subscription_definition in res - ] - } - ), - ) - - def subscription_definition( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_subscription_definition() - - if self.method == "DELETE": - return self.delete_subscription_definition() - - if self.method == "PUT": - return self.update_subscription_definition() - - def get_subscription_definition(self) -> TYPE_RESPONSE: + return json.dumps( + { + "Definitions": [ + subscription_definition.to_dict() for subscription_definition in res + ] + } + ) + + def get_subscription_definition(self) -> str: subscription_definition_id = self.path.split("/")[-1] res = self.greengrass_backend.get_subscription_definition( subscription_definition_id=subscription_definition_id ) - return 200, {"status": 200}, json.dumps(res.to_dict()) + return json.dumps(res.to_dict()) - def delete_subscription_definition(self) -> TYPE_RESPONSE: + def delete_subscription_definition(self) -> str: subscription_definition_id = self.path.split("/")[-1] self.greengrass_backend.delete_subscription_definition( subscription_definition_id=subscription_definition_id ) - return 200, {"status": 200}, json.dumps({}) + return "{}" - def update_subscription_definition(self) -> TYPE_RESPONSE: + def update_subscription_definition(self) -> str: subscription_definition_id = self.path.split("/")[-1] name = self._get_param("Name") self.greengrass_backend.update_subscription_definition( subscription_definition_id=subscription_definition_id, name=name ) - return 200, {"status": 200}, json.dumps({}) - - def subscription_definition_versions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_subscription_definition_version() - - if self.method == "GET": - return self.list_subscription_definition_versions() + return "{}" def create_subscription_definition_version(self) -> TYPE_RESPONSE: subscription_definition_id = self.path.split("/")[-2] @@ -563,39 +315,22 @@ def create_subscription_definition_version(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_subscription_definition_versions(self) -> TYPE_RESPONSE: + def list_subscription_definition_versions(self) -> str: subscription_definition_id = self.path.split("/")[-2] res = self.greengrass_backend.list_subscription_definition_versions( subscription_definition_id=subscription_definition_id ) versions = [i.to_dict() for i in res.values()] - return 200, {"status": 200}, json.dumps({"Versions": versions}) + return json.dumps({"Versions": versions}) - def subscription_definition_version( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_subscription_definition_version() - - def get_subscription_definition_version(self) -> TYPE_RESPONSE: + def get_subscription_definition_version(self) -> str: subscription_definition_id = self.path.split("/")[-3] subscription_definition_version_id = self.path.split("/")[-1] res = self.greengrass_backend.get_subscription_definition_version( subscription_definition_id=subscription_definition_id, subscription_definition_version_id=subscription_definition_version_id, ) - return 200, {"status": 200}, json.dumps(res.to_dict()) - - def groups(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_group() - - if self.method == "GET": - return self.list_groups() + return json.dumps(res.to_dict()) def create_group(self) -> TYPE_RESPONSE: initial_version = self._get_param("InitialVersion") @@ -605,54 +340,25 @@ def create_group(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_groups(self) -> TYPE_RESPONSE: + def list_groups(self) -> str: res = self.greengrass_backend.list_groups() - return ( - 200, - {"status": 200}, - json.dumps({"Groups": [group.to_dict() for group in res]}), - ) - - def group(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_group() - - if self.method == "DELETE": - return self.delete_group() + return json.dumps({"Groups": [group.to_dict() for group in res]}) - if self.method == "PUT": - return self.update_group() - - def get_group(self) -> TYPE_RESPONSE: + def get_group(self) -> str: group_id = self.path.split("/")[-1] res = self.greengrass_backend.get_group(group_id=group_id) - return 200, {"status": 200}, json.dumps(res.to_dict()) # type: ignore + return json.dumps(res.to_dict()) # type: ignore - def delete_group(self) -> TYPE_RESPONSE: + def delete_group(self) -> str: group_id = self.path.split("/")[-1] - self.greengrass_backend.delete_group( - group_id=group_id, - ) - return 200, {"status": 200}, json.dumps({}) + self.greengrass_backend.delete_group(group_id=group_id) + return "{}" - def update_group(self) -> TYPE_RESPONSE: + def update_group(self) -> str: group_id = self.path.split("/")[-1] name = self._get_param("Name") self.greengrass_backend.update_group(group_id=group_id, name=name) - return 200, {"status": 200}, json.dumps({}) - - def group_versions( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_group_version() - - if self.method == "GET": - return self.list_group_versions() + return "{}" def create_group_version(self) -> TYPE_RESPONSE: group_id = self.path.split("/")[-2] @@ -679,40 +385,21 @@ def create_group_version(self) -> TYPE_RESPONSE: ) return 201, {"status": 201}, json.dumps(res.to_dict()) - def list_group_versions(self) -> TYPE_RESPONSE: + def list_group_versions(self) -> str: group_id = self.path.split("/")[-2] res = self.greengrass_backend.list_group_versions(group_id=group_id) - return ( - 200, - {"status": 200}, - json.dumps({"Versions": [group_ver.to_dict() for group_ver in res]}), - ) - - def group_version(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - - if self.method == "GET": - return self.get_group_version() + return json.dumps({"Versions": [group_ver.to_dict() for group_ver in res]}) - def get_group_version(self) -> TYPE_RESPONSE: + def get_group_version(self) -> str: group_id = self.path.split("/")[-3] group_version_id = self.path.split("/")[-1] res = self.greengrass_backend.get_group_version( group_id=group_id, group_version_id=group_version_id, ) - return 200, {"status": 200}, json.dumps(res.to_dict(include_detail=True)) - - def deployments(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.create_deployment() - - if self.method == "GET": - return self.list_deployments() + return json.dumps(res.to_dict(include_detail=True)) - def create_deployment(self) -> TYPE_RESPONSE: + def create_deployment(self) -> str: group_id = self.path.split("/")[-2] group_version_id = self._get_param("GroupVersionId") deployment_type = self._get_param("DeploymentType") @@ -724,33 +411,17 @@ def create_deployment(self) -> TYPE_RESPONSE: deployment_type=deployment_type, deployment_id=deployment_id, ) - return 200, {"status": 200}, json.dumps(res.to_dict()) + return json.dumps(res.to_dict()) - def list_deployments(self) -> TYPE_RESPONSE: + def list_deployments(self) -> str: group_id = self.path.split("/")[-2] res = self.greengrass_backend.list_deployments(group_id=group_id) - deployments = ( - [] - if len(res) == 0 - else [deployment.to_dict(include_detail=True) for deployment in res] - ) - - return ( - 200, - {"status": 200}, - json.dumps({"Deployments": deployments}), - ) + deployments = [deployment.to_dict(include_detail=True) for deployment in res] - def deployment_satus( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) + return json.dumps({"Deployments": deployments}) - if self.method == "GET": - return self.get_deployment_status() - - def get_deployment_status(self) -> TYPE_RESPONSE: + def get_deployment_status(self) -> str: group_id = self.path.split("/")[-4] deployment_id = self.path.split("/")[-2] @@ -758,59 +429,29 @@ def get_deployment_status(self) -> TYPE_RESPONSE: group_id=group_id, deployment_id=deployment_id, ) - return 200, {"status": 200}, json.dumps(res.to_dict()) - - def deployments_reset( # type: ignore[return] - self, request: Any, full_url: str, headers: Any - ) -> TYPE_RESPONSE: - self.setup_class(request, full_url, headers) - - if self.method == "POST": - return self.reset_deployments() + return json.dumps(res.to_dict()) - def reset_deployments(self) -> TYPE_RESPONSE: + def reset_deployments(self) -> str: group_id = self.path.split("/")[-3] - res = self.greengrass_backend.reset_deployments( - group_id=group_id, - ) - return 200, {"status": 200}, json.dumps(res.to_dict()) - - def role(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - - if self.method == "PUT": - return self.associate_role_to_group() - - if self.method == "GET": - return self.get_associated_role() - - if self.method == "DELETE": - return self.disassociate_role_from_group() + res = self.greengrass_backend.reset_deployments(group_id=group_id) + return json.dumps(res.to_dict()) - def associate_role_to_group(self) -> TYPE_RESPONSE: + def associate_role_to_group(self) -> str: group_id = self.path.split("/")[-2] role_arn = self._get_param("RoleArn") res = self.greengrass_backend.associate_role_to_group( group_id=group_id, role_arn=role_arn, ) - return 200, {"status": 200}, json.dumps(res.to_dict()) + return json.dumps(res.to_dict()) - def get_associated_role(self) -> TYPE_RESPONSE: + def get_associated_role(self) -> str: group_id = self.path.split("/")[-2] - res = self.greengrass_backend.get_associated_role( - group_id=group_id, - ) - return 200, {"status": 200}, json.dumps(res.to_dict(include_detail=True)) + res = self.greengrass_backend.get_associated_role(group_id=group_id) + return json.dumps(res.to_dict(include_detail=True)) - def disassociate_role_from_group(self) -> TYPE_RESPONSE: + def disassociate_role_from_group(self) -> str: group_id = self.path.split("/")[-2] - self.greengrass_backend.disassociate_role_from_group( - group_id=group_id, - ) - return ( - 200, - {"status": 200}, - json.dumps({"DisassociatedAt": iso_8601_datetime_with_milliseconds()}), - ) + self.greengrass_backend.disassociate_role_from_group(group_id=group_id) + return json.dumps({"DisassociatedAt": iso_8601_datetime_with_milliseconds()}) diff --git a/moto/greengrass/urls.py b/moto/greengrass/urls.py index 70cbedf0d40..ff780751dbb 100644 --- a/moto/greengrass/urls.py +++ b/moto/greengrass/urls.py @@ -5,88 +5,32 @@ ] url_paths = { - "{0}/greengrass/definition/cores$": GreengrassResponse.method_dispatch( - GreengrassResponse.core_definitions - ), - "{0}/greengrass/definition/cores/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.core_definition - ), - "{0}/greengrass/definition/cores/(?P[^/]+)/versions$": GreengrassResponse.method_dispatch( - GreengrassResponse.core_definition_versions - ), - "{0}/greengrass/definition/cores/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.core_definition_version - ), - "{0}/greengrass/definition/devices$": GreengrassResponse.method_dispatch( - GreengrassResponse.device_definitions - ), - "{0}/greengrass/definition/devices/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.device_definition - ), - "{0}/greengrass/definition/devices/(?P[^/]+)/versions$": GreengrassResponse.method_dispatch( - GreengrassResponse.device_definition_versions - ), - "{0}/greengrass/definition/devices/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.device_definition_version - ), - "{0}/greengrass/definition/functions$": GreengrassResponse.method_dispatch( - GreengrassResponse.function_definitions - ), - "{0}/greengrass/definition/functions/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.function_definition - ), - "{0}/greengrass/definition/functions/(?P[^/]+)/versions$": GreengrassResponse.method_dispatch( - GreengrassResponse.function_definition_versions - ), - "{0}/greengrass/definition/functions/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.function_definition_version - ), - "{0}/greengrass/definition/resources$": GreengrassResponse.method_dispatch( - GreengrassResponse.resource_definitions - ), - "{0}/greengrass/definition/resources/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.resource_definition - ), - "{0}/greengrass/definition/resources/(?P[^/]+)/versions$": GreengrassResponse.method_dispatch( - GreengrassResponse.resource_definition_versions - ), - "{0}/greengrass/definition/subscriptions$": GreengrassResponse.method_dispatch( - GreengrassResponse.subscription_definitions - ), - "{0}/greengrass/definition/subscriptions/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.subscription_definition - ), - "{0}/greengrass/definition/subscriptions/(?P[^/]+)/versions$": GreengrassResponse.method_dispatch( - GreengrassResponse.subscription_definition_versions - ), - "{0}/greengrass/definition/subscriptions/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.subscription_definition_version - ), - "{0}/greengrass/definition/resources/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.resource_definition_version - ), - "{0}/greengrass/groups$": GreengrassResponse.method_dispatch( - GreengrassResponse.groups - ), - "{0}/greengrass/groups/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.group - ), - "{0}/greengrass/groups/(?P[^/]+)/role$": GreengrassResponse.method_dispatch( - GreengrassResponse.role - ), - "{0}/greengrass/groups/(?P[^/]+)/versions$": GreengrassResponse.method_dispatch( - GreengrassResponse.group_versions - ), - "{0}/greengrass/groups/(?P[^/]+)/deployments$": GreengrassResponse.method_dispatch( - GreengrassResponse.deployments - ), - "{0}/greengrass/groups/(?P[^/]+)/deployments/\\$reset$": GreengrassResponse.method_dispatch( - GreengrassResponse.deployments_reset - ), - "{0}/greengrass/groups/(?P[^/]+)/deployments/(?P[^/]+)/status$": GreengrassResponse.method_dispatch( - GreengrassResponse.deployment_satus - ), - "{0}/greengrass/groups/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.method_dispatch( - GreengrassResponse.group_version - ), + "{0}/greengrass/definition/cores$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/cores/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/cores/(?P[^/]+)/versions$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/cores/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/devices$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/devices/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/devices/(?P[^/]+)/versions$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/devices/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/functions$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/functions/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/functions/(?P[^/]+)/versions$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/functions/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/resources$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/resources/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/resources/(?P[^/]+)/versions$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/subscriptions$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/subscriptions/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/subscriptions/(?P[^/]+)/versions$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/subscriptions/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/definition/resources/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/groups$": GreengrassResponse.dispatch, + "{0}/greengrass/groups/(?P[^/]+)/?$": GreengrassResponse.dispatch, + "{0}/greengrass/groups/(?P[^/]+)/role$": GreengrassResponse.dispatch, + "{0}/greengrass/groups/(?P[^/]+)/versions$": GreengrassResponse.dispatch, + "{0}/greengrass/groups/(?P[^/]+)/deployments$": GreengrassResponse.dispatch, + "{0}/greengrass/groups/(?P[^/]+)/deployments/\\$reset$": GreengrassResponse.dispatch, + "{0}/greengrass/groups/(?P[^/]+)/deployments/(?P[^/]+)/status$": GreengrassResponse.dispatch, + "{0}/greengrass/groups/(?P[^/]+)/versions/(?P[^/]+)/?$": GreengrassResponse.dispatch, }