interop utilities for mplex ping

interop/exec/config/mod.py

@@ -0,0 +1,57 @@
+from dataclasses import (
+    dataclass,
+)
+import os
+from typing import (
+    Optional,
+)
+
+
+def str_to_bool(val: str) -> bool:
+    return val.lower() in ("true", "1")
+
+
+class ConfigError(Exception):
+    """Raised when the required environment variables are missing or invalid"""
+
+
+@dataclass
+class Config:
+    transport: str
+    sec_protocol: Optional[str]
+    muxer: Optional[str]
+    ip: str
+    is_dialer: bool
+    test_timeout: int
+    redis_addr: str
+    port: str
+
+    @classmethod
+    def from_env(cls) -> "Config":
+        try:
+            transport = os.environ["transport"]
+            ip = os.environ["ip"]
+        except KeyError as e:
+            raise ConfigError(f"{e.args[0]} env variable not set") from None
+
+        try:
+            is_dialer = str_to_bool(os.environ.get("is_dialer", "true"))
+            test_timeout = int(os.environ.get("test_timeout", "180"))
+        except ValueError as e:
+            raise ConfigError(f"Invalid value in env: {e}") from None
+
+        redis_addr = os.environ.get("redis_addr", 6379)
+        sec_protocol = os.environ.get("security")
+        muxer = os.environ.get("muxer")
+        port = os.environ.get("port", "8000")
+
+        return cls(
+            transport=transport,
+            sec_protocol=sec_protocol,
+            muxer=muxer,
+            ip=ip,
+            is_dialer=is_dialer,
+            test_timeout=test_timeout,
+            redis_addr=redis_addr,
+            port=port,
+        )

interop/exec/native_ping.py

@@ -0,0 +1,33 @@
+import trio
+
+from interop.exec.config.mod import (
+    Config,
+    ConfigError,
+)
+from interop.lib import (
+    run_test,
+)
+
+
+async def main() -> None:
+    try:
+        config = Config.from_env()
+    except ConfigError as e:
+        print(f"Config error: {e}")
+        return
+
+    # Uncomment and implement when ready
+    _ = await run_test(
+        config.transport,
+        config.ip,
+        config.port,
+        config.is_dialer,
+        config.test_timeout,
+        config.redis_addr,
+        config.sec_protocol,
+        config.muxer,
+    )
+
+
+if __name__ == "__main__":
+    trio.run(main)