Panorama: Add mock_panorama (#6948)

IMPLEMENTATION_COVERAGE.md

@@ -5173,6 +5173,46 @@
 - [X] update_policy
 </details>
 
+## panorama
+<details>
+<summary>14% implemented</summary>
+
+- [ ] create_application_instance
+- [ ] create_job_for_devices
+- [ ] create_node_from_template_job
+- [ ] create_package
+- [ ] create_package_import_job
+- [X] delete_device
+- [ ] delete_package
+- [ ] deregister_package_version
+- [ ] describe_application_instance
+- [ ] describe_application_instance_details
+- [X] describe_device
+- [ ] describe_device_job
+- [ ] describe_node
+- [ ] describe_node_from_template_job
+- [ ] describe_package
+- [ ] describe_package_import_job
+- [ ] describe_package_version
+- [ ] list_application_instance_dependencies
+- [ ] list_application_instance_node_instances
+- [ ] list_application_instances
+- [X] list_devices
+- [ ] list_devices_jobs
+- [ ] list_node_from_template_jobs
+- [ ] list_nodes
+- [ ] list_package_import_jobs
+- [ ] list_packages
+- [ ] list_tags_for_resource
+- [X] provision_device
+- [ ] register_package_version
+- [ ] remove_application_instance
+- [ ] signal_application_instance_node_instances
+- [ ] tag_resource
+- [ ] untag_resource
+- [X] update_device_metadata
+</details>
+
 ## personalize
 <details>
 <summary>5% implemented</summary>

docs/docs/services/ec2.rst

@@ -600,9 +600,9 @@ ec2
 - [X] modify_vpc_endpoint
 - [ ] modify_vpc_endpoint_connection_notification
 - [X] modify_vpc_endpoint_service_configuration
-  
+
         The following parameters are not yet implemented: RemovePrivateDnsName
-        
+
 
 - [ ] modify_vpc_endpoint_service_payer_responsibility
 - [X] modify_vpc_endpoint_service_permissions
@@ -662,7 +662,7 @@ ec2
 - [X] revoke_security_group_egress
 - [X] revoke_security_group_ingress
 - [X] run_instances
-  
+
         The Placement-parameter is validated to verify the availability-zone exists for the current region.
 
         The InstanceType-parameter can be validated, to see if it is a known instance-type.
@@ -673,15 +673,15 @@ ec2
 
         The KeyPair-parameter can be validated, to see if it is a known key-pair.
         Enable this validation by setting the environment variable `MOTO_ENABLE_KEYPAIR_VALIDATION=true`
-        
+
 
 - [ ] run_scheduled_instances
 - [ ] search_local_gateway_routes
 - [ ] search_transit_gateway_multicast_groups
 - [X] search_transit_gateway_routes
-  
+
         The following filters are currently supported: type, state, route-search.exact-match
-        
+
 
 - [ ] send_diagnostic_interrupt
 - [X] start_instances
@@ -699,4 +699,3 @@ ec2
 - [X] update_security_group_rule_descriptions_egress
 - [X] update_security_group_rule_descriptions_ingress
 - [ ] withdraw_byoip_cidr
-

docs/docs/services/panorama.rst

@@ -0,0 +1,62 @@
+.. _implementedservice_panorama:
+
+.. |start-h3| raw:: html
+
+    <h3>
+
+.. |end-h3| raw:: html
+
+    </h3>
+
+========
+panorama
+========
+
+|start-h3| Example usage |end-h3|
+
+.. sourcecode:: python
+
+            @mock_panorama
+            def test_panorama_behaviour:
+                boto3.client("panorama")
+                ...
+
+
+
+|start-h3| Implemented features for this service |end-h3|
+
+- [ ] create_application_instance
+- [ ] create_job_for_devices
+- [ ] create_node_from_template_job
+- [ ] create_package
+- [ ] create_package_import_job
+- [X] delete_device
+- [ ] delete_package
+- [ ] deregister_package_version
+- [ ] describe_application_instance
+- [ ] describe_application_instance_details
+- [X] describe_device
+- [ ] describe_device_job
+- [ ] describe_node
+- [ ] describe_node_from_template_job
+- [ ] describe_package
+- [ ] describe_package_import_job
+- [ ] describe_package_version
+- [ ] list_application_instance_dependencies
+- [ ] list_application_instance_node_instances
+- [ ] list_application_instances
+- [X] list_devices
+- [ ] list_devices_jobs
+- [ ] list_node_from_template_jobs
+- [ ] list_nodes
+- [ ] list_package_import_jobs
+- [ ] list_packages
+- [ ] list_tags_for_resource
+- [X] provision_device
+- [ ] register_package_version
+- [ ] remove_application_instance
+- [ ] signal_application_instance_node_instances
+- [ ] tag_resource
+- [ ] untag_resource
+- [X] update_device_metadata
+

docs/docs/services/textract.rst

@@ -39,9 +39,9 @@ textract
 - [ ] get_adapter_version
 - [ ] get_document_analysis
 - [X] get_document_text_detection
-  
+
         Pagination has not yet been implemented
-        
+
 
 - [ ] get_expense_analysis
 - [ ] get_lending_analysis
@@ -51,13 +51,12 @@ textract
 - [ ] list_tags_for_resource
 - [ ] start_document_analysis
 - [X] start_document_text_detection
-  
+
         The following parameters have not yet been implemented: ClientRequestToken, JobTag, NotificationChannel, OutputConfig, KmsKeyID
-        
+
 
 - [ ] start_expense_analysis
 - [ ] start_lending_analysis
 - [ ] tag_resource
 - [ ] untag_resource
 - [ ] update_adapter
-

moto/__init__.py

@@ -178,6 +178,7 @@ mock_neptune = lazy_load(".rds", "mock_rds", boto3_name="neptune")
 mock_opensearch = lazy_load(".opensearch", "mock_opensearch")
 mock_opsworks = lazy_load(".opsworks", "mock_opsworks")
 mock_organizations = lazy_load(".organizations", "mock_organizations")
+mock_panorama = lazy_load(".panorama", "mock_panorama")
 mock_personalize = lazy_load(".personalize", "mock_personalize")
 mock_pinpoint = lazy_load(".pinpoint", "mock_pinpoint")
 mock_polly = lazy_load(".polly", "mock_polly")

moto/backend_index.py

@@ -122,6 +122,7 @@ backend_url_patterns = [
     ("mq", re.compile("https?://mq\\.(.+)\\.amazonaws\\.com")),
     ("opsworks", re.compile("https?://opsworks\\.us-east-1\\.amazonaws.com")),
     ("organizations", re.compile("https?://organizations\\.(.+)\\.amazonaws\\.com")),
+    ("panorama", re.compile("https?://panorama\\.(.+)\\.amazonaws.com")),
     ("personalize", re.compile("https?://personalize\\.(.+)\\.amazonaws\\.com")),
     ("pinpoint", re.compile("https?://pinpoint\\.(.+)\\.amazonaws\\.com")),
     ("polly", re.compile("https?://polly\\.(.+)\\.amazonaws.com")),

moto/panorama/__init__.py

@@ -0,0 +1,4 @@
+from ..core.models import base_decorator
+from .models import panorama_backends
+
+mock_panorama = base_decorator(panorama_backends)

moto/panorama/exceptions.py

@@ -0,0 +1,6 @@
+from moto.core.exceptions import JsonRESTError
+
+
+class ValidationError(JsonRESTError):
+    def __init__(self, message: str):
+        super().__init__("ValidationException", message)

moto/panorama/models.py

@@ -0,0 +1,306 @@
+import base64
+import json
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Optional, Union
+
+from dateutil.tz import tzutc
+
+from moto.core import BackendDict, BaseBackend, 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.utilities.paginator import paginate
+
+from .exceptions import (
+    ValidationError,
+)
+
+PAGINATION_MODEL = {
+    "list_devices": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 123,
+        "unique_attribute": "device_id",
+    },
+}
+
+
+class BaseObject(BaseModel):
+    def camelCase(self, key: str) -> str:
+        words = []
+        for word in key.split("_"):
+            words.append(word.title())
+        return "".join(words)
+
+    def update(self, details_json: str) -> None:
+        details = json.loads(details_json)
+        for k in details.keys():
+            setattr(self, k, details[k])
+
+    def gen_response_object(self) -> Dict[str, Any]:
+        response_object: Dict[str, Any] = dict()
+        for key, value in self.__dict__.items():
+            if "_" in key:
+                response_object[self.camelCase(key)] = value
+            else:
+                response_object[key[0].upper() + key[1:]] = value
+        return response_object
+
+    def response_object(self) -> Dict[str, Any]:  # type: ignore[misc]
+        return self.gen_response_object()
+
+
+class Device(BaseObject):
+    def __init__(
+        self,
+        account_id: str,
+        region_name: str,
+        description: Optional[str],
+        name: str,
+        network_configuration: Optional[Dict[str, Any]],
+        tags: Optional[Dict[str, str]],
+    ) -> None:
+        # ManagedState is a class that helps us manage the state of a resource.
+        # A Panorama Device has a lot of different states that has their own lifecycle.
+        # To make all this states in the same object, and avoid changing ManagedState,
+        # we use a ManagedState for each state and we manage them in the Device class.
+        # Each ManagedState has a name composed with attribute name and device name to make subscription easier.
+        self.__device_aggregated_status_manager = ManagedState(
+            model_name=f"panorama::device_{name}_aggregated_status",
+            transitions=[
+                ("NOT-A-STATUS", "AWAITING_PROVISIONING"),
+                ("AWAITING_PROVISIONING", "PENDING"),
+                ("PENDING", "ONLINE"),
+            ],
+        )
+        self.__device_provisioning_status_manager = ManagedState(
+            model_name=f"panorama::device_{name}_provisioning_status",
+            transitions=[
+                ("NOT-A-STATUS", "AWAITING_PROVISIONING"),
+                ("AWAITING_PROVISIONING", "PENDING"),
+                ("PENDING", "SUCCEEDED"),
+            ],
+        )
+        self.account_id = account_id
+        self.region_name = region_name
+        self.description = description
+        self.name = name
+        self.network_configuration = network_configuration
+        self.tags = tags
+
+        self.certificates = base64.b64encode("certificate".encode("utf-8")).decode(
+            "utf-8"
+        )
+        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.iot_thing_name = ""
+
+        self.alternate_softwares = [
+            {"Version": "0.2.1"},
+        ]
+        self.brand: str = "AWS_PANORAMA"  # AWS_PANORAMA | LENOVO
+        self.created_time = datetime.now(tzutc())
+        self.last_updated_time = datetime.now(tzutc())
+        self.current_networking_status = {
+            "Ethernet0Status": {
+                "ConnectionStatus": "CONNECTED",
+                "HwAddress": "8C:0F:5F:60:F5:C4",
+                "IpAddress": "192.168.1.300/24",
+            },
+            "Ethernet1Status": {
+                "ConnectionStatus": "NOT_CONNECTED",
+                "HwAddress": "8C:0F:6F:60:F4:F1",
+                "IpAddress": "--",
+            },
+            "LastUpdatedTime": datetime.now(tzutc()),
+            "NtpStatus": {
+                "ConnectionStatus": "CONNECTED",
+                "IpAddress": "91.224.149.41:123",
+                "NtpServerName": "0.pool.ntp.org",
+            },
+        }
+        self.current_software = "6.2.1"
+        self.device_connection_status: str = "ONLINE"  # "ONLINE"|"OFFLINE"|"AWAITING_CREDENTIALS"|"NOT_AVAILABLE"|"ERROR"
+        self.latest_device_job = {"JobType": "REBOOT", "Status": "COMPLETED"}
+        self.latest_software = "6.2.1"
+        self.lease_expiration_time = datetime.now(tzutc()) + timedelta(days=5)
+        self.serial_number = "GAD81E29013274749"
+        self.type: str = "PANORAMA_APPLIANCE"  # "PANORAMA_APPLIANCE_DEVELOPER_KIT", "PANORAMA_APPLIANCE"
+
+    @property
+    def device_aggregated_status(self) -> str:
+        self.__device_aggregated_status_manager.advance()
+        return self.__device_aggregated_status_manager.status  # type: ignore[return-value]
+
+    @property
+    def provisioning_status(self) -> str:
+        self.__device_provisioning_status_manager.advance()
+        return self.__device_provisioning_status_manager.status  # type: ignore[return-value]
+
+    def response_object(self) -> Dict[str, Any]:
+        response_object = super().gen_response_object()
+        response_object = deep_convert_datetime_to_isoformat(response_object)
+        static_response_fields = [
+            "AlternateSoftwares",
+            "Arn",
+            "Brand",
+            "CreatedTime",
+            "CurrentNetworkingStatus",
+            "CurrentSoftware",
+            "Description",
+            "DeviceConnectionStatus",
+            "DeviceId",
+            "LatestAlternateSoftware",
+            "LatestDeviceJob",
+            "LatestSoftware",
+            "LeaseExpirationTime",
+            "Name",
+            "NetworkConfiguration",
+            "SerialNumber",
+            "Tags",
+            "Type",
+        ]
+        return {
+            **{
+                k: v
+                for k, v in response_object.items()
+                if v is not None and k in static_response_fields
+            },
+            **{
+                "DeviceAggregatedStatus": self.device_aggregated_status,
+                "ProvisioningStatus": self.provisioning_status,
+            },
+        }
+
+    def response_listed(self) -> Dict[str, Any]:
+        response_object = super().gen_response_object()
+        response_object = deep_convert_datetime_to_isoformat(response_object)
+        static_response_fields = [
+            "Brand",
+            "CreatedTime",
+            "CurrentSoftware",
+            "Description",
+            "DeviceId",
+            "LastUpdatedTime",
+            "LatestDeviceJob",
+            "LeaseExpirationTime",
+            "Name",
+            "Tags",
+            "Type",
+        ]
+        return {
+            **{
+                k: v
+                for k, v in response_object.items()
+                if v is not None and k in static_response_fields
+            },
+            **{
+                "DeviceAggregatedStatus": self.device_aggregated_status,
+                "ProvisioningStatus": self.provisioning_status,
+            },
+        }
+
+    @property
+    def response_provision(self) -> Dict[str, Union[str, bytes]]:
+        return {
+            "Arn": self.arn,
+            "Certificates": self.certificates,
+            "DeviceId": self.device_id,
+            "IotThingName": self.iot_thing_name,
+            "Status": self.provisioning_status,
+        }
+
+    @property
+    def response_updated(self) -> Dict[str, str]:
+        return {"DeviceId": self.device_id}
+
+    @property
+    def response_deleted(self) -> Dict[str, str]:
+        return {"DeviceId": self.device_id}
+
+
+class PanoramaBackend(BaseBackend):
+    def __init__(self, region_name: str, account_id: str):
+        super().__init__(region_name, account_id)
+        self.devices_memory: Dict[str, Device] = {}
+
+    def provision_device(
+        self,
+        description: Optional[str],
+        name: str,
+        networking_configuration: Optional[Dict[str, Any]],
+        tags: Optional[Dict[str, str]],
+    ) -> Device:
+        device_obj = Device(
+            account_id=self.account_id,
+            region_name=self.region_name,
+            description=description,
+            name=name,
+            network_configuration=networking_configuration,
+            tags=tags,
+        )
+
+        self.devices_memory[device_obj.device_id] = device_obj
+        return device_obj
+
+    def describe_device(self, device_id: str) -> Device:
+        device = self.devices_memory.get(device_id)
+        if device is None:
+            raise ValidationError(f"Device {device_id} not found")
+        return device
+
+    @paginate(pagination_model=PAGINATION_MODEL)  # type: ignore[misc]
+    def list_devices(
+        self,
+        device_aggregated_status_filter: str,
+        name_filter: str,
+        sort_by: str,  # "DEVICE_ID", "CREATED_TIME", "NAME", "DEVICE_AGGREGATED_STATUS"
+        sort_order: str,  # "ASCENDING", "DESCENDING"
+    ) -> List[Device]:
+        devices_list = list(
+            filter(
+                lambda x: (name_filter is None or x.name.startswith(name_filter))
+                and (
+                    device_aggregated_status_filter is None
+                    or x.device_aggregated_status == device_aggregated_status_filter
+                ),
+                self.devices_memory.values(),
+            )
+        )
+        devices_list = list(
+            sorted(
+                devices_list,
+                key={
+                    "DEVICE_ID": lambda x: x.device_id,
+                    "CREATED_TIME": lambda x: x.created_time,
+                    "NAME": lambda x: x.name,
+                    "DEVICE_AGGREGATED_STATUS": lambda x: x.device_aggregated_status,
+                    None: lambda x: x.created_time,
+                }[sort_by],
+                reverse=sort_order == "DESCENDING",
+            )
+        )
+        return devices_list
+
+    def update_device_metadata(self, device_id: str, description: str) -> Device:
+        self.devices_memory[device_id].description = description
+        return self.devices_memory[device_id]
+
+    def delete_device(self, device_id: str) -> Device:
+        return self.devices_memory.pop(device_id)
+
+
+panorama_backends = BackendDict(
+    PanoramaBackend,
+    "panorama",
+    False,
+    additional_regions=[
+        "us-east-1",
+        "us-west-2",
+        "ca-central-1",
+        "eu-west-1",
+        "ap-southeast-2",
+        "ap-southeast-1",
+    ],
+)

moto/panorama/responses.py

@@ -0,0 +1,72 @@
+import json
+import urllib
+
+from moto.core.responses import BaseResponse
+
+from .models import PanoramaBackend, panorama_backends
+
+
+class PanoramaResponse(BaseResponse):
+    def __init__(self) -> None:
+        super().__init__(service_name="panorama")
+
+    @property
+    def panorama_backend(self) -> PanoramaBackend:
+        return panorama_backends[self.current_account][self.region]
+
+    def provision_device(self) -> str:
+        description = self._get_param("Description")
+        name = self._get_param("Name")
+        networking_configuration = self._get_param("NetworkingConfiguration")
+        tags = self._get_param("Tags")
+        device = self.panorama_backend.provision_device(
+            description=description,
+            name=name,
+            networking_configuration=networking_configuration,
+            tags=tags,
+        )
+        return json.dumps(device.response_provision)
+
+    def describe_device(self) -> str:
+        device_id = urllib.parse.unquote(self._get_param("DeviceId"))
+        device = self.panorama_backend.describe_device(device_id=device_id)
+        return json.dumps(device.response_object())
+
+    def list_devices(
+        self,
+    ) -> str:
+        device_aggregated_status_filter = self._get_param(
+            "DeviceAggregatedStatusFilter"
+        )
+        max_results = self._get_int_param("MaxResults")
+        name_filter = self._get_param("NameFilter")
+        next_token = self._get_param("NextToken")
+        sort_by = self._get_param("SortBy")
+        sort_order = self._get_param("SortOrder")
+        list_devices, next_token = self.panorama_backend.list_devices(
+            device_aggregated_status_filter=device_aggregated_status_filter,
+            max_results=max_results,
+            name_filter=name_filter,
+            next_token=next_token,
+            sort_by=sort_by,
+            sort_order=sort_order,
+        )
+        return json.dumps(
+            {
+                "Devices": [device.response_listed() for device in list_devices],
+                "NextToken": next_token,
+            }
+        )
+
+    def update_device_metadata(self) -> str:
+        device_id = urllib.parse.unquote(self._get_param("DeviceId"))
+        description = self._get_param("Description")
+        device = self.panorama_backend.update_device_metadata(
+            device_id=device_id, description=description
+        )
+        return json.dumps(device.response_updated)
+
+    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)

