ee61485cac
* feat(matrix): support native voice messages * fix: skip matrix voice tests when matrix-nio not installed --------- Co-authored-by: Carlos Alberto Pereira Gomes <carlosapgomes@users.noreply.github.com>
341 lines
12 KiB
Python
341 lines
12 KiB
Python
"""Tests for Matrix voice message support (MSC3245)."""
|
|
import io
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
nio = pytest.importorskip("nio", reason="matrix-nio not installed")
|
|
|
|
from gateway.platforms.base import MessageType
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adapter helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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="***",
|
|
extra={
|
|
"homeserver": "https://matrix.example.org",
|
|
"user_id": "@bot:example.org",
|
|
},
|
|
)
|
|
adapter = MatrixAdapter(config)
|
|
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",
|
|
body: str = "Voice message",
|
|
url: str = "mxc://example.org/abc123",
|
|
is_voice: bool = False,
|
|
mimetype: str = "audio/ogg",
|
|
timestamp: float = 9999999999000, # ms
|
|
):
|
|
"""
|
|
Create a mock RoomMessageAudio event that passes isinstance checks.
|
|
|
|
Args:
|
|
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,
|
|
"url": url,
|
|
"info": {
|
|
"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
|
|
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: MSC3245 Voice Detection (RED -> GREEN)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMatrixVoiceMessageDetection:
|
|
"""Test that MSC3245 voice messages are detected and tagged correctly."""
|
|
|
|
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()
|
|
# 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
|
|
self.adapter._client = MagicMock()
|
|
self.adapter._client.download = AsyncMock(return_value=_make_download_response())
|
|
|
|
@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)
|
|
|
|
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}"
|
|
|
|
@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)
|
|
|
|
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)
|
|
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)
|
|
|
|
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}"
|
|
|
|
@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)
|
|
|
|
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()
|
|
assert captured_event.media_types == ["audio/ogg"]
|
|
|
|
|
|
class TestMatrixVoiceCacheFallback:
|
|
"""Test graceful fallback when voice caching fails."""
|
|
|
|
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()
|
|
self.adapter._mxc_to_http = lambda url: f"https://matrix.example.org/_matrix/media/v3/download/{url[6:]}"
|
|
self.adapter._client = MagicMock()
|
|
|
|
@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()
|
|
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)
|
|
|
|
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)
|
|
|
|
assert captured_event is not None
|
|
assert captured_event.media_urls is not None
|
|
# Should fall back to HTTP URL
|
|
assert captured_event.media_urls[0].startswith("http"), \
|
|
f"Should fall back to HTTP URL on cache failure, got {captured_event.media_urls[0]}"
|
|
|
|
@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"))
|
|
|
|
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)
|
|
|
|
assert captured_event is not None
|
|
assert captured_event.media_urls is not None
|
|
assert captured_event.media_urls[0].startswith("http"), \
|
|
f"Should fall back to HTTP URL on exception, got {captured_event.media_urls[0]}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: send_voice includes MSC3245 field (RED -> GREEN)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMatrixSendVoiceMSC3245:
|
|
"""Test that send_voice includes MSC3245 field for native voice rendering."""
|
|
|
|
def setup_method(self):
|
|
self.adapter = _make_adapter()
|
|
self.adapter._user_id = "@bot:example.org"
|
|
# Mock client with successful upload
|
|
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
|
|
|
|
self.adapter._client.upload = mock_upload
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_voice_includes_msc3245_field(self):
|
|
"""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
|
|
sent_content = None
|
|
|
|
async def mock_room_send(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
|
|
|
|
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")
|
|
|
|
finally:
|
|
os.unlink(temp_path)
|