hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

test(matrix): fold plaintext-room case into the encrypted-drop warn test

Browse Source 
Keep two invariant tests: the warn-once test now also carries a plaintext
room in the same sync (must not warn) and asserts the MATRIX_E2EE_MODE
hint; the crypto-attached test stays. Drops the standalone plaintext test.
This commit is contained in:
kshitijk4poor
2026-10-05 13:45:19 +05:30
committed by kshitij
parent f30d3c870e
commit 4e2aad9baa
1 changed files with 8 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
tests/gateway/test_matrix.py
View File

@@ -1191,17 +1191,21 @@ class TestMatrixSyncLoop:
adapter._client = fake_client
sync_data = {
"rooms": {"join": {"!enc:example.org": {
"timeline": {"events": [{"type": "m.room.encrypted", "event_id": "$e1"}]}}}},
"rooms": {"join": {
"!enc:example.org": {"timeline": {"events": [
{"type": "m.room.encrypted", "event_id": "$e1"}]}},
"!plain:example.org": {"timeline": {"events": [
{"type": "m.room.message", "event_id": "$m1"}]}},
}},
"next_batch": "s1",
}
with caplog.at_level(logging.WARNING, logger="plugins.platforms.matrix.adapter"):
await adapter._absorb_sync(fake_client, sync_data)
await adapter._absorb_sync(fake_client, sync_data) # rate-limited: one warning per room
warnings = [r for r in caplog.records
if "!enc:example.org" in r.getMessage() and "encrypted" in r.getMessage()]
warnings = [r.getMessage() for r in caplog.records if "encrypted" in r.getMessage()]
assert len(warnings) == 1
assert "!enc:example.org" in warnings[0] and "MATRIX_E2EE_MODE" in warnings[0]
@pytest.mark.asyncio
async def test_absorb_sync_no_encrypted_warning_when_crypto_attached(self, caplog):
@@ -1223,26 +1227,6 @@ class TestMatrixSyncLoop:
assert not [r for r in caplog.records if "encrypted" in r.getMessage()]
@pytest.mark.asyncio
async def test_absorb_sync_no_warning_for_plaintext_events_without_crypto(
self, caplog
):
"""Plain rooms must not trip the encrypted-drop warning."""
adapter = _make_adapter()
adapter._closing = False
fake_client = self._cryptoless_client()
adapter._client = fake_client
sync_data = {
"rooms": {"join": {"!plain:example.org": {
"timeline": {"events": [{"type": "m.room.message", "event_id": "$m1"}]}}}},
"next_batch": "s1",
}
with caplog.at_level(logging.WARNING, logger="plugins.platforms.matrix.adapter"):
await adapter._absorb_sync(fake_client, sync_data)
assert not [r for r in caplog.records if "encrypted" in r.getMessage()]
@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."""