diff --git a/.env.example b/.env.example index fef6a29..6992f72 100644 --- a/.env.example +++ b/.env.example @@ -16,4 +16,6 @@ NISSE_MATRIX_HOMESERVER=https://matrix.example NISSE_MATRIX_USER_ID=@nisse-dev:example NISSE_MATRIX_TOKEN=syt_your_matrix_access_token NISSE_MATRIX_ROOM_ID=!roomid:example +# Dedicated room for the live integration test; never the production room. +NISSE_MATRIX_TEST_ROOM_ID=!testroomid:example NISSE_MATRIX_OPERATORS=@you:example diff --git a/nisse-dev/src/nisse_dev/matrix.py b/nisse-dev/src/nisse_dev/matrix.py new file mode 100644 index 0000000..19366b5 --- /dev/null +++ b/nisse-dev/src/nisse_dev/matrix.py @@ -0,0 +1,167 @@ +"""MatrixTransport (spec A1): Matrix as strictly a UI channel. + +Implements the Transport interface directly over the Matrix client-server HTTP +API (token auth, /sync long-poll, /send) with the httpx client already in the +dependency tree — no Matrix SDK. The surface used is four endpoints; an SDK +earns its keep when E2E arrives, and bot rooms are created unencrypted until +then, so encrypted rooms are not supported here. + +Inbound handling follows B2's polling posture: on homeserver errors the sync +loop backs off (capped) and retries indefinitely — an outage delays reactions +but never fails anything. +""" + +import os +import threading +import uuid +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from typing import ClassVar + +import httpx + +from nisse_dev.transport import InboundMessage + + +class TransportError(RuntimeError): + pass + + +@dataclass(frozen=True) +class MatrixConfig: + homeserver: str + user_id: str + token: str + room_id: str + + _ENV: ClassVar[tuple[str, ...]] = ( + "NISSE_MATRIX_HOMESERVER", + "NISSE_MATRIX_USER_ID", + "NISSE_MATRIX_TOKEN", + ) + + @classmethod + def from_env( + cls, + env: Mapping[str, str] = os.environ, + room_var: str = "NISSE_MATRIX_ROOM_ID", + ) -> "MatrixConfig": + needed = (*cls._ENV, room_var) + missing = [var for var in needed if not env.get(var)] + if missing: + raise TransportError(f"matrix config incomplete; set: {', '.join(missing)}") + return cls( + homeserver=env["NISSE_MATRIX_HOMESERVER"].rstrip("/"), + user_id=env["NISSE_MATRIX_USER_ID"], + token=env["NISSE_MATRIX_TOKEN"], + room_id=env[room_var], + ) + + +class MatrixTransport: + def __init__( + self, + config: MatrixConfig, + *, + poll_timeout_ms: int = 30000, + http_transport: httpx.BaseTransport | None = None, + ) -> None: + self._config = config + self._poll_timeout_ms = poll_timeout_ms + self._http = httpx.Client( + base_url=f"{config.homeserver}/_matrix/client/v3", + headers={"Authorization": f"Bearer {config.token}"}, + timeout=(poll_timeout_ms / 1000) + 30, + transport=http_transport, + ) + self._callback: Callable[[InboundMessage], None] | None = None + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + # -- outbound --------------------------------------------------------- + + def send_message(self, text: str) -> None: + self._send(text) + + def send_approval_request(self, text: str) -> None: + self._send(text) + + def ask_question(self, text: str) -> None: + self._send(text) + + def _send(self, text: str) -> None: + room = self._config.room_id + txn = uuid.uuid4().hex + try: + response = self._http.put( + f"/rooms/{room}/send/m.room.message/{txn}", + json={"msgtype": "m.text", "body": text}, + ) + except httpx.HTTPError as exc: + raise TransportError(f"matrix send failed: {exc}") from exc + if response.status_code != 200: + raise TransportError( + f"matrix send returned {response.status_code}: {response.text[:300]}" + ) + + # -- inbound ---------------------------------------------------------- + + def on_message(self, callback: Callable[[InboundMessage], None]) -> None: + self._callback = callback + + def start(self) -> None: + """Begin delivering new room messages to the registered callback. + History from before start() is never delivered (Matrix room history + must not become load-bearing context, A1).""" + if self._callback is None: + raise TransportError("no message handler registered; call on_message first") + self._stop.clear() + self._thread = threading.Thread(target=self._sync_loop, daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=(self._poll_timeout_ms / 1000) + 5) + self._thread = None + + def _sync_loop(self) -> None: + since: str | None = None + backoff = 1.0 + while not self._stop.is_set(): + params: dict[str, str] = {} + if since is None: + params["timeout"] = "0" # first sync: learn position, skip backlog + else: + params["timeout"] = str(self._poll_timeout_ms) + params["since"] = since + try: + response = self._http.get("/sync", params=params) + response.raise_for_status() + except httpx.HTTPError: + self._stop.wait(backoff) + backoff = min(backoff * 2, 60.0) + continue + backoff = 1.0 + data = response.json() + if since is not None: + self._deliver(data) + since = data.get("next_batch", since) + + def _deliver(self, sync_data: dict) -> None: + room = sync_data.get("rooms", {}).get("join", {}).get(self._config.room_id, {}) + for event in room.get("timeline", {}).get("events", []): + if event.get("type") != "m.room.message": + continue + sender = event.get("sender", "") + if sender == self._config.user_id: + continue + body = event.get("content", {}).get("body") + if not body: + continue + assert self._callback is not None + self._callback(InboundMessage(sender=sender, text=body)) + + def close(self) -> None: + self.stop() + self._http.close() diff --git a/nisse-dev/tests/test_matrix.py b/nisse-dev/tests/test_matrix.py new file mode 100644 index 0000000..2e1fd69 --- /dev/null +++ b/nisse-dev/tests/test_matrix.py @@ -0,0 +1,162 @@ +import json +import threading + +import httpx +import pytest + +from nisse_dev.matrix import MatrixConfig, MatrixTransport, TransportError +from nisse_dev.transport import InboundMessage + +CONFIG = MatrixConfig( + homeserver="https://matrix.test", + user_id="@nisse-dev:test", + token="syt_secret", + room_id="!room:test", +) + + +def sync_body(events: list[dict], next_batch: str) -> dict: + return { + "next_batch": next_batch, + "rooms": {"join": {"!room:test": {"timeline": {"events": events}}}}, + } + + +def message_event(sender: str, body: str) -> dict: + return { + "type": "m.room.message", + "sender": sender, + "content": {"msgtype": "m.text", "body": body}, + } + + +def test_send_message_puts_event_with_auth_and_unique_txn() -> None: + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(200, json={"event_id": "$e"}) + + transport = MatrixTransport(CONFIG, http_transport=httpx.MockTransport(handler)) + transport.send_message("hello") + transport.send_approval_request("plan rev-1") + + first, second = seen + assert first.headers["authorization"] == "Bearer syt_secret" + assert "/rooms/!room:test/send/m.room.message/" in str(first.url) + assert json.loads(first.content) == {"msgtype": "m.text", "body": "hello"} + assert str(first.url) != str(second.url) # txn ids must be unique + + +def test_send_failure_raises_transport_error() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(403, text="forbidden") + + transport = MatrixTransport(CONFIG, http_transport=httpx.MockTransport(handler)) + with pytest.raises(TransportError, match="403"): + transport.send_message("hello") + + +class SyncScript: + """Serves scripted /sync responses, then endless empty ones.""" + + def __init__(self, responses: list[dict]) -> None: + self._responses = responses + self._i = 0 + self.batches_seen: list[str | None] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/sync") + self.batches_seen.append(request.url.params.get("since")) + if self._i < len(self._responses): + body = self._responses[self._i] + self._i += 1 + return httpx.Response(200, json=body) + return httpx.Response(200, json=sync_body([], f"b{self._i}")) + + +def collect_messages( + responses: list[dict], expected: int, timeout_s: float = 5.0 +) -> list[InboundMessage]: + received: list[InboundMessage] = [] + done = threading.Event() + + def on_message(msg: InboundMessage) -> None: + received.append(msg) + if len(received) >= expected: + done.set() + + transport = MatrixTransport( + CONFIG, + poll_timeout_ms=10, + http_transport=httpx.MockTransport(SyncScript(responses)), + ) + transport.on_message(on_message) + transport.start() + done.wait(timeout_s) + transport.close() + return received + + +def test_listener_delivers_operator_messages() -> None: + responses = [ + sync_body([message_event("@don:test", "backlog: never delivered")], "b1"), + sync_body([message_event("@don:test", "approve")], "b2"), + ] + received = collect_messages(responses, expected=1) + assert received == [InboundMessage(sender="@don:test", text="approve")] + + +def test_own_and_non_message_events_are_filtered() -> None: + responses = [ + sync_body([], "b1"), + sync_body( + [ + message_event("@nisse-dev:test", "my own echo"), + {"type": "m.room.member", "sender": "@don:test", "content": {}}, + message_event("@don:test", "status"), + ], + "b2", + ), + ] + received = collect_messages(responses, expected=1) + assert received == [InboundMessage(sender="@don:test", text="status")] + + +def test_first_sync_learns_position_without_delivering() -> None: + script = SyncScript([sync_body([message_event("@don:test", "old")], "b1")]) + transport = MatrixTransport( + CONFIG, poll_timeout_ms=10, http_transport=httpx.MockTransport(script) + ) + got = threading.Event() + transport.on_message(lambda msg: got.set()) + transport.start() + assert not got.wait(0.3) + transport.close() + assert script.batches_seen[0] is None # initial sync carries no since + assert "b1" in script.batches_seen # later polls resume from its batch + + +def test_start_without_handler_raises() -> None: + transport = MatrixTransport( + CONFIG, http_transport=httpx.MockTransport(lambda r: httpx.Response(200)) + ) + with pytest.raises(TransportError, match="no message handler"): + transport.start() + + +def test_config_from_env_reports_missing() -> None: + with pytest.raises(TransportError, match="NISSE_MATRIX_TOKEN"): + MatrixConfig.from_env({"NISSE_MATRIX_HOMESERVER": "https://m"}) + + +def test_config_from_env_custom_room_var() -> None: + env = { + "NISSE_MATRIX_HOMESERVER": "https://m/", + "NISSE_MATRIX_USER_ID": "@nisse-dev:test", + "NISSE_MATRIX_TOKEN": "t", + "NISSE_MATRIX_TEST_ROOM_ID": "!t:test", + } + config = MatrixConfig.from_env(env, room_var="NISSE_MATRIX_TEST_ROOM_ID") + assert config.room_id == "!t:test" + assert config.homeserver == "https://m" # trailing slash stripped diff --git a/tests/integration/test_matrix_live.py b/tests/integration/test_matrix_live.py new file mode 100644 index 0000000..26c5b52 --- /dev/null +++ b/tests/integration/test_matrix_live.py @@ -0,0 +1,49 @@ +"""Live smoke test against the real homeserver, bound strictly to the +dedicated test room (never the production room). Skips unless the Matrix env +vars, including NISSE_MATRIX_TEST_ROOM_ID, are present.""" + +import os +import uuid +from pathlib import Path + +import pytest + +from nisse_dev.envfile import load_env_file +from nisse_dev.matrix import MatrixConfig, MatrixTransport + +load_env_file(Path(__file__).resolve().parents[2] / ".env") + +REQUIRED = ( + "NISSE_MATRIX_HOMESERVER", + "NISSE_MATRIX_USER_ID", + "NISSE_MATRIX_TOKEN", + "NISSE_MATRIX_TEST_ROOM_ID", +) + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not all(os.environ.get(var) for var in REQUIRED), + reason=f"live matrix config absent; set {', '.join(REQUIRED)}", + ), +] + + +def test_send_lands_in_test_room() -> None: + config = MatrixConfig.from_env(room_var="NISSE_MATRIX_TEST_ROOM_ID") + marker = f"transport smoke test {uuid.uuid4().hex}" + transport = MatrixTransport(config) + try: + transport.send_message(marker) + response = transport._http.get( + f"/rooms/{config.room_id}/messages", + params={"dir": "b", "limit": "10"}, + ) + response.raise_for_status() + bodies = [ + event.get("content", {}).get("body") + for event in response.json().get("chunk", []) + ] + assert marker in bodies + finally: + transport.close()