diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index bd76a8b7c..0368ce97a 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -5173,6 +5173,46 @@
- [X] update_policy
+## panorama
+
+14% implemented
+
+- [ ] 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
+
+
## personalize
5% implemented
diff --git a/docs/docs/services/ec2.rst b/docs/docs/services/ec2.rst
index 866de236c..a040868ea 100644
--- a/docs/docs/services/ec2.rst
+++ b/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
-
diff --git a/docs/docs/services/panorama.rst b/docs/docs/services/panorama.rst
new file mode 100644
index 000000000..f2a7f25ee
--- /dev/null
+++ b/docs/docs/services/panorama.rst
@@ -0,0 +1,62 @@
+.. _implementedservice_panorama:
+
+.. |start-h3| raw:: html
+
+
+
+.. |end-h3| raw:: html
+
+
+
+========
+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
+
diff --git a/docs/docs/services/textract.rst b/docs/docs/services/textract.rst
index 28f84aafb..ea4dde02c 100644
--- a/docs/docs/services/textract.rst
+++ b/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
-
diff --git a/moto/__init__.py b/moto/__init__.py
index 5da436966..d08b22c0c 100644
--- a/moto/__init__.py
+++ b/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")
diff --git a/moto/backend_index.py b/moto/backend_index.py
index f61af9acf..300784891 100644
--- a/moto/backend_index.py
+++ b/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")),
diff --git a/moto/panorama/__init__.py b/moto/panorama/__init__.py
new file mode 100644
index 000000000..53d80c28b
--- /dev/null
+++ b/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)
diff --git a/moto/panorama/exceptions.py b/moto/panorama/exceptions.py
new file mode 100644
index 000000000..1f61bae33
--- /dev/null
+++ b/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)
diff --git a/moto/panorama/models.py b/moto/panorama/models.py
new file mode 100644
index 000000000..db81af483
--- /dev/null
+++ b/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",
+ ],
+)
diff --git a/moto/panorama/responses.py b/moto/panorama/responses.py
new file mode 100644
index 000000000..3e8b98f59
--- /dev/null
+++ b/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)
diff --git a/moto/panorama/urls.py b/moto/panorama/urls.py
new file mode 100644
index 000000000..13f54d0ff
--- /dev/null
+++ b/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[^/]+)$": PanoramaResponse.dispatch,
+}
diff --git a/moto/panorama/utils.py b/moto/panorama/utils.py
new file mode 100644
index 000000000..ad52c39df
--- /dev/null
+++ b/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")
diff --git a/tests/test_panorama/__init__.py b/tests/test_panorama/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_panorama/test_panorama_device.py b/tests/test_panorama/test_panorama_device.py
new file mode 100644
index 000000000..d452055bd
--- /dev/null
+++ b/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
diff --git a/tests/test_panorama/test_server.py b/tests/test_panorama/test_server.py
new file mode 100644
index 000000000..e10048aeb
--- /dev/null
+++ b/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