Timmy_Foundation/hermes-agent
15
0
Fork 0
Code Issues 247 Pull Requests 14 Actions 1 Packages Projects Releases Wiki Activity

test(matrix): update all test mocks for mautrix-python API

Browse Source 
Rewrite mock infrastructure across three test files:
- test_matrix.py: replace fake nio module with fake mautrix module tree,
  update all client method mocks to new API names and return types
- test_matrix_voice.py: update event construction, download/upload mocks,
  handler invocation (single event arg, no room object)
- test_matrix_mention.py: update mock module, event construction, DM
  detection via _dm_rooms cache instead of room.member_count

157 tests passing.
This commit is contained in:
alt-glitch
2026-04-11 06:58:32 +05:30
committed by Teknium
parent 8053d48c8d
commit 417e28f941
3 changed files with 676 additions and 1156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1435
tests/gateway/test_matrix.py
View File

File diff suppressed because it is too large Load Diff

159
tests/gateway/test_matrix_mention.py
View File

@@ -11,24 +11,59 @@ import pytest
from gateway.config import PlatformConfig
def _ensure_nio_mock():
"""Install a mock nio module when matrix-nio isn't available."""
if "nio" in sys.modules and hasattr(sys.modules["nio"], "__file__"):
def _ensure_mautrix_mock():
"""Install mock mautrix modules when mautrix-python isn't available."""
if "mautrix" in sys.modules and hasattr(sys.modules["mautrix"], "__file__"):
return
nio_mod = MagicMock()
nio_mod.MegolmEvent = type("MegolmEvent", (), {})
nio_mod.RoomMessageText = type("RoomMessageText", (), {})
nio_mod.RoomMessageImage = type("RoomMessageImage", (), {})
nio_mod.RoomMessageAudio = type("RoomMessageAudio", (), {})
nio_mod.RoomMessageVideo = type("RoomMessageVideo", (), {})
nio_mod.RoomMessageFile = type("RoomMessageFile", (), {})
nio_mod.DownloadResponse = type("DownloadResponse", (), {})
nio_mod.MemoryDownloadResponse = type("MemoryDownloadResponse", (), {})
nio_mod.InviteMemberEvent = type("InviteMemberEvent", (), {})
sys.modules.setdefault("nio", nio_mod)
# Root module
mautrix_mod = MagicMock()
# mautrix.types — commonly imported types
types_mod = MagicMock()
types_mod.EventType = MagicMock()
types_mod.RoomID = str
types_mod.UserID = str
types_mod.EventID = str
types_mod.ContentURI = str
types_mod.RoomCreatePreset = MagicMock()
types_mod.PresenceState = MagicMock()
types_mod.PaginationDirection = MagicMock()
types_mod.SyncToken = str
types_mod.TrustState = MagicMock()
# mautrix.client
client_mod = MagicMock()
client_mod.Client = MagicMock()
client_mod.InternalEventType = MagicMock()
# mautrix.client.state_store
state_store_mod = MagicMock()
state_store_mod.MemoryStateStore = MagicMock()
state_store_mod.MemorySyncStore = MagicMock()
# mautrix.api
api_mod = MagicMock()
api_mod.HTTPAPI = MagicMock()
# mautrix.crypto
crypto_mod = MagicMock()
crypto_mod.OlmMachine = MagicMock()
crypto_store_mod = MagicMock()
crypto_store_mod.MemoryCryptoStore = MagicMock()
crypto_attachments_mod = MagicMock()
sys.modules.setdefault("mautrix", mautrix_mod)
sys.modules.setdefault("mautrix.types", types_mod)
sys.modules.setdefault("mautrix.client", client_mod)
sys.modules.setdefault("mautrix.client.state_store", state_store_mod)
sys.modules.setdefault("mautrix.api", api_mod)
sys.modules.setdefault("mautrix.crypto", crypto_mod)
sys.modules.setdefault("mautrix.crypto.store", crypto_store_mod)
sys.modules.setdefault("mautrix.crypto.attachments", crypto_attachments_mod)
_ensure_nio_mock()
_ensure_mautrix_mock()
def _make_adapter(tmp_path=None):
@@ -50,24 +85,25 @@ def _make_adapter(tmp_path=None):
return adapter
def _make_room(room_id="!room1:example.org", member_count=5, is_dm=False):
"""Create a fake Matrix room."""
room = SimpleNamespace(
room_id=room_id,
member_count=member_count,
users={},
)
return room
def _set_dm(adapter, room_id="!room1:example.org", is_dm=True):
"""Mark a room as DM (or not) in the adapter's cache."""
adapter._dm_rooms[room_id] = is_dm
def _make_event(
body,
sender="@alice:example.org",
event_id="$evt1",
room_id="!room1:example.org",
formatted_body=None,
thread_id=None,
):
"""Create a fake RoomMessageText event."""
"""Create a fake room message event.
The mautrix adapter reads ``event.room_id``, ``event.sender``,
``event.event_id``, ``event.timestamp``, and ``event.content``
(a dict with ``msgtype``, ``body``, etc.).
"""
content = {"body": body, "msgtype": "m.text"}
if formatted_body:
content["formatted_body"] = formatted_body
@@ -83,9 +119,9 @@ def _make_event(
return SimpleNamespace(
sender=sender,
event_id=event_id,
server_timestamp=int(time.time() * 1000),
body=body,
source={"content": content},
room_id=room_id,
timestamp=int(time.time() * 1000),
content=content,
)
@@ -152,10 +188,9 @@ async def test_require_mention_default_ignores_unmentioned(monkeypatch):
monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
adapter = _make_adapter()
room = _make_room()
event = _make_event("hello everyone")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_not_awaited()
@@ -167,10 +202,9 @@ async def test_require_mention_default_processes_mentioned(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room()
event = _make_event("@hermes:example.org help me")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.text == "help me"
@@ -184,11 +218,10 @@ async def test_require_mention_html_pill(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room()
formatted = '<a href="https://matrix.to/#/@hermes:example.org">Hermes</a> help'
event = _make_event("Hermes help", formatted_body=formatted)
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
@@ -200,11 +233,11 @@ async def test_require_mention_dm_always_responds(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
# member_count=2 triggers DM detection
room = _make_room(member_count=2)
# Mark the room as a DM via the adapter's cache.
_set_dm(adapter)
event = _make_event("hello without mention")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
@@ -216,10 +249,10 @@ async def test_dm_strips_mention(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room(member_count=2)
_set_dm(adapter)
event = _make_event("@hermes:example.org help me")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.text == "help me"
@@ -233,10 +266,9 @@ async def test_bare_mention_passes_empty_string(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room()
event = _make_event("@hermes:example.org")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.text == ""
@@ -250,10 +282,9 @@ async def test_require_mention_free_response_room(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room(room_id="!room1:example.org")
event = _make_event("hello without mention")
event = _make_event("hello without mention", room_id="!room1:example.org")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
@@ -267,10 +298,9 @@ async def test_require_mention_bot_participated_thread(monkeypatch):
adapter = _make_adapter()
adapter._bot_participated_threads.add("$thread1")
room = _make_room()
event = _make_event("hello without mention", thread_id="$thread1")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
@@ -282,10 +312,9 @@ async def test_require_mention_disabled(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room()
event = _make_event("hello without mention")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.text == "hello without mention"
@@ -303,10 +332,9 @@ async def test_auto_thread_default_creates_thread(monkeypatch):
monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
adapter = _make_adapter()
room = _make_room()
event = _make_event("hello", event_id="$msg1")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.source.thread_id == "$msg1"
@@ -320,10 +348,9 @@ async def test_auto_thread_preserves_existing_thread(monkeypatch):
adapter = _make_adapter()
adapter._bot_participated_threads.add("$thread_root")
room = _make_room()
event = _make_event("reply in thread", thread_id="$thread_root")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.source.thread_id == "$thread_root"
@@ -336,10 +363,10 @@ async def test_auto_thread_skips_dm(monkeypatch):
monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
adapter = _make_adapter()
room = _make_room(member_count=2)
_set_dm(adapter)
event = _make_event("hello dm", event_id="$dm1")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.source.thread_id is None
@@ -352,10 +379,9 @@ async def test_auto_thread_disabled(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room()
event = _make_event("hello", event_id="$msg1")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.source.thread_id is None
@@ -368,11 +394,10 @@ async def test_auto_thread_tracks_participation(monkeypatch):
monkeypatch.delenv("MATRIX_AUTO_THREAD", raising=False)
adapter = _make_adapter()
room = _make_room()
event = _make_event("hello", event_id="$msg1")
with patch.object(adapter, "_save_participated_threads"):
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
assert "$msg1" in adapter._bot_participated_threads
@@ -448,10 +473,10 @@ async def test_dm_mention_thread_disabled_by_default(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room(member_count=2)
_set_dm(adapter)
event = _make_event("@hermes:example.org help me", event_id="$dm1")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.source.thread_id is None
@@ -464,11 +489,11 @@ async def test_dm_mention_thread_creates_thread(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room(member_count=2)
_set_dm(adapter)
event = _make_event("@hermes:example.org help me", event_id="$dm1")
with patch.object(adapter, "_save_participated_threads"):
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
@@ -483,10 +508,10 @@ async def test_dm_mention_thread_no_mention_no_thread(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room(member_count=2)
_set_dm(adapter)
event = _make_event("hello without mention", event_id="$dm1")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.source.thread_id is None
@@ -499,11 +524,11 @@ async def test_dm_mention_thread_preserves_existing_thread(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
_set_dm(adapter)
adapter._bot_participated_threads.add("$existing_thread")
room = _make_room(member_count=2)
event = _make_event("@hermes:example.org help me", thread_id="$existing_thread")
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
adapter.handle_message.assert_awaited_once()
msg = adapter.handle_message.await_args.args[0]
assert msg.source.thread_id == "$existing_thread"
@@ -516,11 +541,11 @@ async def test_dm_mention_thread_tracks_participation(monkeypatch):
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
room = _make_room(member_count=2)
_set_dm(adapter)
event = _make_event("@hermes:example.org help", event_id="$dm1")
with patch.object(adapter, "_save_participated_threads"):
await adapter._on_room_message(room, event)
await adapter._on_room_message(event)
assert "$dm1" in adapter._bot_participated_threads

238
tests/gateway/test_matrix_voice.py
View File

@@ -1,18 +1,23 @@
"""Tests for Matrix voice message support (MSC3245)."""
"""Tests for Matrix voice message support (MSC3245).
Updated for the mautrix-python SDK (no more matrix-nio / nio imports).
"""
import io
import os
import tempfile
import types
from types import SimpleNamespace
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
# Try importing real nio; skip entire file if not available.
# A MagicMock in sys.modules (from another test) is not the real package.
# Try importing mautrix; skip entire file if not available.
try:
import nio as _nio_probe
if not isinstance(_nio_probe, types.ModuleType) or not hasattr(_nio_probe, "__file__"):
pytest.skip("nio in sys.modules is a mock, not the real package", allow_module_level=True)
import mautrix as _mautrix_probe
if not isinstance(_mautrix_probe, types.ModuleType) or not hasattr(_mautrix_probe, "__file__"):
pytest.skip("mautrix in sys.modules is a mock, not the real package", allow_module_level=True)
except ImportError:
pytest.skip("matrix-nio not installed", allow_module_level=True)
pytest.skip("mautrix not installed", allow_module_level=True)
from gateway.platforms.base import MessageType
@@ -25,7 +30,7 @@ def _make_adapter():
"""Create a MatrixAdapter with mocked config."""
from gateway.platforms.matrix import MatrixAdapter
from gateway.config import PlatformConfig
config = PlatformConfig(
enabled=True,
token="***",
@@ -38,32 +43,26 @@ def _make_adapter():
return adapter
def _make_room(room_id: str = "!test:example.org", member_count: int = 2):
"""Create a mock Matrix room."""
room = MagicMock()
room.room_id = room_id
room.member_count = member_count
return room
def _make_audio_event(
event_id: str = "$audio_event",
sender: str = "@alice:example.org",
room_id: str = "!test:example.org",
body: str = "Voice message",
url: str = "mxc://example.org/abc123",
is_voice: bool = False,
mimetype: str = "audio/ogg",
timestamp: float = 9999999999000, # ms
timestamp: int = 9999999999000, # ms
):
"""
Create a mock RoomMessageAudio event that passes isinstance checks.
Create a mock mautrix room message event.
In mautrix, the handler receives a single event object with attributes
``room_id``, ``sender``, ``event_id``, ``timestamp``, and ``content``
(a dict-like or serializable object).
Args:
is_voice: If True, adds org.matrix.msc3245.voice field to content
is_voice: If True, adds org.matrix.msc3245.voice field to content.
"""
import nio
# Build the source dict that nio events expose via .source
content = {
"msgtype": "m.audio",
"body": body,
@@ -72,39 +71,35 @@ def _make_audio_event(
"mimetype": mimetype,
},
}
if is_voice:
content["org.matrix.msc3245.voice"] = {}
# Create a real nio RoomMessageAudio-like object
# We use MagicMock but configure __class__ to pass isinstance check
event = MagicMock(spec=nio.RoomMessageAudio)
event.event_id = event_id
event.sender = sender
event.body = body
event.url = url
event.server_timestamp = timestamp
event.source = {
"type": "m.room.message",
"content": content,
}
# For MIME type extraction - needs to be a dict
event.content = content
event = SimpleNamespace(
event_id=event_id,
sender=sender,
room_id=room_id,
timestamp=timestamp,
content=content,
)
return event
def _make_download_response(body: bytes = b"fake audio data"):
"""Create a mock nio.MemoryDownloadResponse."""
import nio
resp = MagicMock()
resp.body = body
resp.__class__ = nio.MemoryDownloadResponse
return resp
def _make_state_store(member_count: int = 2):
"""Create a mock state store with get_members/get_member support."""
store = MagicMock()
# get_members returns a list of member user IDs
members = [MagicMock() for _ in range(member_count)]
store.get_members = AsyncMock(return_value=members)
# get_member returns a single member info object
member = MagicMock()
member.displayname = "Alice"
store.get_member = AsyncMock(return_value=member)
return store
# ---------------------------------------------------------------------------
# Tests: MSC3245 Voice Detection (RED -> GREEN)
# Tests: MSC3245 Voice Detection
# ---------------------------------------------------------------------------
class TestMatrixVoiceMessageDetection:
@@ -118,27 +113,28 @@ class TestMatrixVoiceMessageDetection:
self.adapter._message_handler = AsyncMock()
# Mock _mxc_to_http to return a fake HTTP URL
self.adapter._mxc_to_http = lambda url: f"https://matrix.example.org/_matrix/media/v3/download/{url[6:]}"
# Mock client for authenticated download
# Mock client for authenticated download — download_media returns bytes directly
self.adapter._client = MagicMock()
self.adapter._client.download = AsyncMock(return_value=_make_download_response())
self.adapter._client.download_media = AsyncMock(return_value=b"fake audio data")
# State store for DM detection
self.adapter._client.state_store = _make_state_store()
@pytest.mark.asyncio
async def test_voice_message_has_type_voice(self):
"""Voice messages (with MSC3245 field) should be MessageType.VOICE."""
room = _make_room()
event = _make_audio_event(is_voice=True)
# Capture the MessageEvent passed to handle_message
captured_event = None
async def capture(msg_event):
nonlocal captured_event
captured_event = msg_event
self.adapter.handle_message = capture
await self.adapter._on_room_message_media(room, event)
await self.adapter._on_room_message(event)
assert captured_event is not None, "No event was captured"
assert captured_event.message_type == MessageType.VOICE, \
f"Expected MessageType.VOICE, got {captured_event.message_type}"
@@ -146,44 +142,43 @@ class TestMatrixVoiceMessageDetection:
@pytest.mark.asyncio
async def test_voice_message_has_local_path(self):
"""Voice messages should have a local cached path in media_urls."""
room = _make_room()
event = _make_audio_event(is_voice=True)
captured_event = None
async def capture(msg_event):
nonlocal captured_event
captured_event = msg_event
self.adapter.handle_message = capture
await self.adapter._on_room_message_media(room, event)
await self.adapter._on_room_message(event)
assert captured_event is not None
assert captured_event.media_urls is not None
assert len(captured_event.media_urls) > 0
# Should be a local path, not an HTTP URL
assert not captured_event.media_urls[0].startswith("http"), \
f"media_urls should contain local path, got {captured_event.media_urls[0]}"
self.adapter._client.download.assert_awaited_once_with(mxc=event.url)
# download_media is called with a ContentURI wrapping the mxc URL
self.adapter._client.download_media.assert_awaited_once()
assert captured_event.media_types == ["audio/ogg"]
@pytest.mark.asyncio
async def test_audio_without_msc3245_stays_audio_type(self):
"""Regular audio uploads (no MSC3245 field) should remain MessageType.AUDIO."""
room = _make_room()
event = _make_audio_event(is_voice=False) # NOT a voice message
captured_event = None
async def capture(msg_event):
nonlocal captured_event
captured_event = msg_event
self.adapter.handle_message = capture
await self.adapter._on_room_message_media(room, event)
await self.adapter._on_room_message(event)
assert captured_event is not None
assert captured_event.message_type == MessageType.AUDIO, \
f"Expected MessageType.AUDIO for non-voice, got {captured_event.message_type}"
@@ -191,25 +186,24 @@ class TestMatrixVoiceMessageDetection:
@pytest.mark.asyncio
async def test_regular_audio_has_http_url(self):
"""Regular audio uploads should keep HTTP URL (not cached locally)."""
room = _make_room()
event = _make_audio_event(is_voice=False)
captured_event = None
async def capture(msg_event):
nonlocal captured_event
captured_event = msg_event
self.adapter.handle_message = capture
await self.adapter._on_room_message_media(room, event)
await self.adapter._on_room_message(event)
assert captured_event is not None
assert captured_event.media_urls is not None
# Should be HTTP URL, not local path
assert captured_event.media_urls[0].startswith("http"), \
f"Non-voice audio should have HTTP URL, got {captured_event.media_urls[0]}"
self.adapter._client.download.assert_not_awaited()
self.adapter._client.download_media.assert_not_awaited()
assert captured_event.media_types == ["audio/ogg"]
@@ -224,29 +218,26 @@ class TestMatrixVoiceCacheFallback:
self.adapter._message_handler = AsyncMock()
self.adapter._mxc_to_http = lambda url: f"https://matrix.example.org/_matrix/media/v3/download/{url[6:]}"
self.adapter._client = MagicMock()
self.adapter._client.state_store = _make_state_store()
@pytest.mark.asyncio
async def test_voice_cache_failure_falls_back_to_http_url(self):
"""If caching fails, voice message should still be delivered with HTTP URL."""
room = _make_room()
"""If caching fails (download returns None), voice message should still be delivered with HTTP URL."""
event = _make_audio_event(is_voice=True)
# Make download fail
import nio
error_resp = MagicMock()
error_resp.__class__ = nio.DownloadError
self.adapter._client.download = AsyncMock(return_value=error_resp)
# download_media returns None on failure
self.adapter._client.download_media = AsyncMock(return_value=None)
captured_event = None
async def capture(msg_event):
nonlocal captured_event
captured_event = msg_event
self.adapter.handle_message = capture
await self.adapter._on_room_message_media(room, event)
await self.adapter._on_room_message(event)
assert captured_event is not None
assert captured_event.media_urls is not None
# Should fall back to HTTP URL
@@ -256,10 +247,9 @@ class TestMatrixVoiceCacheFallback:
@pytest.mark.asyncio
async def test_voice_cache_exception_falls_back_to_http_url(self):
"""Unexpected download exceptions should also fall back to HTTP URL."""
room = _make_room()
event = _make_audio_event(is_voice=True)
self.adapter._client.download = AsyncMock(side_effect=RuntimeError("boom"))
self.adapter._client.download_media = AsyncMock(side_effect=RuntimeError("boom"))
captured_event = None
@@ -269,7 +259,7 @@ class TestMatrixVoiceCacheFallback:
self.adapter.handle_message = capture
await self.adapter._on_room_message_media(room, event)
await self.adapter._on_room_message(event)
assert captured_event is not None
assert captured_event.media_urls is not None
@@ -278,7 +268,7 @@ class TestMatrixVoiceCacheFallback:
# ---------------------------------------------------------------------------
# Tests: send_voice includes MSC3245 field (RED -> GREEN)
# Tests: send_voice includes MSC3245 field
# ---------------------------------------------------------------------------
class TestMatrixSendVoiceMSC3245:
@@ -287,62 +277,52 @@ class TestMatrixSendVoiceMSC3245:
def setup_method(self):
self.adapter = _make_adapter()
self.adapter._user_id = "@bot:example.org"
# Mock client with successful upload
# Mock client — upload_media returns a ContentURI string
self.adapter._client = MagicMock()
self.upload_call = None
async def mock_upload(*args, **kwargs):
self.upload_call = (args, kwargs)
import nio
resp = MagicMock()
resp.content_uri = "mxc://example.org/uploaded"
resp.__class__ = nio.UploadResponse
return resp, None
async def mock_upload_media(data, mime_type=None, filename=None, **kwargs):
self.upload_call = {"data": data, "mime_type": mime_type, "filename": filename}
return "mxc://example.org/uploaded"
self.adapter._client.upload = mock_upload
self.adapter._client.upload_media = mock_upload_media
@pytest.mark.asyncio
async def test_send_voice_includes_msc3245_field(self):
@patch("mimetypes.guess_type", return_value=("audio/ogg", None))
async def test_send_voice_includes_msc3245_field(self, _mock_guess):
"""send_voice should include org.matrix.msc3245.voice in message content."""
import tempfile
import os
# Create a temp audio file
with tempfile.NamedTemporaryFile(suffix=".ogg", delete=False) as f:
f.write(b"fake audio data")
temp_path = f.name
try:
# Capture the message content sent to room_send
# Capture the message content sent via send_message_event
sent_content = None
async def mock_room_send(room_id, event_type, content):
async def mock_send_message_event(room_id, event_type, content):
nonlocal sent_content
sent_content = content
resp = MagicMock()
resp.event_id = "$sent_event"
import nio
resp.__class__ = nio.RoomSendResponse
return resp
self.adapter._client.room_send = mock_room_send
# send_message_event returns an EventID string
return "$sent_event"
self.adapter._client.send_message_event = mock_send_message_event
await self.adapter.send_voice(
chat_id="!room:example.org",
audio_path=temp_path,
caption="Test voice",
)
assert sent_content is not None, "No message was sent"
assert "org.matrix.msc3245.voice" in sent_content, \
f"MSC3245 voice field missing from content: {sent_content.keys()}"
assert sent_content["msgtype"] == "m.audio"
assert sent_content["info"]["mimetype"] == "audio/ogg"
assert self.upload_call is not None, "Expected upload() to be called"
args, kwargs = self.upload_call
assert isinstance(args[0], io.BytesIO)
assert kwargs["content_type"] == "audio/ogg"
assert kwargs["filename"].endswith(".ogg")
assert self.upload_call is not None, "Expected upload_media() to be called"
assert isinstance(self.upload_call["data"], bytes)
assert self.upload_call["mime_type"] == "audio/ogg"
assert self.upload_call["filename"].endswith(".ogg")
finally:
os.unlink(temp_path)