Functional service implementation, tests

* fixed and tested config load
  * Currently the config merge order is the following. CLI overrides environment vars, environment vars override YAML config and YAML config overrides defaults.
None of the separate configs has be complete, but the merge result does. It has to contain all required settings. The reasoning behind the merge order is convenience for Docker image usages. Docker image runs are often customized by `command` and `environment`. Probably the behaviour may be adjusted after some feedback
  * some automation to name environment variables according to config dataclasses. It's tedious to duplicate the names
  * added config tests to verify the merge
* service implementations for server and client have common funtional of pumps. Those pumps is for pumping inbound WS to ZMQ sink and ZMQ source to outbound WS. Fixed wrong `WSFrame` usage, caught by tests (one shouldn't store `WSFrame` as per the doc). Fine tune finalizations when testing.
* tests for multiple identity pipeline, includes no SSL option, multiple message streams, checking conditions when WS write pauses are possible
* test fixtures are refactored to keep them in a more controlled and organized way
* added Ruff check for sorded imports
* added `pytest-xdist` to have parallel workers for tests and corresponding `-n 3` args in `pyproject.toml` in to run `pytest`
* added `pytest-randomly` to control the random seed and be able to reproduce tests with randomized fixtures

refactor tests

cleaner pumps separation

fix config and tests

Author: abukharov268

.github/workflows/checks.yml

@@ -20,6 +20,8 @@ jobs:
         run: pyrefly check --summarize-errors .
       - name: Lint code with Ruff
         run: ruff check --output-format=github
+      - name: Check sorted imports with Ruff
+        run: ruff check --select I --output-format=github
       - name: Check code formatting with Ruff
         run: ruff format --diff
       - name: Run pytest unit/integration tests

pyproject.toml

@@ -18,6 +18,10 @@ project-excludes = [
     "**/*venv/**/*",
 ]
 
+[tool.pytest.ini_options]
+addopts = "-n 3"
+
+
 [project.optional-dependencies]
 prerequisite_platform = [
   # This is just for visibilty, to track prerequisite packages.
@@ -29,8 +33,10 @@ prerequisite_platform = [
 dev = [
     "faker>=37.11.0",
     "filelock>=3.20.0",
-    "pyrefly>=0.37.0",
+    "pyrefly>=0.39.0",
     "pytest>=8.4.2",
     "pytest-asyncio>=1.2.0",
+    "pytest-randomly>=4.0.1",
+    "pytest-xdist>=3.8.0",
     "ruff>=0.14.1",
 ]

savant_cloudpin/cfg/_bootstrap.py

@@ -1,17 +1,16 @@
-import dataclasses
+import os.path
 import re
-from pathlib import Path
 from typing import Any
 
 from omegaconf import OmegaConf
 from omegaconf.base import SCMode
 from omegaconf.dictconfig import DictConfig
 
+from savant_cloudpin.cfg import _utils as utils
 from savant_cloudpin.cfg._defaults import (
     DEFAULT_CLIENT_CONFIG,
     DEFAULT_LOAD_CONFIG,
     DEFAULT_SERVER_CONFIG,
-    NULL_SSL_SERVER_CONFIG,
 )
 from savant_cloudpin.cfg._models import ClientServiceConfig, ServerServiceConfig
 
@@ -41,35 +40,51 @@ def validated_dataclass[T: ClientServiceConfig | ServerServiceConfig](
     return config
 
 
+def merge_env_config(
+    defaults: ServerServiceConfig | ClientServiceConfig,
+    yml_cfg: DictConfig | dict,
+    cli_cfg: DictConfig | dict,
+) -> DictConfig:
+    default_cfg = OmegaConf.structured(defaults)
+    env_cfg = utils.as_value_dict(utils.env_override(defaults, "null"))
+    env_cfg = OmegaConf.to_container(OmegaConf.create(env_cfg), resolve=True)
+    assert isinstance(env_cfg, dict)
+    env_cfg = utils.drop_none_values(env_cfg)
+    cfg = OmegaConf.merge(yml_cfg, env_cfg, cli_cfg)
+
+    ssl = "websockets" in cfg and "ssl" in cfg.websockets
+    ssl = cfg.websockets.ssl if ssl else {}
+
+    cfg = OmegaConf.merge(default_cfg, cfg)
+    assert isinstance(cfg, DictConfig)
+    if not any(val is not None for val in ssl.values()):
+        cfg.websockets.ssl = None
+    return cfg
+
+
 def load_config(
     args_list: list[str] | None = None,
 ) -> ClientServiceConfig | ServerServiceConfig:
-    cli_config = OmegaConf.from_cli(args_list)
-    load_config = OmegaConf.merge(DEFAULT_LOAD_CONFIG, cli_config)
+    cli_cfg = OmegaConf.from_cli(args_list)
+    env_cfg = utils.as_value_dict(utils.env_override(DEFAULT_LOAD_CONFIG))
+    cfg = OmegaConf.merge(env_cfg, cli_cfg)
+
+    yml_exists = cfg.config and os.path.exists(cfg.config)
+    yml_cfg = OmegaConf.load(cfg.config) if yml_exists else OmegaConf.create({})
 
-    if load_config.config and Path(load_config.config).exists():
-        yaml_config = OmegaConf.load(load_config.config)
-    else:
-        yaml_config = OmegaConf.create()
+    cfg = OmegaConf.merge(yml_cfg, env_cfg, cli_cfg)
+    assert isinstance(cfg, DictConfig) and isinstance(yml_cfg, DictConfig)
 
-    config = OmegaConf.merge(DEFAULT_LOAD_CONFIG, yaml_config, cli_config)
-    assert isinstance(config, DictConfig)
+    cli_cfg.pop("config", None)
+    cli_cfg.pop("mode", None)
+    yml_cfg.pop("mode", None)
 
-    config.pop("config", None)
-    match config.pop("mode", None):
+    match cfg.mode:
         case "server":
-            ssl_config = OmegaConf.merge(
-                dataclasses.asdict(NULL_SSL_SERVER_CONFIG), config
-            ).websockets.ssl
-            assert isinstance(ssl_config, DictConfig)
-            missing_ssl = all(v is None for _, v in ssl_config.items())
-
-            config = OmegaConf.merge(DEFAULT_SERVER_CONFIG, config)
-            if missing_ssl:
-                config.websockets.ssl = None
-            return validated_dataclass(config, ServerServiceConfig)
+            cfg = merge_env_config(DEFAULT_SERVER_CONFIG, yml_cfg, cli_cfg)
+            return validated_dataclass(cfg, ServerServiceConfig)
         case "client" | None:
-            config = OmegaConf.merge(DEFAULT_CLIENT_CONFIG, config)
-            return validated_dataclass(config, ClientServiceConfig)
+            cfg = merge_env_config(DEFAULT_CLIENT_CONFIG, yml_cfg, cli_cfg)
+            return validated_dataclass(cfg, ClientServiceConfig)
         case _:
             raise ValueError("Invalid service mode")

savant_cloudpin/cfg/_defaults.py

@@ -1,7 +1,3 @@
-import copy
-
-from omegaconf import SI
-
 from savant_cloudpin.cfg._models import (
     ClientServiceConfig,
     ClientSSLConfig,
@@ -15,74 +11,48 @@
 )
 
 DEFAULT_SOURCE_CONFIG = ReaderConfig(
-    results_queue_size=SI("${oc.env:CLOUDPIN_SOURCE_RESULTS_QUEUE_SIZE,100}"),
-    url="${oc.env:CLOUDPIN_SOURCE_URL,???}",
-    receive_timeout=SI("${oc.env:CLOUDPIN_SOURCE_RECEIVE_TIMEOUT,null}"),
-    receive_hwm=SI("${oc.env:CLOUDPIN_SOURCE_RECEIVE_HWM,null}"),
-    topic_prefix_spec="${oc.env:CLOUDPIN_SOURCE_TOPIC_PREFIX_SPEC,null}",
-    routing_ids_cache_size=SI("${oc.env:CLOUDPIN_ROUTING_IDS_CACHE_SIZE,null}"),
-    fix_ipc_permissions=("${oc.env:CLOUDPIN_SOURCE_FIX_IPC_PERMISSIONS,null}"),
-    source_blacklist_size=SI("${oc.env:CLOUDPIN_SOURCE_SOURCE_BLACKLIST_SIZE,null}"),
-    source_blacklist_ttl=SI("${oc.env:CLOUDPIN_SOURCE_SOURCE_BLACKLIST_TTL,null}"),
+    results_queue_size=100,
+    url="???",
 )
 DEFAULT_SINK_CONFIG = WriterConfig(
-    max_inflight_messages=SI("${oc.env:CLOUDPIN_SINK_MAX_INFLIGHT_MESSAGES,100}"),
-    url="${oc.env:CLOUDPIN_SINK_URL,???}",
-    send_timeout=SI("${oc.env:CLOUDPIN_SINK_SEND_TIMEOUT,null}"),
-    send_retries=SI("${oc.env:CLOUDPIN_SINK_SEND_RETRIES,null}"),
-    send_hwm=SI("${oc.env:CLOUDPIN_SINK_SEND_HWM,null}"),
-    receive_timeout=SI("${oc.env:CLOUDPIN_SINK_RECEIVE_TIMEOUT,null}"),
-    receive_retries=SI("${oc.env:CLOUDPIN_SINK_RECEIVE_RETRIES,null}"),
-    receive_hwm=SI("${oc.env:CLOUDPIN_SINK_RECEIVE_HWM,null}"),
-    fix_ipc_permissions=SI("${oc.env:CLOUDPIN_SINK_FIX_IPC_PERMISSIONS,null}"),
+    max_inflight_messages=100,
+    url="???",
 )
 
 
 DEFAULT_LOAD_CONFIG = LoadConfig(
-    config="${oc.env:CLOUDPIN_CONFIG_FILE,./cloudpin.yml}",
-    mode="${oc.env:CLOUDPIN_MODE,client}",
+    config="./cloudpin.yml",
+    mode="client",
 )
 
 DEFAULT_CLIENT_CONFIG = ClientServiceConfig(
     websockets=ClientWSConfig(
-        server_url="${oc.env:CLOUDPIN_WEBSOCKETS_SERVER_URL,???}",
-        api_key="${oc.env:CLOUDPIN_WEBSOCKETS_API_KEY,???}",
+        server_url="???",
+        api_key="???",
         ssl=ClientSSLConfig(
-            ca_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CA_FILE,null}",
-            cert_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CERT_FILE,???}",
-            key_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_KEY_FILE,???}",
-            check_hostname=SI("${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CHECK_HOSTNAME,false}"),
+            ca_file=None,
+            cert_file="???",
+            key_file="???",
+            check_hostname=False,
         ),
     ),
-    io_timeout=SI("${oc.env:CLOUDPIN_IO_TIMEOUT,0.1}"),
+    io_timeout=0.1,
     source=DEFAULT_SOURCE_CONFIG,
     sink=DEFAULT_SINK_CONFIG,
 )
 
 DEFAULT_SERVER_CONFIG = ServerServiceConfig(
     websockets=ServerWSConfig(
-        server_url="${oc.env:CLOUDPIN_WEBSOCKETS_SERVER_URL,???}",
-        api_key="${oc.env:CLOUDPIN_WEBSOCKETS_API_KEY,???}",
+        server_url="???",
+        api_key="???",
         ssl=ServerSSLConfig(
-            ca_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CA_FILE,null}",
-            cert_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CERT_FILE,???}",
-            key_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_KEY_FILE,???}",
-            client_cert_required=SI(
-                "${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CLIENT_CERT_REQUIRED,true}"
-            ),
+            ca_file=None,
+            cert_file="???",
+            key_file="???",
+            client_cert_required=True,
         ),
     ),