moto/panorama/urls.py

@@ -0,0 +1,11 @@
+from .responses import PanoramaResponse
+
+url_bases = [
+    r"https?://panorama\.(.+)\.amazonaws.com",
+]
+
+url_paths = {
+    "{0}/$": PanoramaResponse.dispatch,
+    "{0}/devices$": PanoramaResponse.dispatch,
+    "{0}/devices/(?P<DeviceId>[^/]+)$": PanoramaResponse.dispatch,
+}

moto/panorama/utils.py

@@ -0,0 +1,21 @@
+import base64
+import hashlib
+from datetime import datetime
+from typing import Any
+
+
+def deep_convert_datetime_to_isoformat(obj: Any) -> Any:
+    if isinstance(obj, datetime):
+        return obj.isoformat()
+    elif isinstance(obj, list):
+        return [deep_convert_datetime_to_isoformat(x) for x in obj]
+    elif isinstance(obj, dict):
+        return {k: deep_convert_datetime_to_isoformat(v) for k, v in obj.items()}
+    else:
+        return obj
+
+
+def hash_device_name(name: str) -> str:
+    digest = hashlib.md5(name.encode("utf-8")).digest()
+    token = base64.b64encode(digest)
+    return token.decode("utf-8")

tests/test_panorama/test_panorama_device.py

