Greengrass Implement function_definition APIs

moto/greengrass/models.py

@@ -167,6 +167,58 @@ def to_dict(self):
         }
 
 
+class FakeFunctionDefinition(BaseModel):
+    def __init__(self, region_name, name, initial_version):
+        self.region_name = region_name
+        self.id = str(uuid.uuid4())
+        self.arn = f"arn:aws:greengrass:{self.region_name}:{get_account_id()}:/greengrass/definition/functions/{self.id}"
+        self.created_at_datetime = datetime.utcnow()
+        self.update_at_datetime = datetime.utcnow()
+        self.latest_version = ""
+        self.latest_version_arn = ""
+        self.name = name
+        self.initial_version = initial_version
+
+    def to_dict(self):
+        res = {
+            "Arn": self.arn,
+            "CreationTimestamp": iso_8601_datetime_with_milliseconds(
+                self.created_at_datetime
+            ),
+            "Id": self.id,
+            "LastUpdatedTimestamp": iso_8601_datetime_with_milliseconds(
+                self.update_at_datetime
+            ),
+            "LatestVersion": self.latest_version,
+            "LatestVersionArn": self.latest_version_arn,
+        }
+        if self.name is not None:
+            res["Name"] = self.name
+        return res
+
+
+class FakeFunctionDefinitionVersion(BaseModel):
+    def __init__(self, region_name, function_definition_id, functions, default_config):
+        self.region_name = region_name
+        self.function_definition_id = function_definition_id
+        self.functions = functions
+        self.default_config = default_config
+        self.version = str(uuid.uuid4())
+        self.arn = f"arn:aws:greengrass:{self.region_name}:{get_account_id()}:/greengrass/definition/functions/{self.function_definition_id}/versions/{self.version}"
+        self.created_at_datetime = datetime.utcnow()
+
+    def to_dict(self):
+        return {
+            "Arn": self.arn,
+            "CreationTimestamp": iso_8601_datetime_with_milliseconds(
+                self.created_at_datetime
+            ),
+            "Definition": {"Functions": self.functions},
+            "Id": self.function_definition_id,
+            "Version": self.version,
+        }
+
+
 class GreengrassBackend(BaseBackend):
     def __init__(self, region_name, account_id):
         super().__init__(region_name, account_id)
@@ -410,5 +462,88 @@ def _validate_resources(resources):
                         f", but got: {device_source_path}])",
                     )
 
+    def create_function_definition(self, name, initial_version):
+        func_def = FakeFunctionDefinition(self.region_name, name, initial_version)
+        self.function_definitions[func_def.id] = func_def
+        init_ver = func_def.initial_version
+        init_func_def = init_ver.get("Functions", {})
+        init_config = init_ver.get("DefaultConfig", {})
+        self.create_function_definition_version(func_def.id, init_func_def, init_config)
+
+        return func_def
+
+    def list_function_definitions(self):
+        return self.function_definitions.values()
+
+    def get_function_definition(self, function_definition_id):
+
+        if function_definition_id not in self.function_definitions:
+            raise IdNotFoundException("That Lambda List Definition does not exist.")
+        return self.function_definitions[function_definition_id]
+
+    def delete_function_definition(self, function_definition_id):
+        if function_definition_id not in self.function_definitions:
+            raise IdNotFoundException("That lambdas definition does not exist.")
+        del self.function_definitions[function_definition_id]
+        del self.function_definition_versions[function_definition_id]
+
+    def update_function_definition(self, function_definition_id, name):
+
+        if name == "":
+            raise InvalidContainerDefinitionException(
+                "Input does not contain any attributes to be updated"
+            )
+        if function_definition_id not in self.function_definitions:
+            raise IdNotFoundException("That lambdas definition does not exist.")
+        self.function_definitions[function_definition_id].name = name
+
+    def create_function_definition_version(
+        self, function_definition_id, functions, default_config
+    ):
+
+        if function_definition_id not in self.function_definitions:
+            raise IdNotFoundException("That lambdas does not exist.")
+
+        func_ver = FakeFunctionDefinitionVersion(
+            self.region_name, function_definition_id, functions, default_config
+        )
+        func_vers = self.function_definition_versions.get(
+            func_ver.function_definition_id, {}
+        )
+        func_vers[func_ver.version] = func_ver
+        self.function_definition_versions[func_ver.function_definition_id] = func_vers
+        self.function_definitions[
+            function_definition_id
+        ].latest_version = func_ver.version
+        self.function_definitions[
+            function_definition_id
+        ].latest_version_arn = func_ver.arn
+
+        return func_ver
+
+    def list_function_definition_versions(self, function_definition_id):
+        if function_definition_id not in self.function_definition_versions:
+            raise IdNotFoundException("That lambdas definition does not exist.")
+        return self.function_definition_versions[function_definition_id]
+
+    def get_function_definition_version(
+        self, function_definition_id, function_definition_version_id
+    ):
+
+        if function_definition_id not in self.function_definition_versions:
+            raise IdNotFoundException("That lambdas definition does not exist.")
+
+        if (
+            function_definition_version_id
+            not in self.function_definition_versions[function_definition_id]
+        ):
+            raise IdNotFoundException(
+                f"Version {function_definition_version_id} of Lambda List Definition {function_definition_id} does not exist."
+            )
+
+        return self.function_definition_versions[function_definition_id][
+            function_definition_version_id
+        ]
+
 
 greengrass_backends = BackendDict(GreengrassBackend, "greengrass")

