"""Tests for Matrix platform adapter (mautrix-python backend).""" import asyncio import re import stat import sys import time import types import pytest from unittest.mock import MagicMock, patch, AsyncMock from gateway.config import Platform, PlatformConfig from gateway.platforms.base import MessageType def _make_fake_mautrix(): """Create a lightweight set of fake ``mautrix`` modules. The adapter does ``from mautrix.api import HTTPAPI``, ``from mautrix.client import Client``, ``from mautrix.types import ...`` at import time and inside methods. We provide just enough stubs for tests that need to mock the mautrix import chain. Use via ``patch.dict("sys.modules", _make_fake_mautrix())``. """ # --- mautrix (root) --- mautrix = types.ModuleType("mautrix") # --- mautrix.api --- mautrix_api = types.ModuleType("mautrix.api") class HTTPAPI: def __init__(self, base_url="", token="", **kwargs): self.base_url = base_url self.token = token self.session = MagicMock() self.session.close = AsyncMock() mautrix_api.HTTPAPI = HTTPAPI mautrix.api = mautrix_api # --- mautrix.types --- mautrix_types = types.ModuleType("mautrix.types") class EventType: ROOM_MESSAGE = "m.room.message" REACTION = "m.reaction" ROOM_ENCRYPTED = "m.room.encrypted" ROOM_NAME = "m.room.name" class UserID(str): pass class RoomID(str): pass class EventID(str): pass class ContentURI(str): pass class SyncToken(str): pass class RoomCreatePreset: PRIVATE = "private_chat" PUBLIC = "public_chat" TRUSTED_PRIVATE = "trusted_private_chat" class PresenceState: ONLINE = "online" OFFLINE = "offline" UNAVAILABLE = "unavailable" class TrustState: UNVERIFIED = 0 VERIFIED = 1 class PaginationDirection: BACKWARD = "b" FORWARD = "f" mautrix_types.EventType = EventType mautrix_types.UserID = UserID mautrix_types.RoomID = RoomID mautrix_types.EventID = EventID mautrix_types.ContentURI = ContentURI mautrix_types.SyncToken = SyncToken mautrix_types.RoomCreatePreset = RoomCreatePreset mautrix_types.PresenceState = PresenceState mautrix_types.TrustState = TrustState mautrix_types.PaginationDirection = PaginationDirection mautrix.types = mautrix_types # --- mautrix.client --- mautrix_client = types.ModuleType("mautrix.client") class Client: def __init__(self, mxid=None, device_id=None, api=None, state_store=None, sync_store=None, **kwargs): self.mxid = mxid self.device_id = device_id self.api = api self.state_store = state_store self.sync_store = sync_store self.crypto = None self._event_handlers = {} def add_event_handler(self, event_type, handler, **kwargs): self._event_handlers.setdefault(event_type, []).append(handler) def add_dispatcher(self, dispatcher_type): pass class InternalEventType: INVITE = "internal.invite" mautrix_client.Client = Client mautrix_client.InternalEventType = InternalEventType mautrix.client = mautrix_client # --- mautrix.client.dispatcher --- mautrix_client_dispatcher = types.ModuleType("mautrix.client.dispatcher") class MembershipEventDispatcher: pass mautrix_client_dispatcher.MembershipEventDispatcher = MembershipEventDispatcher # --- mautrix.client.state_store --- mautrix_client_state_store = types.ModuleType("mautrix.client.state_store") class MemoryStateStore: async def get_member(self, room_id, user_id): return None async def get_members(self, room_id): return [] async def get_member_profiles(self, room_id): return {} class MemorySyncStore: def __init__(self): self.next_batch = None async def get_next_batch(self): return self.next_batch async def put_next_batch(self, token): self.next_batch = token mautrix_client_state_store.MemoryStateStore = MemoryStateStore mautrix_client_state_store.MemorySyncStore = MemorySyncStore # --- mautrix.crypto --- mautrix_crypto = types.ModuleType("mautrix.crypto") class OlmMachine: def __init__(self, client=None, crypto_store=None, state_store=None): self.share_keys_min_trust = None self.send_keys_min_trust = None async def load(self): pass async def share_keys(self): pass async def decrypt_megolm_event(self, event): return event mautrix_crypto.OlmMachine = OlmMachine # --- mautrix.crypto.store --- mautrix_crypto_store = types.ModuleType("mautrix.crypto.store") class MemoryCryptoStore: def __init__(self, account_id="", pickle_key=""): # noqa: S301 self.account_id = account_id self.pickle_key = pickle_key mautrix_crypto_store.MemoryCryptoStore = MemoryCryptoStore # --- mautrix.crypto.attachments --- mautrix_crypto_attachments = types.ModuleType("mautrix.crypto.attachments") def encrypt_attachment(data): encrypted_file = MagicMock() encrypted_file.serialize.return_value = { "key": {"k": "testkey"}, "iv": "testiv", "hashes": {"sha256": "testhash"}, "v": "v2", } return (b"ciphertext_" + data, encrypted_file) mautrix_crypto_attachments.encrypt_attachment = encrypt_attachment # --- mautrix.crypto.store.asyncpg --- mautrix_crypto_store_asyncpg = types.ModuleType("mautrix.crypto.store.asyncpg") class PgCryptoStore: upgrade_table = MagicMock() def __init__(self, account_id="", pickle_key="", db=None): # noqa: S301 self.account_id = account_id self.pickle_key = pickle_key self.db = db self._device_id = "" async def open(self): pass async def put_device_id(self, device_id): self._device_id = device_id mautrix_crypto_store_asyncpg.PgCryptoStore = PgCryptoStore # --- mautrix.util --- mautrix_util = types.ModuleType("mautrix.util") # --- mautrix.util.async_db --- mautrix_util_async_db = types.ModuleType("mautrix.util.async_db") class Database: @classmethod def create(cls, url, upgrade_table=None): db = MagicMock() db.start = AsyncMock() db.stop = AsyncMock() return db mautrix_util_async_db.Database = Database return { "mautrix": mautrix, "mautrix.api": mautrix_api, "mautrix.types": mautrix_types, "mautrix.client": mautrix_client, "mautrix.client.dispatcher": mautrix_client_dispatcher, "mautrix.client.state_store": mautrix_client_state_store, "mautrix.crypto": mautrix_crypto, "mautrix.crypto.attachments": mautrix_crypto_attachments, "mautrix.crypto.store": mautrix_crypto_store, "mautrix.crypto.store.asyncpg": mautrix_crypto_store_asyncpg, "mautrix.util": mautrix_util, "mautrix.util.async_db": mautrix_util_async_db, } # --------------------------------------------------------------------------- # Platform & Config # --------------------------------------------------------------------------- class TestMatrixConfigLoading: def test_apply_env_overrides_with_password(self, monkeypatch): monkeypatch.delenv("MATRIX_ACCESS_TOKEN", raising=False) monkeypatch.setenv("MATRIX_PASSWORD", "secret123") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.setenv("MATRIX_USER_ID", "@bot:example.org") from gateway.config import GatewayConfig, _apply_env_overrides config = GatewayConfig() _apply_env_overrides(config) assert Platform.MATRIX in config.platforms mc = config.platforms[Platform.MATRIX] assert mc.enabled is True assert mc.extra.get("password") == "secret123" assert mc.extra.get("user_id") == "@bot:example.org" def test_matrix_e2ee_mode_optional_sets_config(self, monkeypatch): monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_abc123") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.setenv("MATRIX_E2EE_MODE", "optional") monkeypatch.delenv("MATRIX_ENCRYPTION", raising=False) from gateway.config import GatewayConfig, _apply_env_overrides config = GatewayConfig() _apply_env_overrides(config) mc = config.platforms[Platform.MATRIX] assert mc.extra.get("encryption") is True assert mc.extra.get("e2ee_mode") == "optional" def test_matrix_home_room(self, monkeypatch): monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_abc123") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.setenv("MATRIX_HOME_ROOM", "!room123:example.org") monkeypatch.setenv("MATRIX_HOME_ROOM_NAME", "Bot Room") from gateway.config import GatewayConfig, _apply_env_overrides config = GatewayConfig() _apply_env_overrides(config) home = config.get_home_channel(Platform.MATRIX) assert home is not None assert home.chat_id == "!room123:example.org" assert home.name == "Bot Room" # --------------------------------------------------------------------------- # Adapter helpers # --------------------------------------------------------------------------- def _make_adapter(): """Create a MatrixAdapter with mocked config.""" from plugins.platforms.matrix.adapter import MatrixAdapter config = PlatformConfig( enabled=True, token="syt_test_token", extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", }, ) adapter = MatrixAdapter(config) return adapter # --------------------------------------------------------------------------- # Typing indicator # --------------------------------------------------------------------------- class TestMatrixTypingIndicator: def setup_method(self): self.adapter = _make_adapter() self.adapter._client = MagicMock() self.adapter._client.set_typing = AsyncMock() @pytest.mark.asyncio async def test_stop_typing_clears_matrix_typing_state(self): """stop_typing() should send typing=false instead of waiting for timeout expiry.""" from plugins.platforms.matrix.adapter import RoomID await self.adapter.stop_typing("!room:example.org") self.adapter._client.set_typing.assert_awaited_once_with( RoomID("!room:example.org"), timeout=0, ) # --------------------------------------------------------------------------- # mxc:// URL conversion # --------------------------------------------------------------------------- class TestMatrixMxcToHttp: def setup_method(self): self.adapter = _make_adapter() def test_mxc_with_different_server(self): """mxc:// from a different server should still use our homeserver.""" mxc = "mxc://other.server/media456" result = self.adapter._mxc_to_http(mxc) assert result.startswith("https://matrix.example.org/") assert "other.server/media456" in result # --------------------------------------------------------------------------- # DM detection # --------------------------------------------------------------------------- class TestMatrixDmDetection: def setup_method(self): self.adapter = _make_adapter() def test_room_in_m_direct_is_dm(self): """A room listed in m.direct should be detected as DM.""" self.adapter._joined_rooms = {"!dm_room:ex.org", "!group_room:ex.org"} self.adapter._dm_rooms = { "!dm_room:ex.org": True, "!group_room:ex.org": False, } assert self.adapter._dm_rooms.get("!dm_room:ex.org") is True assert self.adapter._dm_rooms.get("!group_room:ex.org") is False def test_unknown_room_not_in_cache(self): """Unknown rooms should not be in the DM cache.""" self.adapter._dm_rooms = {} assert self.adapter._dm_rooms.get("!unknown:ex.org") is None @pytest.mark.asyncio async def test_named_two_member_dm_is_dm(self): """A named two-member room in m.direct is a DM (not a room). Most Matrix clients auto-name DM rooms (e.g. "Alice & Bot"), so the old `not has_explicit_name` override misclassified them as rooms. """ self.adapter._joined_rooms = {"!named_dm:ex.org"} self.adapter._dm_rooms = {"!named_dm:ex.org": True} self.adapter._client = MagicMock() self.adapter._client.get_state_event = AsyncMock( side_effect=lambda room_id, event_type: {"name": "Alice & Bot"} if event_type == "m.room.name" else (_ for _ in ()).throw(Exception("no alias")) ) self.adapter._client.state_store = MagicMock() self.adapter._client.state_store.get_members = AsyncMock( return_value=["@bot:ex.org", "@alice:ex.org"] ) identity = await self.adapter._resolve_room_identity("!named_dm:ex.org") assert identity.chat_type == "dm" assert identity.conflict is False assert identity.joined_member_count == 2 assert await self.adapter._is_dm_room("!named_dm:ex.org") is True # --------------------------------------------------------------------------- # Reply fallback stripping # --------------------------------------------------------------------------- class TestMatrixReplyFallbackStripping: """Test that Matrix reply fallback lines ('> ' prefix) are stripped.""" def setup_method(self): self.adapter = _make_adapter() self.adapter._user_id = "@bot:example.org" self.adapter._startup_ts = 0.0 self.adapter._dm_rooms = {} self.adapter._message_handler = AsyncMock() def _strip_fallback(self, body: str, has_reply: bool = True) -> str: """Simulate the reply fallback stripping logic from _on_room_message.""" reply_to = "some_event_id" if has_reply else None if reply_to and body.startswith("> "): lines = body.split("\n") stripped = [] past_fallback = False for line in lines: if not past_fallback: if line.startswith("> ") or line == ">": continue if line == "": past_fallback = True continue past_fallback = True stripped.append(line) body = "\n".join(stripped) if stripped else body return body def test_simple_reply_fallback(self): body = "> <@alice:ex.org> Original message\n\nActual reply" result = self._strip_fallback(body) assert result == "Actual reply" def test_multiline_reply_fallback(self): body = "> <@alice:ex.org> Line 1\n> Line 2\n\nMy response" result = self._strip_fallback(body) assert result == "My response" # --------------------------------------------------------------------------- # Matrix-friendly command aliases # --------------------------------------------------------------------------- class TestMatrixBangCommandAlias: """Matrix clients may reserve /commands, so Hermes supports !commands.""" def setup_method(self): self.adapter = _make_adapter() self.adapter._is_dm_room = AsyncMock(return_value=True) self.adapter._get_display_name = AsyncMock(return_value="Alice") self.adapter._background_read_receipt = MagicMock() self.adapter._text_batch_delay_seconds = 0 async def _dispatch_text(self, body: str, *, is_dm: bool = True): captured_event = None self.adapter._is_dm_room = AsyncMock(return_value=is_dm) self.adapter._require_mention = True self.adapter._free_rooms = set() async def capture(msg_event): nonlocal captured_event captured_event = msg_event self.adapter.handle_message = capture await self.adapter._handle_text_message( room_id="!room:example.org", sender="@alice:example.org", event_id="$matrix-command-test", event_ts=0.0, source_content={"msgtype": "m.text", "body": body}, relates_to={}, ) return captured_event async def _dispatch_text_reply(self, body: str, *, is_dm: bool = True): """Dispatch a message that is a Matrix reply (m.in_reply_to set), so the reply-fallback quote stripping path runs before command detection. """ captured_event = None self.adapter._is_dm_room = AsyncMock(return_value=is_dm) self.adapter._require_mention = True self.adapter._free_rooms = set() async def capture(msg_event): nonlocal captured_event captured_event = msg_event self.adapter.handle_message = capture await self.adapter._handle_text_message( room_id="!room:example.org", sender="@alice:example.org", event_id="$matrix-reply-command-test", event_ts=0.0, source_content={"msgtype": "m.text", "body": body}, relates_to={"m.in_reply_to": {"event_id": "$parent-event"}}, ) return captured_event def test_known_bang_command_normalizes_to_slash_command(self): from plugins.platforms.matrix.adapter import _normalize_matrix_bang_command assert _normalize_matrix_bang_command("!model") == "/model" assert ( _normalize_matrix_bang_command("!queue continue the plan") == "/queue continue the plan" ) assert ( _normalize_matrix_bang_command("!btw research this") == "/btw research this" ) assert _normalize_matrix_bang_command("!tasks") == "/tasks" @pytest.mark.asyncio async def test_unknown_bang_text_stays_normal_text(self): captured_event = await self._dispatch_text("!important note") assert captured_event is not None assert captured_event.text == "!important note" assert captured_event.message_type == MessageType.TEXT assert captured_event.get_command() is None def test_bang_skill_command_normalizes(self): """The get_skill_commands() branch normalizes installed skill commands, not just built-in gateway commands. Skill keys are stored slash-prefixed (e.g. "/arxiv"), which the resolver must account for.""" import agent.skill_commands as skill_commands_mod fake_skills = {"/arxiv": {}, "/obsidian": {}} with patch.object( skill_commands_mod, "get_skill_commands", return_value=fake_skills ): from plugins.platforms.matrix.adapter import _normalize_matrix_bang_command # is_gateway_known_command won't know these; the skill branch must. assert _normalize_matrix_bang_command("!arxiv") == "/arxiv" assert ( _normalize_matrix_bang_command("!obsidian search foo") == "/obsidian search foo" ) # A name in neither registry stays plain text. assert ( _normalize_matrix_bang_command("!definitelynotacommand") == "!definitelynotacommand" ) @pytest.mark.asyncio async def test_slash_command_in_quoted_reply_normalizes(self): """Sanity: the slash equivalent already works post-strip — the bang form above must reach parity with this.""" captured_event = await self._dispatch_text_reply( "> <@bob:example.org> earlier message\n\n/model" ) assert captured_event is not None assert captured_event.text == "/model" assert captured_event.message_type == MessageType.COMMAND # --------------------------------------------------------------------------- # Thread detection # --------------------------------------------------------------------------- class TestMatrixThreadDetection: def test_no_thread_for_edit(self): """m.replace relation should not set thread_id.""" relates_to = { "rel_type": "m.replace", "event_id": "$edited_event", } thread_id = None if relates_to.get("rel_type") == "m.thread": thread_id = relates_to.get("event_id") assert thread_id is None # --------------------------------------------------------------------------- # Format message # --------------------------------------------------------------------------- class TestMatrixFormatMessage: def setup_method(self): self.adapter = _make_adapter() def test_image_markdown_stripped(self): """![alt](url) should be converted to just the URL.""" result = self.adapter.format_message("![cat](https://img.example.com/cat.png)") assert result == "https://img.example.com/cat.png" # --------------------------------------------------------------------------- # Rendering payloads # --------------------------------------------------------------------------- class TestMatrixRenderingPayloads: def setup_method(self): self.adapter = _make_adapter() self.mock_client = MagicMock() self.mock_client.send_message_event = AsyncMock(return_value="$evt") self.adapter._client = self.mock_client def _sent_contents(self): return [ call.args[2] if len(call.args) > 2 else call.kwargs["content"] for call in self.mock_client.send_message_event.await_args_list ] @pytest.mark.asyncio async def test_thread_payload_uses_m_thread_with_reply_fallback(self): result = await self.adapter.send( "!room:example.org", "threaded", metadata={"thread_id": "$root"}, ) assert result.success is True relates_to = self._sent_contents()[0]["m.relates_to"] assert relates_to == { "rel_type": "m.thread", "event_id": "$root", "is_falling_back": True, "m.in_reply_to": {"event_id": "$root"}, } @pytest.mark.asyncio async def test_long_response_split_preserves_thread_context(self): # Build a payload guaranteed to exceed the adapter's outbound chunk # size (configurable since #53026) so send() must split it. repeats = (self.adapter.max_message_length // 15) + 200 long_text = "Intro\n```python\n" + ("print('hello')\n" * repeats) + "```\nDone" result = await self.adapter.send( "!room:example.org", long_text, metadata={"thread_id": "$root"}, ) assert result.success is True contents = self._sent_contents() assert len(contents) > 1 for content in contents: assert content["m.relates_to"]["rel_type"] == "m.thread" assert content["m.relates_to"]["event_id"] == "$root" assert content["m.relates_to"]["m.in_reply_to"] == {"event_id": "$root"} assert content["body"].count("```") % 2 == 0 # --------------------------------------------------------------------------- # Markdown to HTML conversion # --------------------------------------------------------------------------- class TestMatrixMarkdownToHtml: def setup_method(self): self.adapter = _make_adapter() def test_bold_conversion(self): """**bold** should produce tags.""" result = self.adapter._markdown_to_html("**bold**") assert "" in result or "" in result assert "bold" in result def test_italic_conversion(self): """*italic* should produce tags.""" result = self.adapter._markdown_to_html("*italic*") assert "" in result or "" in result def test_inline_code(self): """`code` should produce tags.""" result = self.adapter._markdown_to_html("`code`") assert "" in result def test_plain_text_returns_html(self): """Plain text should still be returned (possibly with
or

).""" result = self.adapter._markdown_to_html("Hello world") assert "Hello world" in result def test_matrix_markdown_preserves_table_structure(self): table = "\n".join( [ "| Item | Quantity |", "| --- | --- |", "| Apples | 4 |", "| Bread | 1 |", ] ) result = self.adapter._markdown_to_html(table) assert "" in result assert "" in result assert "" in result assert "" in result assert "" in result # --------------------------------------------------------------------------- # Helper: display name extraction # --------------------------------------------------------------------------- class TestMatrixDisplayName: def setup_method(self): self.adapter = _make_adapter() @pytest.mark.asyncio async def test_get_display_name_from_state_store(self): """Should get display name from state_store.get_member().""" mock_member = MagicMock() mock_member.displayname = "Alice" mock_state_store = MagicMock() mock_state_store.get_member = AsyncMock(return_value=mock_member) mock_client = MagicMock() mock_client.state_store = mock_state_store self.adapter._client = mock_client name = await self.adapter._get_display_name("!room:ex.org", "@alice:ex.org") assert name == "Alice" # --------------------------------------------------------------------------- # Requirements check # --------------------------------------------------------------------------- class TestMatrixModuleImport: def test_module_importable_without_mautrix(self): """plugins.platforms.matrix.adapter must be importable even when mautrix is not installed — otherwise the gateway crashes for ALL platforms. This test uses a subprocess to avoid polluting the current process's sys.modules (reimporting a module creates a second module object whose classes don't share globals with the original — breaking patch.object in subsequent tests). """ import subprocess result = subprocess.run( [sys.executable, "-c", ( "import sys\n" "# Block mautrix completely\n" "class _Blocker:\n" " def find_module(self, name, path=None):\n" " if name.startswith('mautrix'): return self\n" " def load_module(self, name):\n" " raise ImportError(f'blocked: {name}')\n" "sys.meta_path.insert(0, _Blocker())\n" "for k in list(sys.modules):\n" " if k.startswith('mautrix'): del sys.modules[k]\n" "from unittest.mock import patch\n" "from plugins.platforms.matrix.adapter import check_matrix_requirements\n" "with patch('tools.lazy_deps.ensure', side_effect=ImportError('blocked')):\n" " assert not check_matrix_requirements()\n" "print('OK')\n" )], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0, ( f"Subprocess failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" ) class TestMatrixRequirements: def test_check_requirements_encryption_true_no_e2ee_deps(self, monkeypatch): """MATRIX_ENCRYPTION=true should fail if python-olm is not installed.""" monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.setenv("MATRIX_ENCRYPTION", "true") import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False), \ patch("tools.lazy_deps.feature_missing", return_value=()): assert matrix_mod.check_matrix_requirements() is False def test_check_requirements_e2ee_optional_no_deps_ok(self, monkeypatch): """MATRIX_E2EE_MODE=optional should not block startup without python-olm.""" monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.setenv("MATRIX_E2EE_MODE", "optional") monkeypatch.delenv("MATRIX_ENCRYPTION", raising=False) import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False), \ patch("tools.lazy_deps.feature_missing", return_value=()), \ patch("tools.lazy_deps.ensure_and_bind", return_value=True): assert matrix_mod.check_matrix_requirements() is True def test_check_requirements_encryption_false_no_e2ee_deps_ok(self, monkeypatch): """Without encryption, missing E2EE deps should not block startup.""" monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.delenv("MATRIX_ENCRYPTION", raising=False) import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False), \ patch("tools.lazy_deps.feature_missing", return_value=()): assert matrix_mod.check_matrix_requirements() is True def test_check_requirements_encryption_true_with_e2ee_deps(self, monkeypatch): """MATRIX_ENCRYPTION=true should pass if E2EE deps are available.""" monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.setenv("MATRIX_ENCRYPTION", "true") import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True), \ patch("tools.lazy_deps.feature_missing", return_value=()): assert matrix_mod.check_matrix_requirements() is True def test_check_e2ee_deps_requires_asyncpg(self, monkeypatch): """E2EE deps check must reject when asyncpg is missing — even if olm is present. Regression for #31116: ``mautrix[encryption]`` extra installs python-olm but NOT asyncpg/aiosqlite, which are required by mautrix's crypto store at connect time. ``_check_e2ee_deps`` previously only tested ``OlmMachine`` import and returned True, so the failure manifested as a confusing ``No module named 'asyncpg'`` deep in ``MatrixAdapter.connect()``. """ from plugins.platforms.matrix.adapter import _check_e2ee_deps import builtins real_import = builtins.__import__ def _blocking_import(name, *args, **kwargs): if name == "asyncpg" or name.startswith("asyncpg."): raise ImportError("blocked for test") return real_import(name, *args, **kwargs) with patch.object(builtins, "__import__", _blocking_import): assert _check_e2ee_deps() is False def test_check_e2ee_deps_requires_aiosqlite(self): """E2EE deps check must reject when aiosqlite is missing. Mautrix's ``Database.create("sqlite:///...")`` driver lookup imports aiosqlite lazily — without it, connect fails at ``crypto_db.start()``. """ from plugins.platforms.matrix.adapter import _check_e2ee_deps import builtins real_import = builtins.__import__ def _blocking_import(name, *args, **kwargs): if name == "aiosqlite" or name.startswith("aiosqlite."): raise ImportError("blocked for test") return real_import(name, *args, **kwargs) with patch.object(builtins, "__import__", _blocking_import): assert _check_e2ee_deps() is False def test_check_requirements_runs_lazy_install_when_partial(self, monkeypatch): """When mautrix is installed but asyncpg/aiosqlite are missing, check_matrix_requirements must still run the lazy installer. Regression for #31116: the previous ``try: import mautrix`` gate short-circuited the install of the OTHER 4 platform.matrix packages, so a partial install (mautrix only) was treated as fully installed. """ monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.delenv("MATRIX_ENCRYPTION", raising=False) import plugins.platforms.matrix.adapter as matrix_mod # Simulate "mautrix installed, asyncpg missing" → feature_missing # returns a non-empty tuple → ensure_and_bind MUST be called. called = {"ensure_and_bind": False} def _fake_ensure_and_bind(feature, importer, target_globals, **kwargs): called["ensure_and_bind"] = True assert feature == "platform.matrix" return True # Pretend install succeeded. with patch("tools.lazy_deps.feature_missing", return_value=("asyncpg==0.31.0",)), \ patch("tools.lazy_deps.ensure_and_bind", side_effect=_fake_ensure_and_bind): matrix_mod.check_matrix_requirements() assert called["ensure_and_bind"], ( "check_matrix_requirements must call ensure_and_bind whenever ANY " "platform.matrix dep is missing, not just when mautrix itself is " "missing (#31116)" ) # --------------------------------------------------------------------------- # Access-token auth / E2EE bootstrap # --------------------------------------------------------------------------- class TestMatrixAccessTokenAuth: @pytest.mark.asyncio async def test_connect_with_access_token_and_encryption(self): """connect() should call whoami, set user_id/device_id, set up crypto.""" from plugins.platforms.matrix.adapter import MatrixAdapter config = PlatformConfig( enabled=True, token="syt_test_access_token", extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", "encryption": True, }, ) adapter = MatrixAdapter(config) class FakeWhoamiResponse: def __init__(self, user_id, device_id): self.user_id = user_id self.device_id = device_id fake_mautrix_mods = _make_fake_mautrix() # Create a mock client that returns from the mautrix.client.Client constructor mock_client = MagicMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = None mock_client.state_store = MagicMock() mock_client.sync_store = MagicMock() mock_client.crypto = None mock_client.whoami = AsyncMock(return_value=FakeWhoamiResponse("@bot:example.org", "DEV123")) mock_client.sync = AsyncMock(return_value={"rooms": {"join": {"!room:server": {}}}}) mock_client.add_event_handler = MagicMock() mock_client.handle_sync = MagicMock(return_value=[]) mock_client.query_keys = AsyncMock(return_value={ "device_keys": {"@bot:example.org": {"DEV123": { "keys": {"ed25519:DEV123": "fake_ed25519_key"}, }}}, }) mock_client.api = MagicMock() mock_client.api.token = "syt_test_access_token" mock_client.api.session = MagicMock() mock_client.api.session.close = AsyncMock() # Mock the crypto setup mock_olm = MagicMock() mock_olm.load = AsyncMock() mock_olm.share_keys = AsyncMock() mock_olm.share_keys_min_trust = None mock_olm.send_keys_min_trust = None mock_olm.account = MagicMock() mock_olm.account.identity_keys = {"ed25519": "fake_ed25519_key"} # Patch Client constructor to return our mock fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client) fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm) import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True): with patch.dict("sys.modules", fake_mautrix_mods): with patch.object(adapter, "_refresh_dm_cache", AsyncMock()): with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)): assert await adapter.connect() is True mock_client.whoami.assert_awaited_once() assert adapter._user_id == "@bot:example.org" await adapter.disconnect() class TestDeviceKeyReVerification: @pytest.mark.asyncio async def test_verify_fails_when_server_keys_mismatch_after_upload(self): """share_keys() succeeds but server still has old keys -> should return False.""" adapter = _make_adapter() mock_client = MagicMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = "TESTDEVICE" # First query: keys missing -> triggers share_keys # Second query: keys still don't match -> should fail mock_keys_missing = MagicMock() mock_keys_missing.device_keys = {"@bot:example.org": {}} mock_keys_mismatch = MagicMock() mock_device = MagicMock() mock_device.keys = {"ed25519:TESTDEVICE": "server_old_key"} mock_keys_mismatch.device_keys = {"@bot:example.org": {"TESTDEVICE": mock_device}} mock_client.query_keys = AsyncMock(side_effect=[mock_keys_missing, mock_keys_mismatch]) mock_olm = MagicMock() mock_olm.account = MagicMock() mock_olm.account.shared = False mock_olm.account.identity_keys = {"ed25519": "local_new_key"} mock_olm.share_keys = AsyncMock() result = await adapter._verify_device_keys_on_server(mock_client, mock_olm) assert result is False mock_olm.share_keys.assert_awaited_once() class TestMatrixE2EEHardFail: """connect() must refuse to start when E2EE is requested but deps are missing.""" @pytest.mark.asyncio async def test_connect_fails_when_encryption_true_but_no_e2ee_deps(self): from plugins.platforms.matrix.adapter import MatrixAdapter config = PlatformConfig( enabled=True, token="syt_test_access_token", extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", "encryption": True, }, ) adapter = MatrixAdapter(config) fake_mautrix_mods = _make_fake_mautrix() mock_client = MagicMock() mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="DEV123")) mock_client.api = MagicMock() mock_client.api.token = "syt_test_access_token" mock_client.api.session = MagicMock() mock_client.api.session.close = AsyncMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = None mock_client.crypto = None fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client) import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False): with patch.dict("sys.modules", fake_mautrix_mods): with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)): result = await adapter.connect() assert result is False @pytest.mark.asyncio async def test_connect_continues_when_e2ee_optional_but_no_deps(self): from plugins.platforms.matrix.adapter import MatrixAdapter config = PlatformConfig( enabled=True, token="syt_test_access_token", extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", "e2ee_mode": "optional", }, ) adapter = MatrixAdapter(config) fake_mautrix_mods = _make_fake_mautrix() mock_sync_store = MagicMock() mock_sync_store.get_next_batch = AsyncMock(return_value=None) mock_sync_store.put_next_batch = AsyncMock() mock_client = MagicMock() mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="DEV123")) mock_client.api = MagicMock() mock_client.api.token = "syt_test_access_token" mock_client.api.session = MagicMock() mock_client.api.session.close = AsyncMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = None mock_client.crypto = None mock_client.sync_store = mock_sync_store mock_client.sync = AsyncMock(return_value={"rooms": {"join": {}}, "next_batch": "s1"}) mock_client.get_account_data = AsyncMock(return_value=MagicMock(content={})) mock_client.add_dispatcher = MagicMock() mock_client.add_event_handler = MagicMock() mock_client.handle_sync = MagicMock(return_value=[]) fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client) import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False): with patch.dict("sys.modules", fake_mautrix_mods): with patch.object(matrix_mod, "_create_matrix_session", return_value=MagicMock()): with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)): result = await adapter.connect() assert result is True assert adapter._encryption is False await adapter.disconnect() class TestMatrixDeviceId: """MATRIX_DEVICE_ID should be used for stable device identity.""" def test_device_id_config_takes_precedence_over_env(self, monkeypatch): monkeypatch.setenv("MATRIX_DEVICE_ID", "FROM_ENV") from plugins.platforms.matrix.adapter import MatrixAdapter config = PlatformConfig( enabled=True, token="syt_test", extra={ "homeserver": "https://matrix.example.org", "device_id": "FROM_CONFIG", }, ) adapter = MatrixAdapter(config) assert adapter._device_id == "FROM_CONFIG" @pytest.mark.asyncio async def test_connect_keeps_configured_device_id_on_adapter(self): """MATRIX_DEVICE_ID stays on the adapter regardless of whoami. Note: this test previously asserted that the configured device_id overrides the whoami device_id outright. That is no longer true for the *client* identity — a token can only upload keys for its own device, so a conflicting whoami device now wins (see TestCryptoStoreResetOnDeviceChange). The configured value is still preferred when whoami reports no device, and is still recorded on the adapter, which is what this test pins. """ from plugins.platforms.matrix.adapter import MatrixAdapter config = PlatformConfig( enabled=True, token="syt_test_access_token", extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", "encryption": True, "device_id": "MY_STABLE_DEVICE", }, ) adapter = MatrixAdapter(config) fake_mautrix_mods = _make_fake_mautrix() mock_client = MagicMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = None mock_client.state_store = MagicMock() mock_client.sync_store = MagicMock() mock_client.crypto = None mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="WHOAMI_DEV")) mock_client.sync = AsyncMock(return_value={"rooms": {"join": {"!room:server": {}}}}) mock_client.add_event_handler = MagicMock() mock_client.handle_sync = MagicMock(return_value=[]) mock_client.query_keys = AsyncMock(return_value={ "device_keys": {"@bot:example.org": {"MY_STABLE_DEVICE": { "keys": {"ed25519:MY_STABLE_DEVICE": "fake_ed25519_key"}, }}}, }) mock_client.api = MagicMock() mock_client.api.token = "syt_test_access_token" mock_client.api.session = MagicMock() mock_client.api.session.close = AsyncMock() mock_olm = MagicMock() mock_olm.load = AsyncMock() mock_olm.share_keys = AsyncMock() mock_olm.share_keys_min_trust = None mock_olm.send_keys_min_trust = None mock_olm.account = MagicMock() mock_olm.account.identity_keys = {"ed25519": "fake_ed25519_key"} fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client) fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm) import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True): with patch.dict("sys.modules", fake_mautrix_mods): with patch.object(adapter, "_refresh_dm_cache", AsyncMock()): with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)): assert await adapter.connect() is True # The configured device_id is retained on the adapter. assert adapter._device_id == "MY_STABLE_DEVICE" # But the token's own device is what the client claims, because the # homeserver will not accept key uploads for any other device. assert mock_client.device_id == "WHOAMI_DEV" await adapter.disconnect() class TestMatrixPasswordLoginDeviceId: """MATRIX_DEVICE_ID should be passed to mautrix Client even with password login.""" @pytest.mark.asyncio async def test_password_login_uses_device_id(self): from plugins.platforms.matrix.adapter import MatrixAdapter config = PlatformConfig( enabled=True, extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", "password": "secret", "device_id": "STABLE_PW_DEVICE", }, ) adapter = MatrixAdapter(config) fake_mautrix_mods = _make_fake_mautrix() mock_client = MagicMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = None mock_client.state_store = MagicMock() mock_client.sync_store = MagicMock() mock_client.crypto = None mock_client.login = AsyncMock(return_value=MagicMock(device_id="STABLE_PW_DEVICE", access_token="tok")) mock_client.sync = AsyncMock(return_value={"rooms": {"join": {}}}) mock_client.add_event_handler = MagicMock() mock_client.api = MagicMock() mock_client.api.token = "" mock_client.api.session = MagicMock() mock_client.api.session.close = AsyncMock() fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client) with patch.dict("sys.modules", fake_mautrix_mods): with patch.object(adapter, "_refresh_dm_cache", AsyncMock()): with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)): assert await adapter.connect() is True mock_client.login.assert_awaited_once() assert adapter._device_id == "STABLE_PW_DEVICE" await adapter.disconnect() class TestMatrixDeviceIdConfig: """MATRIX_DEVICE_ID should be plumbed through gateway config.""" def test_device_id_in_config_extra(self, monkeypatch): monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_abc123") monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") monkeypatch.setenv("MATRIX_DEVICE_ID", "HERMES_BOT") from gateway.config import GatewayConfig, _apply_env_overrides config = GatewayConfig() _apply_env_overrides(config) mc = config.platforms[Platform.MATRIX] assert mc.extra.get("device_id") == "HERMES_BOT" class TestMatrixSyncLoop: @pytest.mark.asyncio async def test_dispatch_sync_accepts_async_handle_sync(self): """Some fake clients expose handle_sync as an async dispatcher.""" adapter = _make_adapter() called = False async def handle_sync(sync_data): nonlocal called called = sync_data["next_batch"] == "s1" return [] adapter._client = types.SimpleNamespace(handle_sync=handle_sync) await adapter._dispatch_sync({"next_batch": "s1"}) assert called is True @pytest.mark.asyncio async def test_sync_loop_dispatches_registered_room_message_handler(self): """Inbound sync data should flow through handle_sync into message handling.""" adapter = _make_adapter() adapter._closing = False adapter._user_id = "@bot:example.org" adapter._startup_ts = time.time() - 10 adapter._dm_rooms = {"!dm:example.org": True} adapter._text_batch_delay_seconds = 0 adapter._background_read_receipt = MagicMock() captured = [] async def capture(event): captured.append(event) adapter.handle_message = capture event = types.SimpleNamespace( sender="@alice:example.org", event_id="$dm1", room_id="!dm:example.org", timestamp=int(time.time() * 1000), content={"msgtype": "m.text", "body": "hello"}, ) async def _sync_once(**kwargs): adapter._closing = True return {"rooms": {"join": {"!dm:example.org": {}}}, "next_batch": "s1234"} mock_sync_store = MagicMock() mock_sync_store.get_next_batch = AsyncMock(return_value=None) mock_sync_store.put_next_batch = AsyncMock() fake_client = MagicMock() fake_client.sync = AsyncMock(side_effect=_sync_once) fake_client.sync_store = mock_sync_store fake_client.get_state_event = AsyncMock(side_effect=Exception("no state")) fake_client.state_store = MagicMock() fake_client.state_store.get_members = AsyncMock(return_value=["@bot:example.org", "@alice:example.org"]) fake_client.state_store.get_member = AsyncMock(return_value=None) def handle_sync(sync_data): return [asyncio.create_task(adapter._on_room_message(event))] fake_client.handle_sync = MagicMock(side_effect=handle_sync) adapter._client = fake_client await adapter._sync_loop() assert len(captured) == 1 assert captured[0].text == "hello" assert captured[0].source.chat_type == "dm" @pytest.mark.asyncio async def test_connect_receives_dm_from_initial_sync_dispatch(self): """A DM delivered by initial sync should reach the message handler after connect.""" from plugins.platforms.matrix.adapter import MatrixAdapter adapter = MatrixAdapter( PlatformConfig( enabled=True, token="syt_test_access_token", extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", "encryption": False, }, ) ) adapter._text_batch_delay_seconds = 0 adapter._background_read_receipt = MagicMock() captured = [] async def capture(event): captured.append(event) adapter.handle_message = capture fake_mautrix_mods = _make_fake_mautrix() mock_sync_store = MagicMock() mock_sync_store.get_next_batch = AsyncMock(return_value=None) mock_sync_store.put_next_batch = AsyncMock() mock_client = MagicMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = None mock_client.crypto = None mock_client.sync_store = mock_sync_store mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="DEV123")) mock_client.sync = AsyncMock(return_value={ "rooms": {"join": {"!dm:example.org": {}}}, "next_batch": "s1", }) mock_client.get_account_data = AsyncMock( return_value=MagicMock(content={"@alice:example.org": ["!dm:example.org"]}) ) mock_client.get_state_event = AsyncMock(side_effect=Exception("no state")) mock_client.state_store = MagicMock() mock_client.state_store.get_members = AsyncMock(return_value=["@bot:example.org", "@alice:example.org"]) mock_client.state_store.get_member = AsyncMock(return_value=None) mock_client.add_event_handler = MagicMock() mock_client.add_dispatcher = MagicMock() mock_client.api = MagicMock() mock_client.api.token = "syt_test_access_token" mock_client.api.session = MagicMock() mock_client.api.session.close = AsyncMock() event = types.SimpleNamespace( sender="@alice:example.org", event_id="$initial-dm", room_id="!dm:example.org", timestamp=int(time.time() * 1000), content={"msgtype": "m.text", "body": "hello after connect"}, ) def handle_sync(sync_data): return [asyncio.create_task(adapter._on_room_message(event))] mock_client.handle_sync = MagicMock(side_effect=handle_sync) fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client) import plugins.platforms.matrix.adapter as matrix_mod with patch.dict("sys.modules", fake_mautrix_mods): with patch.object(matrix_mod, "_create_matrix_session", return_value=MagicMock()): with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)): assert await adapter.connect() is True assert len(captured) == 1 assert captured[0].text == "hello after connect" assert captured[0].source.chat_type == "dm" await adapter.disconnect() class TestMatrixUploadAndSend: @pytest.mark.asyncio async def test_upload_encrypted_room_uses_file_payload(self): """Encrypted rooms should use 'file' key with crypto metadata.""" adapter = _make_adapter() adapter._encryption = True mock_client = MagicMock() mock_client.crypto = object() mock_client.state_store = MagicMock() mock_client.state_store.is_encrypted = AsyncMock(return_value=True) mock_client.upload_media = AsyncMock(return_value="mxc://example.org/enc") mock_client.send_message_event = AsyncMock(return_value="$event") adapter._client = mock_client with patch.dict("sys.modules", _make_fake_mautrix()): result = await adapter._upload_and_send( "!room:example.org", b"secret", "secret.txt", "text/plain", "m.file", ) assert result.success is True # Should have uploaded ciphertext, not plaintext uploaded_data = mock_client.upload_media.await_args.args[0] assert uploaded_data != b"secret" sent = mock_client.send_message_event.await_args.args[2] assert "url" not in sent assert "file" in sent assert sent["file"]["url"] == "mxc://example.org/enc" @pytest.mark.asyncio async def test_media_preserves_caption_and_thread(self): adapter = _make_adapter() mock_client = MagicMock() mock_client.upload_media = AsyncMock(return_value="mxc://example.org/plain") mock_client.send_message_event = AsyncMock(return_value="$event") adapter._client = mock_client result = await adapter._upload_and_send( "!room:example.org", b"image", "chart.png", "image/png", "m.image", caption="Chart caption", metadata={"thread_id": "$root"}, ) assert result.success is True sent = mock_client.send_message_event.await_args.args[2] assert sent["body"] == "Chart caption" assert sent["m.relates_to"]["rel_type"] == "m.thread" assert sent["m.relates_to"]["event_id"] == "$root" assert sent["m.relates_to"]["m.in_reply_to"] == {"event_id": "$root"} class TestMatrixDiagnostics: def test_diagnostics_redacts_credentials_and_reports_status(self, monkeypatch): import plugins.platforms.matrix.adapter as matrix_mod monkeypatch.setenv("MATRIX_RECOVERY_KEY", "secret recovery key") adapter = _make_adapter() adapter._access_token = "syt_super_secret" adapter._password = "password" adapter._user_id = "@bot:example.org" adapter._device_id = "DEV123" adapter._joined_rooms = {"!one:example.org", "!two:example.org"} adapter._last_sync_ts = time.time() - 7 adapter._max_media_bytes = 123 adapter._client = MagicMock() with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True): diagnostics = adapter.get_diagnostics() assert diagnostics["auth"]["token_preview"] == "***" assert "syt_super_secret" not in str(diagnostics) assert "DEV123" not in str(diagnostics) assert diagnostics["auth"]["device_id_present"] is True assert diagnostics["auth"]["device_id_preview"] == "***" assert diagnostics["sync"]["connected"] is True assert diagnostics["sync"]["joined_room_count"] == 2 assert diagnostics["sync"]["last_sync_age_seconds"] >= 0 assert diagnostics["e2ee"]["recovery_key_configured"] is True assert diagnostics["media"]["max_media_bytes"] == 123 @pytest.mark.asyncio async def test_matrix_recovery_key_bootstrap_skips_existing_output_file( self, tmp_path, monkeypatch, caplog, ): from plugins.platforms.matrix.adapter import MatrixAdapter output_path = tmp_path / "matrix-recovery-key.txt" output_path.write_text("existing\n") monkeypatch.delenv("MATRIX_RECOVERY_KEY", raising=False) monkeypatch.setenv("MATRIX_RECOVERY_KEY_OUTPUT_FILE", str(output_path)) config = PlatformConfig( enabled=True, token="syt_test_token", extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", "encryption": True, }, ) adapter = MatrixAdapter(config) fake_mautrix_mods = _make_fake_mautrix() mock_client = MagicMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = None mock_client.state_store = MagicMock() mock_client.sync_store = MagicMock() mock_client.crypto = None mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="DEV123")) mock_client.sync = AsyncMock(return_value={"rooms": {"join": {}}}) mock_client.add_event_handler = MagicMock() mock_client.add_dispatcher = MagicMock() mock_client.handle_sync = MagicMock(return_value=[]) mock_client.query_keys = AsyncMock(return_value={ "device_keys": {"@bot:example.org": {"DEV123": { "keys": {"ed25519:DEV123": "fake_ed25519_key"}, }}}, }) mock_client.api = MagicMock() mock_client.api.token = "syt_test_token" mock_client.api.session = MagicMock() mock_client.api.session.close = AsyncMock() mock_olm = MagicMock() mock_olm.load = AsyncMock() mock_olm.share_keys = AsyncMock() mock_olm.get_own_cross_signing_public_keys = AsyncMock(return_value=None) mock_olm.generate_recovery_key = AsyncMock(return_value="super-secret-key") mock_olm.share_keys_min_trust = None mock_olm.send_keys_min_trust = None mock_olm.account = MagicMock() mock_olm.account.identity_keys = {"ed25519": "fake_ed25519_key"} fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client) fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm) import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True): with patch.dict("sys.modules", fake_mautrix_mods): with patch.object(adapter, "_refresh_dm_cache", AsyncMock()): with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)): assert await adapter.connect() is True mock_olm.generate_recovery_key.assert_not_called() assert "already exists" in caplog.text assert "super-secret-key" not in caplog.text assert output_path.read_text() == "existing\n" await adapter.disconnect() def test_matrix_diagnostics_redacts_recovery_key(self, monkeypatch): monkeypatch.setenv("MATRIX_RECOVERY_KEY", "diagnostic-secret-recovery-key") adapter = _make_adapter() diagnostics = adapter.get_diagnostics() assert diagnostics["e2ee"]["recovery_key_configured"] is True assert "diagnostic-secret-recovery-key" not in str(diagnostics) class TestMatrixEncryptedSendFallback: @pytest.mark.asyncio async def test_send_retries_after_e2ee_error(self): """send() should retry with crypto.share_keys() on E2EE errors.""" adapter = _make_adapter() adapter._encryption = True fake_client = MagicMock() fake_client.send_message_event = AsyncMock(side_effect=[ Exception("encryption error"), "$event123", # mautrix returns EventID string directly ]) mock_crypto = MagicMock() mock_crypto.share_keys = AsyncMock() fake_client.crypto = mock_crypto adapter._client = fake_client result = await adapter.send("!room:example.org", "hello") assert result.success is True assert result.message_id == "$event123" mock_crypto.share_keys.assert_awaited_once() assert fake_client.send_message_event.await_count == 2 # --------------------------------------------------------------------------- # E2EE: _joined_rooms reference preservation for CryptoStateStore # --------------------------------------------------------------------------- class TestJoinedRoomsReference: def test_joined_rooms_reference_preserved_after_reassignment(self): """_CryptoStateStore must see updates after initial sync populates rooms.""" from plugins.platforms.matrix.adapter import _CryptoStateStore joined = set() store = _CryptoStateStore(MagicMock(), joined) # Simulate what connect() should do: mutate in place, not reassign. joined.clear() joined.update(["!room1:example.org", "!room2:example.org"]) import asyncio rooms = asyncio.get_event_loop().run_until_complete(store.find_shared_rooms("@user:ex")) assert set(rooms) == {"!room1:example.org", "!room2:example.org"} # --------------------------------------------------------------------------- # E2EE: connect registers encrypted event handler # --------------------------------------------------------------------------- class TestMatrixEncryptedEventHandler: @pytest.mark.asyncio async def test_connect_registers_encrypted_event_handler_when_encryption_on(self): from plugins.platforms.matrix.adapter import MatrixAdapter config = PlatformConfig( enabled=True, token="syt_test_token", extra={ "homeserver": "https://matrix.example.org", "user_id": "@bot:example.org", "encryption": True, }, ) adapter = MatrixAdapter(config) fake_mautrix_mods = _make_fake_mautrix() mock_client = MagicMock() mock_client.mxid = "@bot:example.org" mock_client.device_id = None mock_client.state_store = MagicMock() mock_client.sync_store = MagicMock() mock_client.crypto = None # Will be set during connect mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="DEV123")) mock_client.sync = AsyncMock(return_value={"rooms": {"join": {"!room:server": {}}}}) mock_client.add_event_handler = MagicMock() mock_client.handle_sync = MagicMock(return_value=[]) mock_client.query_keys = AsyncMock(return_value={ "device_keys": {"@bot:example.org": {"DEV123": { "keys": {"ed25519:DEV123": "fake_ed25519_key"}, }}}, }) mock_client.api = MagicMock() mock_client.api.token = "syt_test_token" mock_client.api.session = MagicMock() mock_client.api.session.close = AsyncMock() mock_olm = MagicMock() mock_olm.load = AsyncMock() mock_olm.share_keys = AsyncMock() mock_olm.share_keys_min_trust = None mock_olm.send_keys_min_trust = None mock_olm.account = MagicMock() mock_olm.account.identity_keys = {"ed25519": "fake_ed25519_key"} fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client) fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm) import plugins.platforms.matrix.adapter as matrix_mod with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True): with patch.dict("sys.modules", fake_mautrix_mods): with patch.object(adapter, "_refresh_dm_cache", AsyncMock()): with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)): assert await adapter.connect() is True # Verify inbound event handlers were registered as sync-awaited # callbacks. mautrix only returns waited handler tasks from # handle_sync(), so background-only handlers leave _dispatch_sync() # without a completion point for Hermes' Matrix intake. handler_calls = mock_client.add_event_handler.call_args_list waited_types = { str(call.args[0]) for call in handler_calls if call.kwargs.get("wait_sync") is True } assert "m.room.message" in waited_types assert "m.reaction" in waited_types assert "internal.invite" in waited_types await adapter.disconnect() # --------------------------------------------------------------------------- # Disconnect # --------------------------------------------------------------------------- class TestMatrixDisconnect: @pytest.mark.asyncio async def test_disconnect_closes_api_session(self): """disconnect() should close client.api.session.""" adapter = _make_adapter() adapter._sync_task = None mock_session = MagicMock() mock_session.close = AsyncMock() mock_api = MagicMock() mock_api.session = mock_session fake_client = MagicMock() fake_client.api = mock_api adapter._client = fake_client await adapter.disconnect() mock_session.close.assert_awaited_once() assert adapter._client is None # --------------------------------------------------------------------------- # Markdown to HTML: security tests # --------------------------------------------------------------------------- class TestMatrixMarkdownHtmlSecurity: """Tests for HTML injection prevention in _markdown_to_html_fallback.""" def setup_method(self): from plugins.platforms.matrix.adapter import MatrixAdapter self.convert = MatrixAdapter._markdown_to_html_fallback def test_script_injection_in_header(self): result = self.convert("# ") assert "") assert "*") assert "\n```') assert "<script>" in result assert "
ItemApples