Panorama: Add more features about applcations instances

moto/panorama/models.py

@@ -8,7 +8,11 @@
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.moto_api._internal.managed_state_model import ManagedState
-from moto.panorama.utils import deep_convert_datetime_to_isoformat, hash_device_name
+from moto.panorama.utils import (
+    arn_formatter,
+    deep_convert_datetime_to_isoformat,
+    hash_name,
+)
 from moto.utilities.paginator import paginate
 
 from .exceptions import (
@@ -22,6 +26,12 @@
         "limit_default": 123,
         "unique_attribute": "device_id",
     },
+    "list_application_instances": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 123,
+        "unique_attribute": "application_instance_id",
+    },
 }
 
 
@@ -94,7 +104,7 @@ def __init__(
         self.arn = (
             f"arn:aws:panorama:{self.region_name}:{self.account_id}:device/{self.name}"
         )
-        self.device_id = f"device-{hash_device_name(name)}"
+        self.device_id = f"device-{hash_name(name)}"
         self.iot_thing_name = ""
 
         self.alternate_softwares = [
@@ -221,10 +231,82 @@ def response_deleted(self) -> Dict[str, str]:
         return {"DeviceId": self.device_id}
 
 
+class ApplicationInstance(BaseObject):
+    def __init__(
+        self,
+        account_id: str,
+        region_name: str,
+        default_runtime_context_device: str,
+        default_runtime_context_device_name: str,
+        description: str,
+        manifest_overrides_payload: Dict[str, str],
+        manifest_payload: Dict[str, str],
+        name: str,
+        runtime_role_arn: str,
+        tags: Dict[str, str],
+    ) -> None:
+        self.default_runtime_context_device = default_runtime_context_device
+        self.default_runtime_context_device_name = default_runtime_context_device_name
+        self.description = description
+        self.manifest_overrides_payload = manifest_overrides_payload
+        self.manifest_payload = manifest_payload
+        self.name = name
+        self.runtime_role_arn = runtime_role_arn
+        self.tags = tags
+        now = datetime.now(tzutc())
+        self.created_time = now
+        self.last_updated_time = now
+        name = f"{self.name}-{self.created_time}"
+        self.application_instance_id = f"applicationInstance-{hash_name(name).lower()}"
+        self.arn = arn_formatter(
+            "application-instance",
+            self.application_instance_id,
+            account_id=account_id,
+            region_name=region_name,
+        )
+        self.health_status = "RUNNING"
+        self.status = "DEPLOYMENT_SUCCEEDED"
+        self.status_description = "string"
+        self.runtime_context_states = [
+            {
+                "DesiredState": "RUNNING",
+                "DeviceReportedStatus": "RUNNING",
+                "DeviceReportedTime": now,
+                "RuntimeContextName": "string",
+            },
+        ]
+
+    def add_new_runtime_context_states(
+        self, desired_state: str, device_reported_status: str
+    ) -> None:
+        now = datetime.now(tzutc())
+        self.runtime_context_states.append(
+            {
+                "DesiredState": desired_state,
+                "DeviceReportedStatus": device_reported_status,
+                "DeviceReportedTime": now,
+                "RuntimeContextName": "string",
+            }
+        )
+
+    def response_object(self) -> Dict[str, Any]:
+        response_object = super().gen_response_object()
+        response_object = deep_convert_datetime_to_isoformat(response_object)
+        return response_object
+
+    def response_created(self) -> Dict[str, str]:
+        return {"ApplicationInstanceId": self.application_instance_id}
+
+    def response_describe(self) -> Dict[str, str]:
+        response_object = self.response_object()
+        return response_object
+
+
 class PanoramaBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.devices_memory: Dict[str, Device] = {}
+        self.application_instances_memory: Dict[str, ApplicationInstance] = {}
 
     def provision_device(
         self,
@@ -291,6 +373,75 @@ def update_device_metadata(self, device_id: str, description: str) -> Device:
     def delete_device(self, device_id: str) -> Device:
         return self.devices_memory.pop(device_id)
 
+    def create_application_instance(
+        self,
+        application_instance_id_to_replace: Optional[str],
+        default_runtime_context_device: str,
+        description: str,
+        manifest_overrides_payload: Dict[str, str],
+        manifest_payload: Dict[str, str],
+        name: str,
+        runtime_role_arn: str,
+        tags: Dict[str, str],
+    ) -> ApplicationInstance:
+        device = self.devices_memory.get(default_runtime_context_device)
+        if device is None:
+            raise ValidationError(f"Device {default_runtime_context_device} not found")
+        if (
+            application_instance_id_to_replace
+            and application_instance_id_to_replace in self.application_instances_memory
+        ):
+            removed_application_instance = self.application_instances_memory[
+                application_instance_id_to_replace
+            ]
+            removed_application_instance.status = "REMOVAL_SUCCEEDED"
+            removed_application_instance.add_new_runtime_context_states(
+                desired_state="REMOVED", device_reported_status="REMOVAL_IN_PROGRESS"
+            )
+
+        application_instance = ApplicationInstance(
+            account_id=self.account_id,
+            region_name=self.region_name,
+            default_runtime_context_device=default_runtime_context_device,
+            default_runtime_context_device_name=device.name,
+            description=description,
+            manifest_overrides_payload=manifest_overrides_payload,
+            manifest_payload=manifest_payload,
+            name=name,
+            runtime_role_arn=runtime_role_arn,
+            tags=tags,
+        )
+        self.application_instances_memory[
+            application_instance.application_instance_id
+        ] = application_instance
+        return application_instance
+
+    def describe_application_instance(
+        self, application_instance_id: str
+    ) -> ApplicationInstance:
+        application_instance = self.application_instances_memory[
+            application_instance_id
+        ]
+        return application_instance
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_application_instances(
+        self,
+        device_id: Optional[str],
+        status_filter: Optional[str],
+    ) -> List[ApplicationInstance]:
+        filtered_application_instances = self.application_instances_memory.values()
+        if device_id:
+            filtered_application_instances = filter(
+                lambda x: x.default_runtime_context_device == device_id,
+                filtered_application_instances,
+            )
+        if status_filter:
+            filtered_application_instances = filter(
+                lambda x: x.status == status_filter, filtered_application_instances
+            )
+        return list(filtered_application_instances)
+
 
 panorama_backends = BackendDict(
     PanoramaBackend,

moto/panorama/responses.py

@@ -70,3 +70,59 @@ def delete_device(self) -> str:
         device_id = urllib.parse.unquote(self._get_param("DeviceId"))
         device = self.panorama_backend.delete_device(device_id=device_id)
         return json.dumps(device.response_deleted)
+
+    def create_application_instance(self) -> str:
+        application_instance_id_to_replace = self._get_param(
+            "ApplicationInstanceIdToReplace"
+        )
+        default_runtime_context_device = self._get_param("DefaultRuntimeContextDevice")
+        description = self._get_param("Description")
+        manifest_overrides_payload = self._get_param("ManifestOverridesPayload")
+        manifest_payload = self._get_param("ManifestPayload")
+        name = self._get_param("Name")
+        runtime_role_arn = self._get_param("RuntimeRoleArn")
+        tags = self._get_param("Tags")
+        application_instance = self.panorama_backend.create_application_instance(
+            application_instance_id_to_replace=application_instance_id_to_replace,
+            default_runtime_context_device=default_runtime_context_device,
+            description=description,
+            manifest_overrides_payload=manifest_overrides_payload,
+            manifest_payload=manifest_payload,
+            name=name,
+            runtime_role_arn=runtime_role_arn,
+            tags=tags,
+        )
+        return json.dumps(application_instance.response_created())
+
+    def describe_application_instance(self) -> str:
+        application_instance_id = self._get_param("ApplicationInstanceId")
+        application_instance = self.panorama_backend.describe_application_instance(
+            application_instance_id=application_instance_id
+        )
+        return json.dumps(application_instance.response_describe())
+
+    def describe_application_instance_details(self) -> str:
+        return self.describe_application_instance()
+
+    def list_application_instances(self) -> str:
+        device_id = self._get_param("deviceId")
+        max_results = self._get_int_param("maxResults")
+        status_filter = self._get_param("statusFilter")
+        next_token = self._get_param("nextToken")
+        list_application_instances, next_token = (
+            self.panorama_backend.list_application_instances(
+                device_id=device_id,
+                max_results=max_results,
+                status_filter=status_filter,
+                next_token=next_token,
+            )
+        )
+        return json.dumps(
+            {
+                "ApplicationInstances": [
+                    application_instance.response_describe()
+                    for application_instance in list_application_instances
+                ],
+                "NextToken": next_token,
+            }
+        )

moto/panorama/urls.py

@@ -8,4 +8,7 @@
     "{0}/$": PanoramaResponse.dispatch,
     "{0}/devices$": PanoramaResponse.dispatch,
     "{0}/devices/(?P<DeviceId>[^/]+)$": PanoramaResponse.dispatch,
+    "{0}/application-instances$": PanoramaResponse.dispatch,
+    "{0}/application-instances/(?P<ApplicationInstanceId>[^/]+)$": PanoramaResponse.dispatch,
+    "{0}/application-instances/(?P<ApplicationInstanceId>[^/]+)/details$": PanoramaResponse.dispatch,
 }

moto/panorama/utils.py

@@ -15,7 +15,11 @@ def deep_convert_datetime_to_isoformat(obj: Any) -> Any:
         return obj
 
 
-def hash_device_name(name: str) -> str:
+def hash_name(name: str) -> str:
     digest = hashlib.md5(name.encode("utf-8")).digest()
-    token = base64.b64encode(digest)
-    return token.decode("utf-8")
+    token = base64.urlsafe_b64encode(digest)
+    return token.decode("utf-8").rstrip("=")
+
+
+def arn_formatter(_type: str, _id: str, account_id: str, region_name: str) -> str:
+    return f"arn:aws:panorama:{region_name}:{account_id}:{_type}/{_id}"