moto/greengrass/responses.py

@@ -260,3 +260,110 @@ def create_resource_definition_version(self):
             resource_definition_id=resource_definition_id, resources=resources
         )
         return 201, {"status": 201}, json.dumps(res.to_dict())
+
+    def function_definitions(self, request, full_url, headers):
+        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()
+
+    def create_function_definition(self):
+
+        initial_version = self._get_param("InitialVersion")
+        name = self._get_param("Name")
+        res = self.greengrass_backend.create_function_definition(
+            name=name, initial_version=initial_version
+        )
+        return 201, {"status": 201}, json.dumps(res.to_dict())
+
+    def list_function_definitions(self):
+        res = self.greengrass_backend.list_function_definitions()
+        return (
+            200,
+            {"status": 200},
+            json.dumps(
+                {"Definitions": [func_definition.to_dict() for func_definition in res]}
+            ),
+        )
+
+    def function_definition(self, request, full_url, headers):
+        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):
+        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())
+
+    def delete_function_definition(self):
+        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({})
+
+    def update_function_definition(self):
+        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(self, request, full_url, headers):
+        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()
+
+    def create_function_definition_version(self):
+
+        default_config = self._get_param("DefaultConfig")
+        function_definition_id = self.path.split("/")[-2]
+        functions = self._get_param("Functions")
+
+        res = self.greengrass_backend.create_function_definition_version(
+            default_config=default_config,
+            function_definition_id=function_definition_id,
+            functions=functions,
+        )
+        return 201, {"status": 201}, json.dumps(res.to_dict())
+
+    def list_function_definition_versions(self):
+        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(self, request, full_url, headers):
+        self.setup_class(request, full_url, headers)
+
+        if self.method == "GET":
+            return self.get_function_definition_version()
+
+    def get_function_definition_version(self):
+        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())

moto/greengrass/urls.py

@@ -17,6 +17,10 @@
     "{0}/greengrass/definition/devices/(?P<definition_id>[^/]+)/?$": response.device_definition,
     "{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/functions$": response.function_definitions,
+    "{0}/greengrass/definition/functions/(?P<definition_id>[^/]+)/?$": response.function_definition,
+    "{0}/greengrass/definition/functions/(?P<definition_id>[^/]+)/versions$": response.function_definition_versions,
+    "{0}/greengrass/definition/functions/(?P<definition_id>[^/]+)/versions/(?P<definition_version_id>[^/]+)/?$": response.function_definition_version,
     "{0}/greengrass/definition/resources$": response.resource_definitions,
     "{0}/greengrass/definition/resources/(?P<definition_id>[^/]+)/versions$": response.resource_definition_versions,
 }