standalone: add unit-tests

Author: rnovatorov

src/enapter/mqtt/api/config.py

@@ -1,9 +1,18 @@
+import dataclasses
 import os
 from typing import MutableMapping, Self
 
 
+@dataclasses.dataclass(repr=False)
 class TLSConfig:
 
+    secret_key: str
+    cert: str
+    ca_cert: str
+
+    def __repr__(self) -> str:
+        return "mqtt.api.TLSConfig(...)"
+
     @classmethod
     def from_env(
         cls, env: MutableMapping[str, str] = os.environ, namespace: str = "ENAPTER_"
@@ -30,14 +39,16 @@ def pem(value: str) -> str:
 
         return cls(secret_key=pem(secret_key), cert=pem(cert), ca_cert=pem(ca_cert))
 
-    def __init__(self, secret_key: str, cert: str, ca_cert: str) -> None:
-        self.secret_key = secret_key
-        self.cert = cert
-        self.ca_cert = ca_cert
-
 
+@dataclasses.dataclass
 class Config:
 
+    host: str
+    port: int
+    user: str | None = None
+    password: str | None = None
+    tls_config: TLSConfig | None = None
+
     @classmethod
     def from_env(
         cls, env: MutableMapping[str, str] = os.environ, namespace: str = "ENAPTER_"
@@ -51,27 +62,6 @@ def from_env(
             tls_config=TLSConfig.from_env(env, namespace=namespace),
         )
 
-    def __init__(
-        self,
-        host: str,
-        port: int,
-        user: str | None = None,
-        password: str | None = None,
-        tls_config: TLSConfig | None = None,
-    ) -> None:
-        self.host = host
-        self.port = port
-        self.user = user
-        self.password = password
-        self.tls_config = tls_config
-
     @property
     def tls(self) -> TLSConfig | None:
         return self.tls_config
-
-    def __repr__(self) -> str:
-        return "mqtt.api.Config(host=%r, port=%r, tls=%r)" % (
-            self.host,
-            self.port,
-            self.tls is not None,
-        )

src/enapter/standalone/__init__.py

@@ -1,4 +1,9 @@
-from .config import Config
+from .config import (
+    CommunicationConfig,
+    CommunicationConfigV1,
+    CommunicationConfigV3,
+    Config,
+)
 from .device import Device
 from .device_protocol import (
     CommandArgs,
@@ -10,16 +15,21 @@
 )
 from .logger import Logger
 from .run import run
+from .ucm import UCM
 
 __all__ = [
     "CommandArgs",
     "CommandResult",
     "Config",
+    "CommunicationConfig",
+    "CommunicationConfigV1",
+    "CommunicationConfigV3",
     "Device",
     "DeviceProtocol",
     "Log",
     "Logger",
     "Properties",
     "Telemetry",
     "run",
+    "UCM",
 ]

src/enapter/standalone/config.py

@@ -47,7 +47,7 @@ def from_env(
         prefix = namespace + "STANDALONE_COMMUNICATION_"
         try:
             blob = env[namespace + "VUCM_BLOB"]
-            LOGGER.warn(
+            LOGGER.warning(
                 "`%s` is deprecated and will be removed soon. Please use `%s`.",
                 namespace + "VUCM_BLOB",
                 prefix + "CONFIG",
@@ -115,8 +115,8 @@ def from_config_v3(cls, config: "CommunicationConfigV3") -> Self:
                         ca_cert=config.mqtt_credentials.ca_chain,
                     ),
                 )
-            case _:
-                raise NotImplementedError(config.mqtt_protocol)
+            case _:  # pragma: no cover
+                raise NotImplementedError(config.mqtt_protocol)  # pragma: no cover
         assert mqtt_api_config is not None
         return cls(
             mqtt_api_config=mqtt_api_config,
@@ -205,8 +205,8 @@ def from_dto(cls, dto: dict[str, Any]) -> Self:
                 mqtt_credentials = cls.MQTTSCredentials.from_dto(
                     dto["mqtt_credentials"]
                 )
-            case _:
-                raise NotImplementedError(mqtt_protocol)
+            case _:  # pragma: no cover
+                raise NotImplementedError(mqtt_protocol)  # pragma: no cover
         assert mqtt_credentials is not None
         return cls(
             mqtt_host=dto["mqtt_host"],

src/enapter/standalone/device.py

@@ -27,7 +27,7 @@ def __init__(
 
     @abc.abstractmethod
     async def run(self) -> None:
-        pass
+        pass  # pragma: no cover
 
     @property
     def logger(self) -> Logger:

src/enapter/standalone/device_protocol.py

@@ -18,16 +18,16 @@ class Log:
 class DeviceProtocol(Protocol):
 
     async def run(self) -> None:
-        pass
+        pass  # pragma: no cover
 
     async def stream_properties(self) -> AsyncGenerator[Properties, None]:
-        yield {}
+        yield {}  # pragma: no cover
 
     async def stream_telemetry(self) -> AsyncGenerator[Telemetry, None]:
-        yield {}
+        yield {}  # pragma: no cover
 
     async def stream_logs(self) -> AsyncGenerator[Log, None]:
-        yield Log("debug", "", False)
+        yield Log("debug", "", False)  # pragma: no cover
 
     async def execute_command(self, name: str, args: CommandArgs) -> CommandResult:
-        pass
+        pass  # pragma: no cover

src/enapter/standalone/mqtt_adapter.py

@@ -81,8 +81,8 @@ async def _stream_logs(self) -> None:
                         self._logger.warning(log.message)
                     case "error":
                         self._logger.error(log.message)
-                    case _:
-                        raise NotImplementedError(log.severity)
+                    case _:  # pragma: no cover
+                        raise NotImplementedError(log.severity)  # pragma: no cover
                 await self._publish_log(
                     mqtt.api.device.Log(
                         timestamp=int(time.time()),