Fix lint and test issues on main (#247)

Author: MartinHjelmare

matter_server/server/const.py

@@ -1,4 +1,18 @@
 """Server-only constants for the Python Matter Server."""
+import pathlib
+from typing import Final
 
 # The minimum schema version (of a client) the server can support
 MIN_SCHEMA_VERSION = 2
+
+# the paa-root-certs path is hardcoded in the sdk at this time
+# and always uses the development subfolder
+# regardless of anything you pass into instantiating the controller
+# revisit this once matter 1.1 is released
+PAA_ROOT_CERTS_DIR: Final[pathlib.Path] = (
+    pathlib.Path(__file__)
+    .parent.resolve()
+    .parent.resolve()
+    .parent.resolve()
+    .joinpath("credentials/development/paa-root-certs")
+)

matter_server/server/device_controller.py

@@ -7,16 +7,13 @@
 from datetime import datetime
 from functools import partial
 import logging
-import pathlib
-from typing import TYPE_CHECKING, Any, Callable, Deque, Final, Type, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Callable, Deque, Type, TypeVar, cast
 
 from chip.ChipDeviceCtrl import CommissionableNode
 from chip.clusters import Attribute, Objects as Clusters
 from chip.clusters.ClusterObjects import ALL_CLUSTERS, Cluster
 from chip.exceptions import ChipStackError
 
-from matter_server.server.helpers.paa_certificates import fetch_certificates
-
 from ..common.const import SCHEMA_VERSION
 from ..common.errors import (
     NodeCommissionFailed,
@@ -31,30 +28,21 @@
     dataclass_from_dict,
 )
 from ..common.models import APICommand, EventType, MatterNodeData
+from .const import PAA_ROOT_CERTS_DIR
+from .helpers.paa_certificates import fetch_certificates
 
 if TYPE_CHECKING:
-    from .server import MatterServer
     from chip.ChipDeviceCtrl import ChipDeviceController
 
+    from .server import MatterServer
+
 _T = TypeVar("_T")
 
 DATA_KEY_NODES = "nodes"
 DATA_KEY_LAST_NODE_ID = "last_node_id"
 
 LOGGER = logging.getLogger(__name__)
 
-# the paa-root-certs path is hardcoded in the sdk at this time
-# and always uses the development subfolder
-# regardless of anything you pass into instantiating the controller
-# revisit this once matter 1.1 is released
-PAA_ROOT_CERTS_DIR: Final[pathlib.Path] = (
-    pathlib.Path(__file__)
-    .parent.resolve()
-    .parent.resolve()
-    .parent.resolve()
-    .joinpath("credentials/development/paa-root-certs")
-)
-
 
 class MatterDeviceController:
     """Class that manages the Matter devices."""
@@ -67,10 +55,6 @@ def __init__(
     ):
         """Initialize the device controller."""
         self.server = server
-        # Instantiate the underlying ChipDeviceController instance on the Fabric
-        if not PAA_ROOT_CERTS_DIR.is_dir():
-            raise RuntimeError("PAA certificates directory not found")
-
         # we keep the last events in memory so we can include them in the diagnostics dump
         self.event_history: Deque[Attribute.EventReadResult] = deque(maxlen=25)
         self._subscriptions: dict[int, Attribute.SubscriptionTransaction] = {}
@@ -84,7 +68,8 @@ async def initialize(self) -> None:
         """Async initialize of controller."""
         # (re)fetch all PAA certificates once at startup
         # NOTE: this must be done before initializing the controller
-        await fetch_certificates(PAA_ROOT_CERTS_DIR)
+        await fetch_certificates()
+        # Instantiate the underlying ChipDeviceController instance on the Fabric
         self.chip_controller = self.server.stack.fabric_admin.NewController(
             paaTrustStorePath=str(PAA_ROOT_CERTS_DIR)
         )
@@ -117,6 +102,9 @@ async def start(self) -> None:
 
     async def stop(self) -> None:
         """Handle logic on server stop."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         # unsubscribe all node subscriptions
         for sub in self._subscriptions.values():
             await self._call_sdk(sub.Shutdown)
@@ -147,9 +135,12 @@ async def commission_with_code(self, code: str) -> MatterNodeData:
 
         Returns full NodeInfo once complete.
         """
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         # perform a quick delta sync of certificates to make sure
         # we have the latest paa root certs
-        await fetch_certificates(PAA_ROOT_CERTS_DIR)
+        await fetch_certificates()
         node_id = self._get_next_node_id()
 
         success = await self._call_sdk(
@@ -184,11 +175,14 @@ async def commission_on_network(
         a string or None depending on the actual type of selected filter.
         Returns full NodeInfo once complete.
         """
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         # perform a quick delta sync of certificates to make sure
         # we have the latest paa root certs
         # NOTE: Its not very clear if the newly fetched certificates can be used without
         # restarting the device controller
-        await fetch_certificates(PAA_ROOT_CERTS_DIR)
+        await fetch_certificates()
 
         node_id = self._get_next_node_id()
 
@@ -214,6 +208,9 @@ async def commission_on_network(
     @api_command(APICommand.SET_WIFI_CREDENTIALS)
     async def set_wifi_credentials(self, ssid: str, credentials: str) -> None:
         """Set WiFi credentials for commissioning to a (new) device."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         await self._call_sdk(
             self.chip_controller.SetWiFiCredentials,
             ssid=ssid,
@@ -225,6 +222,9 @@ async def set_wifi_credentials(self, ssid: str, credentials: str) -> None:
     @api_command(APICommand.SET_THREAD_DATASET)
     async def set_thread_operational_dataset(self, dataset: str) -> None:
         """Set Thread Operational dataset in the stack."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         await self._call_sdk(
             self.chip_controller.SetThreadOperationalDataset,
             threadOperationalDataset=bytes.fromhex(dataset),
@@ -246,6 +246,9 @@ async def open_commissioning_window(
 
         Returns code to use as discriminator.
         """
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         if discriminator is None:
             discriminator = 3840  # TODO generate random one
 
@@ -264,6 +267,8 @@ async def discover_commissionable_nodes(
         self,
     ) -> CommissionableNode | list[CommissionableNode] | None:
         """Discover Commissionable Nodes (discovered on BLE or mDNS)."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
 
         result = await self._call_sdk(
             self.chip_controller.DiscoverCommissionableNodes,
@@ -273,6 +278,9 @@ async def discover_commissionable_nodes(
     @api_command(APICommand.INTERVIEW_NODE)
     async def interview_node(self, node_id: int) -> None:
         """Interview a node."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         LOGGER.debug("Interviewing node: %s", node_id)
         try:
             await self._call_sdk(self.chip_controller.ResolveNode, nodeid=node_id)
@@ -328,6 +336,9 @@ async def send_device_command(
         interaction_timeout_ms: int | None = None,
     ) -> Any:
         """Send a command to a Matter node/device."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         cluster_cls: Cluster = ALL_CLUSTERS[cluster_id]
         command_cls = getattr(cluster_cls.Commands, command_name)
         command = dataclass_from_dict(command_cls, payload)
@@ -343,6 +354,9 @@ async def send_device_command(
     @api_command(APICommand.REMOVE_NODE)
     async def remove_node(self, node_id: int) -> None:
         """Remove a Matter node/device from the fabric."""
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         if node_id not in self._nodes:
             raise NodeNotExists(
                 f"Node {node_id} does not exist or has not been interviewed."
@@ -378,6 +392,9 @@ async def subscribe_node(self, node_id: int) -> None:
 
         Note that by using the listen command at server level, you will receive all node events.
         """
+        if self.chip_controller is None:
+            raise RuntimeError("Device Controller not initialized.")
+
         if node_id not in self._nodes:
             raise NodeNotExists(
                 f"Node {node_id} does not exist or has not been interviewed."

matter_server/server/helpers/paa_certificates.py

@@ -9,30 +9,30 @@
 
 import asyncio
 import logging
-import pathlib
+from os import makedirs
 import re
 
-from aiohttp import ClientSession, ClientError
+from aiohttp import ClientError, ClientSession
 from cryptography import x509
 from cryptography.hazmat.primitives import serialization
 
+from matter_server.server.const import PAA_ROOT_CERTS_DIR
+
 LOGGER = logging.getLogger(__name__)
 PRODUCTION_URL = "https://on.dcl.csa-iot.org"
 TEST_URL = "https://on.test-net.dcl.csa-iot.org"
 
 LAST_CERT_IDS: set[str] = set()
 
 
-async def write_paa_root_cert(
-    certificate: str, subject: str, root_path: pathlib.Path
-) -> None:
+async def write_paa_root_cert(certificate: str, subject: str) -> None:
     """Write certificate from string to file."""
 
     def _write() -> None:
         filename_base = "dcld_mirror_" + re.sub(
             "[^a-zA-Z0-9_-]", "", re.sub("[=, ]", "_", subject)
         )
-        filepath_base = root_path.joinpath(filename_base)
+        filepath_base = PAA_ROOT_CERTS_DIR.joinpath(filename_base)
         # handle PEM certificate file
         file_path_pem = f"{filepath_base}.pem"
         LOGGER.debug("Writing certificate %s", file_path_pem)
@@ -50,12 +50,14 @@ def _write() -> None:
 
 
 async def fetch_certificates(
-    paa_trust_store_path: pathlib.Path,
     fetch_test_certificates: bool = True,
     fetch_production_certificates: bool = True,
 ) -> int:
     """Fetch PAA Certificates."""
     LOGGER.info("Fetching the latest PAA root certificates from DCL.")
+    if not PAA_ROOT_CERTS_DIR.is_dir():
+        loop = asyncio.get_running_loop()
+        await loop.run_in_executor(None, makedirs, PAA_ROOT_CERTS_DIR)
     fetch_count: int = 0
     base_urls = set()
     # determine which url's need to be queried.
@@ -94,7 +96,6 @@ async def fetch_certificates(
                     await write_paa_root_cert(
                         certificate,
                         subject,
-                        paa_trust_store_path,
                     )
                     LAST_CERT_IDS.add(paa["subjectKeyId"])
                     fetch_count += 1

tests/server/test_server.py

@@ -87,6 +87,15 @@ def storage_controller_fixture() -> Generator[MagicMock, None, None]:
         yield storage_controller
 
 
+@pytest.fixture(name="fetch_certificates", autouse=True)
+def fetch_certificates_fixture() -> Generator[MagicMock, None, None]:
+    """Return a mocked fetch certificates."""
+    with patch(
+        "matter_server.server.device_controller.fetch_certificates", autospec=True
+    ) as fetch_certificates:
+        yield fetch_certificates
+
+
 @pytest.fixture(name="server")
 async def server_fixture() -> AsyncGenerator[MatterServer, None]:
     """Yield a server."""