From cc5cfe68477ce793b8f121cfe3de000b67530d8e Mon Sep 17 00:00:00 2001 From: Alexander Minges Date: Tue, 10 Feb 2026 15:30:54 +0100 Subject: [PATCH] test(matrix): cover mention policy and sender filtering --- tests/test_matrix_channel.py | 145 ++++++++++++++++++++++++++++++++++- 1 file changed, 142 insertions(+), 3 deletions(-) diff --git a/tests/test_matrix_channel.py b/tests/test_matrix_channel.py index e3bea9e..cc834c3 100644 --- a/tests/test_matrix_channel.py +++ b/tests/test_matrix_channel.py @@ -197,7 +197,7 @@ async def test_on_message_sets_typing_for_allowed_sender() -> None: channel._handle_message = _fake_handle_message # type: ignore[method-assign] room = SimpleNamespace(room_id="!room:matrix.org", display_name="Test room") - event = SimpleNamespace(sender="@alice:matrix.org", body="Hello") + event = SimpleNamespace(sender="@alice:matrix.org", body="Hello", source={}) await channel._on_message(room, event) @@ -214,7 +214,7 @@ async def test_on_message_skips_typing_for_self_message() -> None: channel.client = client room = SimpleNamespace(room_id="!room:matrix.org", display_name="Test room") - event = SimpleNamespace(sender="@bot:matrix.org", body="Hello") + event = SimpleNamespace(sender="@bot:matrix.org", body="Hello", source={}) await channel._on_message(room, event) @@ -227,14 +227,153 @@ async def test_on_message_skips_typing_for_denied_sender() -> None: client = _FakeAsyncClient("", "", "", None) channel.client = client + handled: list[str] = [] + + async def _fake_handle_message(**kwargs) -> None: + handled.append(kwargs["sender_id"]) + + channel._handle_message = _fake_handle_message # type: ignore[method-assign] + room = SimpleNamespace(room_id="!room:matrix.org", display_name="Test room") - event = SimpleNamespace(sender="@alice:matrix.org", body="Hello") + event = SimpleNamespace(sender="@alice:matrix.org", body="Hello", source={}) await channel._on_message(room, event) + assert handled == [] assert client.typing_calls == [] +@pytest.mark.asyncio +async def test_on_message_mention_policy_requires_mx_mentions() -> None: + channel = MatrixChannel(_make_config(group_policy="mention"), MessageBus()) + client = _FakeAsyncClient("", "", "", None) + channel.client = client + + handled: list[str] = [] + + async def _fake_handle_message(**kwargs) -> None: + handled.append(kwargs["sender_id"]) + + channel._handle_message = _fake_handle_message # type: ignore[method-assign] + + room = SimpleNamespace(room_id="!room:matrix.org", display_name="Test room", member_count=3) + event = SimpleNamespace(sender="@alice:matrix.org", body="Hello", source={"content": {}}) + + await channel._on_message(room, event) + + assert handled == [] + assert client.typing_calls == [] + + +@pytest.mark.asyncio +async def test_on_message_mention_policy_accepts_bot_user_mentions() -> None: + channel = MatrixChannel(_make_config(group_policy="mention"), MessageBus()) + client = _FakeAsyncClient("", "", "", None) + channel.client = client + + handled: list[str] = [] + + async def _fake_handle_message(**kwargs) -> None: + handled.append(kwargs["sender_id"]) + + channel._handle_message = _fake_handle_message # type: ignore[method-assign] + + room = SimpleNamespace(room_id="!room:matrix.org", display_name="Test room", member_count=3) + event = SimpleNamespace( + sender="@alice:matrix.org", + body="Hello", + source={"content": {"m.mentions": {"user_ids": ["@bot:matrix.org"]}}}, + ) + + await channel._on_message(room, event) + + assert handled == ["@alice:matrix.org"] + assert client.typing_calls == [("!room:matrix.org", True, TYPING_NOTICE_TIMEOUT_MS)] + + +@pytest.mark.asyncio +async def test_on_message_mention_policy_allows_direct_room_without_mentions() -> None: + channel = MatrixChannel(_make_config(group_policy="mention"), MessageBus()) + client = _FakeAsyncClient("", "", "", None) + channel.client = client + + handled: list[str] = [] + + async def _fake_handle_message(**kwargs) -> None: + handled.append(kwargs["sender_id"]) + + channel._handle_message = _fake_handle_message # type: ignore[method-assign] + + room = SimpleNamespace(room_id="!dm:matrix.org", display_name="DM", member_count=2) + event = SimpleNamespace(sender="@alice:matrix.org", body="Hello", source={"content": {}}) + + await channel._on_message(room, event) + + assert handled == ["@alice:matrix.org"] + assert client.typing_calls == [("!dm:matrix.org", True, TYPING_NOTICE_TIMEOUT_MS)] + + +@pytest.mark.asyncio +async def test_on_message_allowlist_policy_requires_room_id() -> None: + channel = MatrixChannel( + _make_config(group_policy="allowlist", group_allow_from=["!allowed:matrix.org"]), + MessageBus(), + ) + client = _FakeAsyncClient("", "", "", None) + channel.client = client + + handled: list[str] = [] + + async def _fake_handle_message(**kwargs) -> None: + handled.append(kwargs["chat_id"]) + + channel._handle_message = _fake_handle_message # type: ignore[method-assign] + + denied_room = SimpleNamespace(room_id="!denied:matrix.org", display_name="Denied", member_count=3) + event = SimpleNamespace(sender="@alice:matrix.org", body="Hello", source={"content": {}}) + await channel._on_message(denied_room, event) + + allowed_room = SimpleNamespace( + room_id="!allowed:matrix.org", + display_name="Allowed", + member_count=3, + ) + await channel._on_message(allowed_room, event) + + assert handled == ["!allowed:matrix.org"] + assert client.typing_calls == [("!allowed:matrix.org", True, TYPING_NOTICE_TIMEOUT_MS)] + + +@pytest.mark.asyncio +async def test_on_message_room_mention_requires_opt_in() -> None: + channel = MatrixChannel(_make_config(group_policy="mention"), MessageBus()) + client = _FakeAsyncClient("", "", "", None) + channel.client = client + + handled: list[str] = [] + + async def _fake_handle_message(**kwargs) -> None: + handled.append(kwargs["sender_id"]) + + channel._handle_message = _fake_handle_message # type: ignore[method-assign] + + room = SimpleNamespace(room_id="!room:matrix.org", display_name="Test room", member_count=3) + room_mention_event = SimpleNamespace( + sender="@alice:matrix.org", + body="Hello everyone", + source={"content": {"m.mentions": {"room": True}}}, + ) + + await channel._on_message(room, room_mention_event) + assert handled == [] + assert client.typing_calls == [] + + channel.config.allow_room_mentions = True + await channel._on_message(room, room_mention_event) + assert handled == ["@alice:matrix.org"] + assert client.typing_calls == [("!room:matrix.org", True, TYPING_NOTICE_TIMEOUT_MS)] + + @pytest.mark.asyncio async def test_send_clears_typing_after_send() -> None: channel = MatrixChannel(_make_config(), MessageBus())