"""Contract tests: hermes-meshtastic vs the REAL Hermes gateway API. These tests exist so adapter development converges WITHOUT deploying: they replay the gateway's adapter bootstrap against the actual ``hermes-agent`` package (pinned to the deployed image tag, see ``scripts/contract-test.sh``) and fail loudly on any API drift that used to surface only as a gateway crash after release. Runtime ordering modelled here (verified against ``gateway/run.py`` and ``gateway/config.py``): 1. plugin discovery runs the plugin's ``register(ctx)`` (registers the platform in ``gateway.platform_registry``), 2. the gateway later instantiates the adapter via the registry entry and wires it: ``set_message_handler`` -> ... -> ``connect()``, 3. ``Platform("meshtastic")`` (auto-created enum member) is only valid AFTER registration — the enum's ``_missing_`` queries the registry. Layers covered -------------- L0 The adapter subclasses the REAL ``gateway.platforms.base.BasePlatformAdapter`` (regression for the fallback-stub bug that caused the ``AttributeError: ... 'set_message_handler'`` crash at ``gateway/run.py``). L0b ABC instantiation gate — construction with a real ``PlatformConfig`` fails if any abstract member is missing or the platform enum is misused. L1 Wiring replay — every ``set_*`` call ``gateway/run.py`` makes at startup must exist and accept a handler. L2 Signature conformance — ``connect``/``send`` match the gateway contract. L3 Event shape — the inbound ``MessageEvent``/``SessionSource`` construction must be valid against the real gateway types. Skip policy: skipped when ``hermes-agent`` is not installed so the standalone unit suite (CI without Hermes) stays green. Set ``HERMES_CONTRACT_REQUIRED=1`` to turn absence into a hard failure — the dedicated contract CI job sets this so the gate can never silently skip. """ import inspect import os from types import SimpleNamespace import pytest try: from gateway.config import Platform, PlatformConfig from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType except ImportError: # pragma: no cover - exercised when hermes-agent is absent if os.environ.get("HERMES_CONTRACT_REQUIRED"): raise AssertionError( "hermes-agent is required for contract tests (HERMES_CONTRACT_REQUIRED=1). " "Run `scripts/contract-test.sh` or the CI contract job." ) pytest.skip( "hermes-agent not installed — contract tests skipped (scripts/contract-test.sh)", allow_module_level=True, ) import hermes_meshtastic # noqa: F401,E402 (package must import without gateway side effects) from hermes_meshtastic import adapter as plugin_adapter # noqa: E402 MINIMAL_CONFIG = PlatformConfig( enabled=True, extra={"host": "127.0.0.1", "port": 1, "channel_index": 1}, ) # Exactly the set_* wiring gateway/run.py performs right before connect() # (observed at the crash site: adapter.set_message_handler(...) etc.). GATEWAY_WIRING_SETTERS = [ "set_message_handler", "set_fatal_error_handler", "set_session_store", "set_busy_session_handler", "set_topic_recovery_fn", "set_authorization_check", "set_platform_event_handler", "set_reaction_handler", ] _REGISTER_KWARGS = {} class _Ctx: """Minimal ctx mirroring how discovery calls the plugin's register().""" def register_platform(self, **kwargs): _REGISTER_KWARGS.update(kwargs) def _noop(*_args, **_kwargs): return None @pytest.fixture(scope="module", autouse=True) def registered_entry(): """Replay discovery: run register(), then register the PlatformEntry. Mirrors how gateway/run.py later instantiates the adapter from the registry — and why Platform("meshtastic") is valid only after this. """ from gateway.platform_registry import PlatformEntry, platform_registry hermes_meshtastic.register(_Ctx()) kw = dict(_REGISTER_KWARGS) assert kw.get("name") == "meshtastic" if not platform_registry.is_registered("meshtastic"): entry = PlatformEntry( name=kw["name"], label=kw["label"], adapter_factory=kw["adapter_factory"], check_fn=kw["check_fn"], required_env=[], max_message_length=kw.get("max_message_length", 0), emoji=kw.get("emoji", "🔌"), platform_hint=kw.get("platform_hint", ""), ) platform_registry.register(entry) return platform_registry.get("meshtastic") def _make_adapter(entry=None): from gateway.platform_registry import platform_registry if entry is None: entry = platform_registry.get("meshtastic") assert entry is not None, "meshtastic platform not registered (fixture order broken)" return entry.adapter_factory(MINIMAL_CONFIG) # --- L0: real base binding ------------------------------------------------- def test_adapter_subclasses_real_gateway_base(): """Regression: adapter must bind the real base, never a local stub.""" cls = plugin_adapter.get_adapter_class() assert inspect.isclass(cls) assert issubclass(cls, BasePlatformAdapter) assert cls.__mro__[1].__module__ == "gateway.platforms.base" def test_abstract_members_are_implemented(): """ABC gate: BasePlatformAdapter requires connect/send/disconnect/get_chat_info.""" cls = plugin_adapter.get_adapter_class() for member in BasePlatformAdapter.__abstractmethods__: impl = getattr(cls, member, None) assert callable(impl), f"adapter missing abstract member: {member}" # --- L0b: instantiation against the real contract -------------------------- def test_adapter_instantiates_with_real_platform_config(registered_entry): adapter = _make_adapter(registered_entry) assert isinstance(adapter, BasePlatformAdapter) # super().__init__ must receive the Platform enum (auto-created member), # not a bare string — the gateway calls Platform methods on adapter.platform. assert isinstance(adapter.platform, Platform) assert adapter.platform.value == "meshtastic" def test_registered_entry_carries_radio_limits(registered_entry): assert registered_entry.max_message_length == plugin_adapter.MAX_LORA_MESSAGE_LENGTH assert registered_entry.emoji == "📻" # --- L1: gateway wiring replay --------------------------------------------- def test_adapter_survives_gateway_wiring(registered_entry): """Replay run.py's startup wiring; any missing setter crashes the gateway.""" adapter = _make_adapter(registered_entry) for setter in GATEWAY_WIRING_SETTERS: fn = getattr(adapter, setter, None) assert callable(fn), ( f"adapter missing callable '{setter}' — gateway startup crashes here " "(the original set_message_handler AttributeError)" ) fn(_noop) # run.py also pokes these attributes directly after wiring. adapter._busy_text_mode = False # --- L2: signature conformance ---------------------------------------------- def test_connect_signature_matches_gateway_contract(): cls = plugin_adapter.get_adapter_class() sig = inspect.signature(cls.connect) assert inspect.iscoroutinefunction(cls.connect) assert "is_reconnect" in sig.parameters, "connect() must accept is_reconnect kwarg" param = sig.parameters["is_reconnect"] assert param.kind == inspect.Parameter.KEYWORD_ONLY assert param.default is False def test_send_signature_matches_gateway_contract(): base_sig = inspect.signature(BasePlatformAdapter.send) cls = plugin_adapter.get_adapter_class() sig = inspect.signature(cls.send) base_params = [p for p in base_sig.parameters if p != "self"] own_params = [p for p in sig.parameters if p != "self"] assert own_params == base_params, ( f"send() params {own_params} drift from gateway contract {base_params}" ) assert inspect.iscoroutinefunction(cls.send) # --- L3: inbound event shape against real types ----------------------------- def test_inbound_event_construction_is_valid(registered_entry): """The exact source/event the adapter builds on receive must type-check.""" adapter = _make_adapter(registered_entry) source = adapter.build_source( chat_id="!abcd1234", chat_name="Meshtastic !abcd1234", chat_type="dm", user_id="!abcd1234", user_name="!abcd1234", message_id="42", ) event = MessageEvent( text="hello over LoRa", message_type=MessageType.TEXT, user_id="!abcd1234", user_name="!abcd1234", source=source, raw_message={"id": 42}, message_id="42", ) assert event.source.platform.value == "meshtastic" assert event.text == "hello over LoRa" def test_pki_direct_message_without_channel_reaches_gateway( registered_entry, monkeypatch ): """PKI DMs omit channel and must bypass the secondary-channel filter.""" adapter = _make_adapter(registered_entry) adapter._loop = SimpleNamespace(is_running=lambda: True) captured = {} async def handle_message(event): captured["event"] = event def submit(coro, loop): captured["loop"] = loop with pytest.raises(StopIteration): coro.send(None) adapter.handle_message = handle_message monkeypatch.setattr(plugin_adapter.asyncio, "run_coroutine_threadsafe", submit) interface = SimpleNamespace(myInfo=SimpleNamespace(my_node_num=0x1BA1BC60)) adapter._on_meshtastic_receive( { "from": 0x1BA1496C, "to": 0x1BA1BC60, "toId": "!1ba1bc60", "pkiEncrypted": True, "decoded": {"text": "hello over PKI"}, "id": 42, }, interface, ) assert captured["loop"] is adapter._loop assert captured["event"].text == "hello over PKI" assert captured["event"].source.chat_id == "!1ba1496c"