shawn
3ddf8170f8
- rooms: create, join, leave, forget, list, resolve alias, get info/members, invite/kick/ban/unban, set name/topic. RoomInfo/MemberInfo/PowerLevels dataclasses. - messaging: send text (with HTML, reply, thread), notice, emote, image, file, sticker, edit, redact, react, typing, read receipts, fully-read markers. - sync: SyncEngine with event-type/room/global handler dispatch, incremental sync with backoff retry, SyncFilter, paginated message history. - CLI: rooms (list/create/join/leave/info/members/invite), send (text/image/file), sync (listen with filters). - 182 tests pass (60 existing + 122 new).
361 lines
13 KiB
Python
361 lines
13 KiB
Python
"""Test Matrix messaging: send text, media, edits, reactions, typing, receipts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from matrix_hermes.messaging import (
|
|
SendResult,
|
|
edit_message,
|
|
redact_message,
|
|
send_emote,
|
|
send_file,
|
|
send_image,
|
|
send_notice,
|
|
send_reaction,
|
|
send_read_receipt,
|
|
send_sticker,
|
|
send_text,
|
|
send_typing,
|
|
set_fully_read,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_client():
|
|
"""Return a mock mautrix Client with all messaging methods as AsyncMock."""
|
|
client = MagicMock()
|
|
client.send_text = AsyncMock(return_value="$event123:example.org")
|
|
client.send_message = AsyncMock(return_value="$event456:example.org")
|
|
client.send_message_event = AsyncMock(return_value="$edit789:example.org")
|
|
client.send_emote = AsyncMock(return_value="$emote000:example.org")
|
|
client.send_image = AsyncMock(return_value="$img111:example.org")
|
|
client.send_file = AsyncMock(return_value="$file222:example.org")
|
|
client.send_sticker = AsyncMock(return_value="$sticker333:example.org")
|
|
client.redact = AsyncMock()
|
|
client.react = AsyncMock()
|
|
client.set_typing = AsyncMock()
|
|
client.send_receipt = AsyncMock()
|
|
client.set_fully_read_marker = AsyncMock()
|
|
return client
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data class tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSendResult:
|
|
def test_success(self) -> None:
|
|
r = SendResult(success=True, event_id="$abc:example.org")
|
|
assert r.success is True
|
|
assert r.event_id == "$abc:example.org"
|
|
assert r.error == ""
|
|
|
|
def test_failure(self) -> None:
|
|
r = SendResult(success=False, error="Rate limited")
|
|
assert r.success is False
|
|
assert r.error == "Rate limited"
|
|
assert r.event_id == ""
|
|
|
|
def test_default_values(self) -> None:
|
|
r = SendResult(success=True)
|
|
assert r.event_id == ""
|
|
assert r.error == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Core messaging
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSendText:
|
|
@pytest.mark.asyncio
|
|
async def test_send_simple(self, mock_client) -> None:
|
|
result = await send_text(mock_client, "!room:example.org", "Hello world")
|
|
assert result.success is True
|
|
assert result.event_id == "$event123:example.org"
|
|
mock_client.send_text.assert_called_once_with(
|
|
"!room:example.org",
|
|
text="Hello world",
|
|
html=None,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_html(self, mock_client) -> None:
|
|
result = await send_text(
|
|
mock_client, "!room:example.org", "Hello", html="<b>Hello</b>"
|
|
)
|
|
assert result.success is True
|
|
mock_client.send_text.assert_called_once_with(
|
|
"!room:example.org",
|
|
text="Hello",
|
|
html="<b>Hello</b>",
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_with_reply(self, mock_client) -> None:
|
|
result = await send_text(
|
|
mock_client, "!room:example.org", "Replying!", reply_to="$orig:example.org"
|
|
)
|
|
assert result.success is True
|
|
call_kwargs = mock_client.send_text.call_args[1]
|
|
assert "relates_to" in call_kwargs
|
|
assert call_kwargs["relates_to"]["m.in_reply_to"]["event_id"] == "$orig:example.org"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_in_thread(self, mock_client) -> None:
|
|
result = await send_text(
|
|
mock_client, "!room:example.org", "Thread reply", thread_root="$thread:example.org"
|
|
)
|
|
assert result.success is True
|
|
call_kwargs = mock_client.send_text.call_args[1]
|
|
assert "relates_to" in call_kwargs
|
|
assert call_kwargs["relates_to"]["rel_type"] == "m.thread"
|
|
assert call_kwargs["relates_to"]["event_id"] == "$thread:example.org"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_failure(self, mock_client) -> None:
|
|
mock_client.send_text.side_effect = Exception("Network error")
|
|
result = await send_text(mock_client, "!room:example.org", "fail")
|
|
assert result.success is False
|
|
assert "Network error" in result.error
|
|
|
|
|
|
class TestSendNotice:
|
|
@pytest.mark.asyncio
|
|
async def test_send_notice(self, mock_client) -> None:
|
|
result = await send_notice(mock_client, "!room:example.org", "Notice text")
|
|
assert result.success is True
|
|
assert result.event_id == "$event456:example.org"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_notice_with_html(self, mock_client) -> None:
|
|
result = await send_notice(
|
|
mock_client, "!room:example.org", "Notice", html="<i>Notice</i>"
|
|
)
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_notice_failure(self, mock_client) -> None:
|
|
mock_client.send_message.side_effect = Exception("Fail")
|
|
result = await send_notice(mock_client, "!room:example.org", "fail")
|
|
assert result.success is False
|
|
|
|
|
|
class TestSendEmote:
|
|
@pytest.mark.asyncio
|
|
async def test_send_emote(self, mock_client) -> None:
|
|
result = await send_emote(mock_client, "!room:example.org", "waves")
|
|
assert result.success is True
|
|
assert result.event_id == "$emote000:example.org"
|
|
mock_client.send_emote.assert_called_once_with("!room:example.org", "waves")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_emote_failure(self, mock_client) -> None:
|
|
mock_client.send_emote.side_effect = Exception("Fail")
|
|
result = await send_emote(mock_client, "!room:example.org", "fail")
|
|
assert result.success is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Media
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSendImage:
|
|
@pytest.mark.asyncio
|
|
async def test_send_image(self, mock_client) -> None:
|
|
result = await send_image(mock_client, "!room:example.org", "/tmp/photo.jpg")
|
|
assert result.success is True
|
|
assert result.event_id == "$img111:example.org"
|
|
mock_client.send_image.assert_called_once_with(
|
|
"!room:example.org",
|
|
"/tmp/photo.jpg",
|
|
caption=None,
|
|
thread_root=None,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_image_with_caption(self, mock_client) -> None:
|
|
result = await send_image(
|
|
mock_client, "!room:example.org", "/tmp/photo.jpg", caption="My photo"
|
|
)
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_image_failure(self, mock_client) -> None:
|
|
mock_client.send_image.side_effect = Exception("Upload failed")
|
|
result = await send_image(mock_client, "!room:example.org", "/tmp/bad.jpg")
|
|
assert result.success is False
|
|
|
|
|
|
class TestSendFile:
|
|
@pytest.mark.asyncio
|
|
async def test_send_file(self, mock_client) -> None:
|
|
result = await send_file(
|
|
mock_client, "!room:example.org", "/tmp/doc.pdf", filename="report.pdf"
|
|
)
|
|
assert result.success is True
|
|
assert result.event_id == "$file222:example.org"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_file_failure(self, mock_client) -> None:
|
|
mock_client.send_file.side_effect = Exception("Upload failed")
|
|
result = await send_file(mock_client, "!room:example.org", "/tmp/bad.pdf")
|
|
assert result.success is False
|
|
|
|
|
|
class TestSendSticker:
|
|
@pytest.mark.asyncio
|
|
async def test_send_sticker(self, mock_client) -> None:
|
|
result = await send_sticker(mock_client, "!room:example.org", "/tmp/sticker.png")
|
|
assert result.success is True
|
|
mock_client.send_sticker.assert_called_once_with("!room:example.org", "/tmp/sticker.png")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_sticker_failure(self, mock_client) -> None:
|
|
mock_client.send_sticker.side_effect = Exception("Fail")
|
|
result = await send_sticker(mock_client, "!room:example.org", "/tmp/sticker.png")
|
|
assert result.success is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Editing & redaction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEditMessage:
|
|
@pytest.mark.asyncio
|
|
async def test_edit_message(self, mock_client) -> None:
|
|
result = await edit_message(
|
|
mock_client, "!room:example.org", "$orig:example.org", "Fixed text"
|
|
)
|
|
assert result.success is True
|
|
assert result.event_id == "$edit789:example.org"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_message_with_html(self, mock_client) -> None:
|
|
result = await edit_message(
|
|
mock_client,
|
|
"!room:example.org",
|
|
"$orig:example.org",
|
|
"Fixed **bold**",
|
|
new_html="Fixed <b>bold</b>",
|
|
)
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_message_failure(self, mock_client) -> None:
|
|
mock_client.send_message_event.side_effect = Exception("Fail")
|
|
result = await edit_message(
|
|
mock_client, "!room:example.org", "$orig:example.org", "edit"
|
|
)
|
|
assert result.success is False
|
|
|
|
|
|
class TestRedactMessage:
|
|
@pytest.mark.asyncio
|
|
async def test_redact_success(self, mock_client) -> None:
|
|
ok = await redact_message(mock_client, "!room:example.org", "$msg:example.org")
|
|
assert ok is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redact_with_reason(self, mock_client) -> None:
|
|
ok = await redact_message(
|
|
mock_client, "!room:example.org", "$msg:example.org", reason="Offensive"
|
|
)
|
|
assert ok is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redact_failure(self, mock_client) -> None:
|
|
mock_client.redact.side_effect = Exception("No permission")
|
|
ok = await redact_message(mock_client, "!room:example.org", "$msg:example.org")
|
|
assert ok is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reactions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSendReaction:
|
|
@pytest.mark.asyncio
|
|
async def test_send_reaction(self, mock_client) -> None:
|
|
ok = await send_reaction(
|
|
mock_client, "!room:example.org", "$msg:example.org", "👍"
|
|
)
|
|
assert ok is True
|
|
mock_client.react.assert_called_once_with("!room:example.org", "$msg:example.org", "👍")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_reaction_failure(self, mock_client) -> None:
|
|
mock_client.react.side_effect = Exception("Fail")
|
|
ok = await send_reaction(mock_client, "!room:example.org", "$msg:example.org", "❤️")
|
|
assert ok is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Typing & receipts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSendTyping:
|
|
@pytest.mark.asyncio
|
|
async def test_typing_on(self, mock_client) -> None:
|
|
ok = await send_typing(mock_client, "!room:example.org", typing=True, timeout=5000)
|
|
assert ok is True
|
|
mock_client.set_typing.assert_called_once_with(
|
|
"!room:example.org", typing=True, timeout=5000
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_typing_off(self, mock_client) -> None:
|
|
ok = await send_typing(mock_client, "!room:example.org", typing=False)
|
|
assert ok is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_typing_failure(self, mock_client) -> None:
|
|
mock_client.set_typing.side_effect = Exception("Fail")
|
|
ok = await send_typing(mock_client, "!room:example.org")
|
|
assert ok is False
|
|
|
|
|
|
class TestSendReadReceipt:
|
|
@pytest.mark.asyncio
|
|
async def test_send_receipt(self, mock_client) -> None:
|
|
ok = await send_read_receipt(mock_client, "!room:example.org", "$msg:example.org")
|
|
assert ok is True
|
|
mock_client.send_receipt.assert_called_once_with(
|
|
"!room:example.org", "$msg:example.org", "m.read"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_receipt_failure(self, mock_client) -> None:
|
|
mock_client.send_receipt.side_effect = Exception("Fail")
|
|
ok = await send_read_receipt(mock_client, "!room:example.org", "$msg:example.org")
|
|
assert ok is False
|
|
|
|
|
|
class TestSetFullyRead:
|
|
@pytest.mark.asyncio
|
|
async def test_set_fully_read(self, mock_client) -> None:
|
|
ok = await set_fully_read(mock_client, "!room:example.org", "$msg:example.org")
|
|
assert ok is True
|
|
mock_client.set_fully_read_marker.assert_called_once_with(
|
|
"!room:example.org", "$msg:example.org"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_fully_read_failure(self, mock_client) -> None:
|
|
mock_client.set_fully_read_marker.side_effect = Exception("Fail")
|
|
ok = await set_fully_read(mock_client, "!room:example.org", "$msg:example.org")
|
|
assert ok is False
|