-    io_timeout=SI("${oc.env:CLOUDPIN_IO_TIMEOUT,0.1}"),
+    io_timeout=0.1,
     source=DEFAULT_SOURCE_CONFIG,
     sink=DEFAULT_SINK_CONFIG,
 )
-
-NULL_SSL_SERVER_CONFIG = copy.deepcopy(DEFAULT_SERVER_CONFIG)
-NULL_SSL_SERVER_CONFIG.websockets.ssl = ServerSSLConfig(
-    ca_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CA_FILE,null}",
-    cert_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CERT_FILE,null}",
-    key_file="${oc.env:CLOUDPIN_WEBSOCKETS_SSL_KEY_FILE,null}",
-    client_cert_required=SI(
-        "${oc.env:CLOUDPIN_WEBSOCKETS_SSL_CLIENT_CERT_REQUIRED,null}"
-    ),
-)

savant_cloudpin/cfg/_models.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass, replace
-from typing import Self, TypedDict
+from typing import Self
 
 from savant_rs import zmq
 
@@ -21,13 +21,10 @@ class ReaderConfig:
     def as_router(self) -> Self:
         return replace(self, url="router+" + self.url.split("+")[-1])
 
-    def blocking_params(self) -> zmq.ReaderConfig:
+    def to_args(self) -> tuple[zmq.ReaderConfig, int]:
         cfg = zmq.ReaderConfigBuilder(self.url)
         cfg.with_map_config(to_map_config(self, excluded=("url", "results_queue_size")))
