test: harden conformance discovery, trait boundaries, and enum resilience

Author: allenporter

tests/conformance/discovery.py

@@ -21,10 +21,7 @@ def walk_modules(package: types.ModuleType) -> list[types.ModuleType]:
     if not hasattr(package, "__path__"):
         return modules
     for _, modname, _ in pkgutil.walk_packages(package.__path__, package.__name__ + "."):
-        try:
-            modules.append(importlib.import_module(modname))
-        except (ImportError, AttributeError):
-            continue
+        modules.append(importlib.import_module(modname))
     return modules
 
 

tests/conformance/test_enum_conformance.py

@@ -8,11 +8,14 @@
 
 from __future__ import annotations
 
+import enum
+import inspect
+
 import pytest
 
 import roborock
 from roborock.data.code_mappings import RoborockEnum, RoborockModeEnum
-from tests.conformance.discovery import discover_subclasses, to_pytest_params
+from tests.conformance.discovery import discover_subclasses, to_pytest_params, walk_modules
 
 # Baseline inventory of legacy RoborockEnum classes that do not yet define an
 # explicit `unknown` member. When newer firmware emits an undocumented code,
@@ -87,21 +90,62 @@
 _ALL_MODE_ENUMS = discover_subclasses(roborock, RoborockModeEnum, exclude=(RoborockModeEnum,))
 
 
+# Enums in wire code mapping modules that intentionally remain standard Enum/StrEnum/IntEnum
+# (e.g., domain categories, product nicknames, or outgoing command identifiers).
+ALLOWED_NON_RESILIENT_CODE_MAPPING_ENUMS = {
+    "roborock.data.b01_q10.b01_q10_code_mappings.RemoteCommand",
+    "roborock.data.code_mappings.RoborockCategory",
+    "roborock.data.code_mappings.RoborockProductNickname",
+    "roborock.data.v1.v1_code_mappings.RoborockDockState",
+}
+
+
+def _discover_code_mapping_enums() -> list[type[enum.Enum]]:
+    enums: list[type[enum.Enum]] = []
+    for mod in walk_modules(roborock):
+        if "code_mapping" in mod.__name__:
+            for _, obj in inspect.getmembers(mod, inspect.isclass):
+                if (
+                    obj.__module__ == mod.__name__
+                    and issubclass(obj, enum.Enum)
+                    and obj not in (enum.Enum, enum.IntEnum, enum.StrEnum, RoborockEnum, RoborockModeEnum)
+                ):
+                    enums.append(obj)
+    return sorted(enums, key=lambda c: f"{c.__module__}.{c.__name__}")
+
+
+_ALL_CODE_MAPPING_ENUMS = _discover_code_mapping_enums()
+
+
 @pytest.mark.parametrize("enum_cls", to_pytest_params(_ALL_ROBOROCK_ENUMS, marks_by_fqn=_XFAIL_MARKS))
 def test_roborock_enum_has_unknown_fallback(enum_cls: type[RoborockEnum]) -> None:
     """All RoborockEnum subclasses must define an explicit 'unknown' member."""
     assert hasattr(enum_cls, "unknown"), (
         f"{enum_cls.__module__}.{enum_cls.__name__} must define an 'unknown' member to prevent "
         "crashing or defaulting to arbitrary states on new firmware."
     )
-    # Also verify that resolving an unknown int code returns the unknown member
-    assert enum_cls(99999) == enum_cls.unknown
+    # Derive an integer sentinel guaranteed to not exist in the enum
+    sentinel = max(item.value for item in enum_cls) + 1 if list(enum_cls) else 99999
+    assert enum_cls(sentinel) == enum_cls.unknown
 
 
 @pytest.mark.parametrize("mode_enum_cls", to_pytest_params(_ALL_MODE_ENUMS))
 def test_roborock_mode_enum_handles_unknown_code(mode_enum_cls: type[RoborockModeEnum]) -> None:
     """RoborockModeEnum subclasses must return None when an unknown code is provided."""