@@ -0,0 +1,435 @@
+from datetime import datetime
+from typing import List
+from unittest import SkipTest
+
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+from dateutil.tz import tzutc
+from freezegun import freeze_time
+
+from moto import mock_panorama, settings
+from moto.moto_api import state_manager
+
+
+@mock_panorama
+def test_provision_device() -> None:
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't use ManagedState in ServerMode")
+    client = boto3.client("panorama", region_name="eu-west-1")
+    given_device_name = "test-device-name"
+    state_manager.set_transition(
+        model_name=f"panorama::device_{given_device_name}_provisioning_status",
+        transition={"progression": "manual", "times": 1},
+    )
+    resp = client.provision_device(
+        Description=given_device_name,
+        Name="test-device-name",
+        NetworkingConfiguration={
+            "Ethernet0": {
+                "ConnectionType": "STATIC_IP",
+                "StaticIpConnectionInfo": {
+                    "DefaultGateway": "192.168.1.1",
+                    "Dns": [
+                        "8.8.8.8",
+                    ],
+                    "IpAddress": "192.168.1.10",
+                    "Mask": "255.255.255.0",
+                },
+            },
+            "Ethernet1": {
+                "ConnectionType": "dhcp",
+            },
+            "Ntp": {
+                "NtpServers": [
+                    "0.pool.ntp.org",
+                    "1.pool.ntp.org",
+                    "0.fr.pool.ntp.org",
+                ]
+            },
+        },
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+    assert (
+        resp["Arn"] == "arn:aws:panorama:eu-west-1:123456789012:device/test-device-name"
+    )
+    assert resp["Certificates"] == b"certificate"
+    assert resp["DeviceId"] == "device-RsozEWjZpeNe3SXHidX3mg=="
+    assert resp["IotThingName"] == ""
+    assert resp["Status"] == "AWAITING_PROVISIONING"
+
+
+@mock_panorama
+def test_describe_device() -> None:
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't freeze time in ServerMode")
+    client = boto3.client("panorama", region_name="eu-west-1")
+    given_device_name = "test-device-name"
+    state_manager.set_transition(
+        model_name=f"panorama::device_{given_device_name}_provisioning_status",
+        transition={"progression": "immediate"},
+    )
+    with freeze_time("2020-01-01 12:00:00"):
+        resp = client.provision_device(
+            Description="test device description",
+            Name=given_device_name,
+            NetworkingConfiguration={
+                "Ethernet0": {
+                    "ConnectionType": "STATIC_IP",
+                    "StaticIpConnectionInfo": {
+                        "DefaultGateway": "192.168.1.1",
+                        "Dns": [
+                            "8.8.8.8",
+                        ],
+                        "IpAddress": "192.168.1.10",
+                        "Mask": "255.255.255.0",
+                    },
+                },
+                "Ethernet1": {
+                    "ConnectionType": "dhcp",
+                },
+                "Ntp": {
+                    "NtpServers": [
+                        "0.pool.ntp.org",
+                        "1.pool.ntp.org",
+                        "0.fr.pool.ntp.org",
+                    ]
+                },
+            },
+            Tags={"Key": "test-key", "Value": "test-value"},
+        )
+
+    resp = client.describe_device(DeviceId=resp["DeviceId"])
+
+    assert resp["AlternateSoftwares"] == [{"Version": "0.2.1"}]
+    assert (
+        resp["Arn"] == "arn:aws:panorama:eu-west-1:123456789012:device/test-device-name"
+    )
+    assert resp["Brand"] == "AWS_PANORAMA"
+    assert resp["CreatedTime"] == datetime(2020, 1, 1, 12, 0, tzinfo=tzutc())
+    assert resp["CurrentNetworkingStatus"] == {
+        "Ethernet0Status": {
+            "ConnectionStatus": "CONNECTED",
+            "HwAddress": "8C:0F:5F:60:F5:C4",
+            "IpAddress": "192.168.1.300/24",
+        },
+        "Ethernet1Status": {
+            "ConnectionStatus": "NOT_CONNECTED",
+            "HwAddress": "8C:0F:6F:60:F4:F1",
+            "IpAddress": "--",
+        },
+        "LastUpdatedTime": datetime(2020, 1, 1, 12, 0, tzinfo=tzutc()),
+        "NtpStatus": {
+            "ConnectionStatus": "CONNECTED",
+            "IpAddress": "91.224.149.41:123",
+            "NtpServerName": "0.pool.ntp.org",
+        },
+    }
+    assert resp["CurrentSoftware"] == "6.2.1"
+    assert resp["Description"] == "test device description"
+    assert resp["DeviceAggregatedStatus"] == "ONLINE"
+    assert resp["DeviceConnectionStatus"] == "ONLINE"
+    assert resp["DeviceId"] == "device-RsozEWjZpeNe3SXHidX3mg=="
+    assert resp["LatestDeviceJob"] == {"JobType": "REBOOT", "Status": "COMPLETED"}
+    assert resp["LatestSoftware"] == "6.2.1"
+    assert resp["LeaseExpirationTime"] == datetime(2020, 1, 6, 12, 0, tzinfo=tzutc())
+    assert resp["Name"] == "test-device-name"
+    assert resp["ProvisioningStatus"] == "SUCCEEDED"
+    assert resp["SerialNumber"] == "GAD81E29013274749"
+    assert resp["Tags"] == {"Key": "test-key", "Value": "test-value"}
+    assert resp["Type"] == "PANORAMA_APPLIANCE"
+
+
+@mock_panorama
+def test_provision_device_aggregated_status_lifecycle() -> None:
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't use ManagedState in ServerMode")
+    client = boto3.client("panorama", region_name="eu-west-1")
+    given_device_name = "test-device-name"
+    state_manager.set_transition(
+        model_name=f"panorama::device_{given_device_name}_aggregated_status",
+        transition={"progression": "manual", "times": 1},
+    )
+    state_manager.set_transition(
+        model_name=f"panorama::device_{given_device_name}_provisioning_status",
+        transition={"progression": "manual", "times": 2},
+    )
+    device_id = client.provision_device(
+        Description="test device description",
+        Name=given_device_name,
+        NetworkingConfiguration={
+            "Ethernet0": {
+                "ConnectionType": "STATIC_IP",
+                "StaticIpConnectionInfo": {
+                    "DefaultGateway": "192.168.1.1",
+                    "Dns": [
+                        "8.8.8.8",
+                    ],
+                    "IpAddress": "192.168.1.10",
+                    "Mask": "255.255.255.0",
+                },
+            },
+            "Ethernet1": {
+                "ConnectionType": "dhcp",
+            },
+            "Ntp": {
+                "NtpServers": [
+                    "0.pool.ntp.org",
+                    "1.pool.ntp.org",
+                    "0.fr.pool.ntp.org",
+                ]
+            },
+        },
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )["DeviceId"]
+
+    resp_1 = client.describe_device(DeviceId=device_id)
+    assert (
+        resp_1["Arn"]
+        == "arn:aws:panorama:eu-west-1:123456789012:device/test-device-name"
+    )
+    assert resp_1["DeviceAggregatedStatus"] == "AWAITING_PROVISIONING"
+    assert resp_1["ProvisioningStatus"] == "AWAITING_PROVISIONING"
+
+    resp_2 = client.describe_device(DeviceId=device_id)
+    assert (
+        resp_2["Arn"]
+        == "arn:aws:panorama:eu-west-1:123456789012:device/test-device-name"
+    )
+    assert resp_2["DeviceAggregatedStatus"] == "PENDING"
+    assert resp_2["ProvisioningStatus"] == "AWAITING_PROVISIONING"
+
+    resp_3 = client.describe_device(DeviceId=device_id)
+    assert (
+        resp_3["Arn"]
+        == "arn:aws:panorama:eu-west-1:123456789012:device/test-device-name"
+    )
+    assert resp_3["DeviceAggregatedStatus"] == "ONLINE"
+    assert resp_3["ProvisioningStatus"] == "PENDING"
+
+
+@mock_panorama
+def test_list_device() -> None:
+    client = boto3.client("panorama", region_name="eu-west-1")
+    resp_1 = client.provision_device(
+        Description="test device description 1",
+        Name="test-device-name-1",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+    resp_2 = client.provision_device(
+        Description="test device description 2",
+        Name="test-device-name-2",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+
+    resp = client.list_devices()
+
+    assert len(resp["Devices"]) == 2
+    assert "Brand" in resp["Devices"][0]
+    assert "CreatedTime" in resp["Devices"][0]
+    assert "CurrentSoftware" in resp["Devices"][0]
+    assert "Description" in resp["Devices"][0]
+    assert "DeviceAggregatedStatus" in resp["Devices"][0]
+    assert "DeviceId" in resp["Devices"][0]
+    assert "LastUpdatedTime" in resp["Devices"][0]
+    assert "LatestDeviceJob" in resp["Devices"][0]
+    assert "LeaseExpirationTime" in resp["Devices"][0]
+    assert "Name" in resp["Devices"][0]
+    assert "ProvisioningStatus" in resp["Devices"][0]
+    assert "Tags" in resp["Devices"][0]
+    assert "Type" in resp["Devices"][0]
+    assert resp["Devices"][0]["DeviceId"] == resp_1["DeviceId"]
+
+    assert "Brand" in resp["Devices"][1]
+    assert "CreatedTime" in resp["Devices"][1]
+    assert "CurrentSoftware" in resp["Devices"][1]
+    assert "Description" in resp["Devices"][1]
+    assert "DeviceAggregatedStatus" in resp["Devices"][1]
+    assert "DeviceId" in resp["Devices"][1]
+    assert "LastUpdatedTime" in resp["Devices"][1]
+    assert "LatestDeviceJob" in resp["Devices"][1]
+    assert "LeaseExpirationTime" in resp["Devices"][1]
+    assert "Name" in resp["Devices"][1]
+    assert "ProvisioningStatus" in resp["Devices"][1]
+    assert "Tags" in resp["Devices"][1]
+    assert "Type" in resp["Devices"][1]
+    assert resp["Devices"][1]["DeviceId"] == resp_2["DeviceId"]
+
+
+@mock_panorama
+def test_list_device_name_filter() -> None:
+    client = boto3.client("panorama", region_name="eu-west-1")
+    resp_1 = client.provision_device(
+        Description="test device description 1",
+        Name="test-device-name-1",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+    resp_2 = client.provision_device(
+        Description="test device description 2",
+        Name="test-device-name-2",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+    _ = client.provision_device(
+        Description="test device description 3",
+        Name="another-test-device-name",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+
+    resp = client.list_devices(NameFilter="test-")
+
+    assert len(resp["Devices"]) == 2
+    assert resp["Devices"][0]["DeviceId"] == resp_1["DeviceId"]
+    assert resp["Devices"][1]["DeviceId"] == resp_2["DeviceId"]
+
+
+@mock_panorama
+def test_list_device_max_result_and_next_token() -> None:
+    client = boto3.client("panorama", region_name="eu-west-1")
+    _ = client.provision_device(
+        Description="test device description 1",
+        Name="test-device-name-1",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+    _ = client.provision_device(
+        Description="test device description 2",
+        Name="test-device-name-2",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+
+    resp = client.list_devices(MaxResults=1)
+
+    assert len(resp["Devices"]) == 1
+    assert "NextToken" in resp
+
+    resp = client.list_devices(MaxResults=1, NextToken=resp["NextToken"])
+
+    assert len(resp["Devices"]) == 1
+    assert "NextToken" not in resp
+
+
+@pytest.mark.parametrize(
+    "sort_order, indexes",
+    [
+        ("ASCENDING", [0, 1]),
+        ("DESCENDING", [1, 0]),
+    ],
+)
+@mock_panorama
+def test_list_devices_sort_order(sort_order: str, indexes: List[int]) -> None:
+    client = boto3.client("panorama", region_name="eu-west-1")
+    resp_1 = client.provision_device(
+        Description="test device description 1",
+        Name="test-device-name-1",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+    resp_2 = client.provision_device(
+        Description="test device description 2",
+        Name="test-device-name-2",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+
+    resp = client.list_devices(SortOrder=sort_order)
+
+    assert len(resp["Devices"]) == 2
+    assert resp["Devices"][indexes[0]]["DeviceId"] == resp_1["DeviceId"]
+    assert resp["Devices"][indexes[1]]["DeviceId"] == resp_2["DeviceId"]
+
+
+@pytest.mark.parametrize(
+    "sort_by, indexes",
+    [
+        ("DEVICE_ID", [0, 1]),
+        ("CREATED_TIME", [1, 0]),
+        ("NAME", [0, 1]),
+        ("DEVICE_AGGREGATED_STATUS", [1, 0]),
+    ],
+)
+@mock_panorama
+def test_list_devices_sort_by(sort_by: str, indexes: List[int]) -> None:
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't freeze time in ServerMode")
+    client = boto3.client("panorama", region_name="eu-west-1")
+    state_manager.set_transition(
+        model_name="panorama::device_test-device-name-2_aggregated_status",
+        transition={"progression": "manual", "times": 1},
+    )
+    with freeze_time("2021-01-01 12:00:00"):
+        resp_1 = client.provision_device(
+            Description="test device description 1",
+            Name="test-device-name-1",
+            Tags={"Key": "test-key", "Value": "test-value"},
+        )
+    with freeze_time("2021-01-01 10:00:00"):
+        resp_2 = client.provision_device(
+            Description="test device description 2",
+            Name="test-device-name-2",
+            Tags={"Key": "test-key", "Value": "test-value"},
+        )
+
+    resp = client.list_devices(SortBy=sort_by)
+
+    assert len(resp["Devices"]) == 2
+    assert resp["Devices"][indexes[0]]["DeviceId"] == resp_1["DeviceId"]
+    assert resp["Devices"][indexes[1]]["DeviceId"] == resp_2["DeviceId"]
+
+
+@mock_panorama
+def test_list_devices_device_aggregated_status_filter() -> None:
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Can't use ManagedState in ServerMode")
+    client = boto3.client("panorama", region_name="eu-west-1")
+    state_manager.set_transition(
+        model_name="panorama::device_test-device-name-2_aggregated_status",
+        transition={"progression": "manual", "times": 1},
+    )
+    _ = client.provision_device(
+        Description="test device description 1",
+        Name="test-device-name-1",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+    resp_2 = client.provision_device(
+        Description="test device description 2",
+        Name="test-device-name-2",
+        Tags={"Key": "test-key", "Value": "test-value"},
+    )
+    # Need two advance to go from not-a-status to Pending
+    client.describe_device(DeviceId=resp_2["DeviceId"])
+
+    resp = client.list_devices(DeviceAggregatedStatusFilter="PENDING")
+
+    assert len(resp["Devices"]) == 1
+    assert resp["Devices"][0]["DeviceId"] == resp_2["DeviceId"]
+
+
+@mock_panorama
+def test_update_device_metadata() -> None:
+    client = boto3.client("panorama", region_name="eu-west-1")
+    resp = client.provision_device(
+        Description="test device description", Name="test-device-name"
+    )
+
+    client.update_device_metadata(
+        DeviceId=resp["DeviceId"],
+        Description="updated device description",
+    )
+
+    resp_updated = client.describe_device(DeviceId=resp["DeviceId"])
+
+    assert resp_updated["Description"] == "updated device description"
+
+
+@mock_panorama
+def test_delete_device() -> None:
+    client = boto3.client("panorama", region_name="eu-west-1")
+    resp = client.provision_device(
+        Description="test device description", Name="test-device-name"
+    )
+
+    client.delete_device(DeviceId=resp["DeviceId"])
+
+    with pytest.raises(ClientError) as ex:
+        client.describe_device(DeviceId=resp["DeviceId"])
+    err = ex.value.response
+    assert err["Error"]["Code"] == "ValidationException"
+    assert f"Device {resp['DeviceId']} not found" in err["Error"]["Message"]
+    assert err["ResponseMetadata"]["HTTPStatusCode"] == 400

tests/test_panorama/test_server.py

@@ -0,0 +1,12 @@
+"""Test different server responses."""
+
+import moto.server as server
+
+
+def test_panorama_list():
+    backend = server.create_backend_app("panorama")
+    test_client = backend.test_client()
+
+    resp = test_client.get("/devices")
+
+    assert resp.status_code == 200