- matrix.py: Transport implementation over the Matrix client-server HTTP API directly (token auth, /sync long-poll, /send) using the existing httpx dependency — no Matrix SDK; four endpoints total. Deviation from the presented plan (matrix-nio) flagged in the PR - First sync learns position and skips backlog, so room history is never delivered as context (A1); own messages and non-message events filtered; sender MXID passed through for the orchestrator's operator allowlist - Sync loop backs off (capped at 60s) and retries indefinitely on homeserver errors, mirroring B2's polling posture - Encrypted rooms unsupported: bot rooms are created unencrypted until E2E lands (operator decision; room recreated accordingly) - NISSE_MATRIX_TEST_ROOM_ID added: the live smoke test binds only to the dedicated test room, never the production room - 8 unit tests over a mocked HTTP layer; live integration test verified against the real homeserver Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
163 lines
5.3 KiB
Python
163 lines
5.3 KiB
Python
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
|