-    assert mode_enum_cls.from_code_optional(99999) is None
+    sentinel = max(member.code for member in mode_enum_cls) + 1 if list(mode_enum_cls) else 99999
+    assert mode_enum_cls.from_code_optional(sentinel) is None
+
+
+@pytest.mark.parametrize("enum_cls", to_pytest_params(_ALL_CODE_MAPPING_ENUMS))
+def test_code_mapping_enums_inherit_resilient_bases(enum_cls: type[enum.Enum]) -> None:
+    """Enums in wire code mapping modules must inherit from RoborockEnum or RoborockModeEnum."""
+    fqn = f"{enum_cls.__module__}.{enum_cls.__name__}"
+    if fqn in ALLOWED_NON_RESILIENT_CODE_MAPPING_ENUMS:
+        return
+    assert issubclass(enum_cls, (RoborockEnum, RoborockModeEnum)), (
+        f"{fqn} in a code mapping module does not inherit from RoborockEnum or RoborockModeEnum. "
+        "Wire status and error code enums must use resilient enum bases to handle unknown firmware codes."
+    )
 
 
 def test_known_missing_unknown_baseline_inventory() -> None:

tests/conformance/test_trait_boundary_conformance.py

@@ -7,13 +7,18 @@
 
 from __future__ import annotations
 
+import asyncio
 import inspect
+import socket
+import typing
 
 import pytest
 
 import roborock.devices.traits
 from roborock.devices.traits import Trait
 from roborock.devices.traits.v1.common import V1TraitMixin
+from roborock.devices.transport.local_channel import LocalChannel, LocalChannelParams
+from roborock.devices.transport.mqtt_channel import MqttParams, MqttSession
 from tests.conformance.discovery import discover_classes, to_pytest_params
 
 FORBIDDEN_PARAMS = {
@@ -27,8 +32,29 @@
     "aes_key",
     "password",
     "secret",
+    "key",
 }
 
+FORBIDDEN_TYPES = (
+    socket.socket,
+    asyncio.BaseTransport,
+    LocalChannel,
+    LocalChannelParams,
+    MqttParams,
+    MqttSession,
+)
+
+
+def _extract_types(annotation: typing.Any) -> set[type]:
+    """Recursively extract underlying concrete types from type hints and unions."""
+    if isinstance(annotation, type):
+        return {annotation}
+    args = typing.get_args(annotation)
+    types_found: set[type] = set()
+    for arg in args:
+        types_found |= _extract_types(arg)
+    return types_found
+
 
 def _is_trait_class(cls: type) -> bool:
     if cls in (Trait, V1TraitMixin):
@@ -43,9 +69,19 @@ def _is_trait_class(cls: type) -> bool:
 def test_trait_constructor_does_not_leak_transport_or_credentials(trait_cls: type) -> None:
     """Trait constructor parameters must never include transport sockets, IPs, or credentials."""
     sig = inspect.signature(trait_cls)
-    param_names = set(sig.parameters.keys()) - {"self", "args", "kwargs"}
-    violations = param_names & FORBIDDEN_PARAMS
-    assert not violations, (
-        f"{trait_cls.__module__}.{trait_cls.__name__}.__init__ accepts forbidden transport/credential "
-        f"parameters {violations}. Per AGENTS.md, traits must receive only abstract channels or domain models."
-    )
+    for param_name, param in sig.parameters.items():
+        if param_name in ("self", "args", "kwargs"):
+            continue
+
+        assert param_name not in FORBIDDEN_PARAMS, (
+            f"{trait_cls.__module__}.{trait_cls.__name__}.__init__ accepts forbidden transport/credential "
+            f"parameter '{param_name}'. Per AGENTS.md, traits must receive only abstract channels or domain models."
+        )
+
+        types_found = _extract_types(param.annotation)
+        for t in types_found:
+            assert not issubclass(t, FORBIDDEN_TYPES), (
+                f"{trait_cls.__module__}.{trait_cls.__name__}.__init__ accepts parameter '{param_name}' "
+                f"typed with forbidden low-level transport/credential class {t.__name__}. "
+                "Traits must depend only on abstract channels (e.g. Channel, RpcChannel)."
+            )