-        return cfg.build()
-
-    def nonblocking_params(self) -> tuple[zmq.ReaderConfig, int]:
-        return self.blocking_params(), self.results_queue_size
+        return cfg.build(), self.results_queue_size
 
 
 @dataclass
@@ -45,7 +42,7 @@ class WriterConfig:
     def as_dealer(self) -> Self:
         return replace(self, url="dealer+" + self.url.split("+")[-1])
 
-    def nonblocking_params(self) -> tuple[zmq.WriterConfig, int]:
+    def to_args(self) -> tuple[zmq.WriterConfig, int]:
         cfg = zmq.WriterConfigBuilder(self.url)
         cfg.with_map_config(
             to_map_config(self, excluded=("url", "max_inflight_messages"))
@@ -73,7 +70,7 @@ class ClientSSLConfig:
 class ServerWSConfig:
     server_url: str
     api_key: str
-    ssl: ServerSSLConfig | None
+    ssl: ServerSSLConfig | None = None
 
 
 @dataclass
@@ -100,6 +97,7 @@ class ClientServiceConfig(BaseServiceConfig):
     websockets: ClientWSConfig
 
 
-class LoadConfig(TypedDict):
+@dataclass
+class LoadConfig:
     config: str | None
     mode: str | None

savant_cloudpin/cfg/_utils.py

@@ -1,6 +1,10 @@
+import dataclasses
+from collections.abc import Mapping
 from dataclasses import asdict
 from typing import Any
 
+ENV_PREFIX = "CLOUDPIN"
+
 
 def to_map_config(
     dataclass_config: Any, /, excluded: tuple[str, ...]
@@ -11,3 +15,46 @@ def to_map_config(
         if isinstance(key, str) and isinstance(val, (str, int))
         if key not in excluded
     }
+
+
+def drop_none_values[K](dct: Mapping[K, Any]) -> dict[K, Any]:
+    result = dict[K, Any]()
+    for key, val in dct.items():
+        if isinstance(val, dict):
+            result[key] = drop_none_values(val)
+        elif val is not None:
+            result[key] = val
+    return result
+
+
+def as_value_dict(obj: Any) -> dict[str, Any]:
+    dct = dataclasses.asdict(obj)
+    return drop_none_values(dct)
+
+
+def env_override(
+    obj: Any,
+    default: str | None = None,
+    prefix: str = ENV_PREFIX,
+) -> Any:
+    if isinstance(obj, type):
+        raise ValueError("Instance is expected")
+    updates = dict[str, Any]()
+    for field in dataclasses.fields(obj):
+        name, val = field.name, getattr(obj, field.name)
+        env_name = f"{prefix}_{name.upper()}"
+        match val:
+            case str() | int() | float() if default is None:
+                updates[name] = f"${{oc.env:{env_name},{val}}}"
+            case str() | int() | float():
+                updates[name] = f"${{oc.env:{env_name},{default}}}"
+            case bool() if default is None:
+                updates[name] = f"${{oc.env:{env_name},{str(val).lower()}}}"
+            case bool():
+                updates[name] = f"${{oc.env:{env_name},{default}}}"
+            case None:
+                continue
+            case _:
+                updates[name] = env_override(val, default, env_name)
+
+    return dataclasses.replace(obj, **updates)