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).
459 lines
17 KiB
Python
459 lines
17 KiB
Python
"""Test Matrix room management: create, join, leave, info, members, invite, kick, ban."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from matrix_hermes.rooms import (
|
|
CreateRoomResult,
|
|
MemberInfo,
|
|
PowerLevels,
|
|
RoomInfo,
|
|
ban_user,
|
|
create_room,
|
|
forget_room,
|
|
get_joined_rooms,
|
|
get_room_info,
|
|
get_room_members,
|
|
invite_user,
|
|
join_room,
|
|
kick_user,
|
|
leave_room,
|
|
resolve_room_alias,
|
|
set_room_name,
|
|
set_room_topic,
|
|
unban_user,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_client():
|
|
"""Return a mock mautrix Client with all room methods as AsyncMock."""
|
|
client = MagicMock()
|
|
client.join_room = AsyncMock()
|
|
client.leave_room = AsyncMock()
|
|
client.forget_room = AsyncMock()
|
|
client.get_joined_rooms = AsyncMock(return_value=["!room1:example.org", "!room2:example.org"])
|
|
client.create_room = AsyncMock(return_value="!newroom:example.org")
|
|
client.resolve_room_alias = AsyncMock(
|
|
return_value={"room_id": "!room1:example.org", "servers": ["example.org"]}
|
|
)
|
|
client.get_state_event = AsyncMock()
|
|
client.get_members = AsyncMock()
|
|
client.invite_user = AsyncMock()
|
|
client.kick_user = AsyncMock()
|
|
client.ban_user = AsyncMock()
|
|
client.unban_user = AsyncMock()
|
|
client.send_state_event = AsyncMock()
|
|
return client
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data class tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRoomInfo:
|
|
def test_default_values(self) -> None:
|
|
info = RoomInfo(room_id="!test:example.org")
|
|
assert info.room_id == "!test:example.org"
|
|
assert info.name == ""
|
|
assert info.topic == ""
|
|
assert info.member_count == 0
|
|
assert info.is_direct is False
|
|
|
|
def test_full_info(self) -> None:
|
|
info = RoomInfo(
|
|
room_id="!test:example.org",
|
|
name="Test Room",
|
|
topic="Testing",
|
|
avatar_url="mxc://example.org/avatar",
|
|
canonical_alias="#test:example.org",
|
|
member_count=42,
|
|
joined_members=40,
|
|
invited_members=2,
|
|
is_direct=False,
|
|
is_public=True,
|
|
encryption=True,
|
|
)
|
|
assert info.name == "Test Room"
|
|
assert info.topic == "Testing"
|
|
assert info.is_public is True
|
|
assert info.encryption is True
|
|
assert info.joined_members == 40
|
|
|
|
|
|
class TestCreateRoomResult:
|
|
def test_success(self) -> None:
|
|
r = CreateRoomResult(success=True, room_id="!abc:example.org")
|
|
assert r.success is True
|
|
assert r.room_id == "!abc:example.org"
|
|
assert r.error == ""
|
|
|
|
def test_failure(self) -> None:
|
|
r = CreateRoomResult(success=False, error="Server error")
|
|
assert r.success is False
|
|
assert r.error == "Server error"
|
|
assert r.room_id == ""
|
|
|
|
|
|
class TestMemberInfo:
|
|
def test_basic(self) -> None:
|
|
m = MemberInfo(user_id="@alice:example.org", display_name="Alice")
|
|
assert m.user_id == "@alice:example.org"
|
|
assert m.display_name == "Alice"
|
|
assert m.membership == ""
|
|
assert m.power_level == 0
|
|
|
|
def test_full(self) -> None:
|
|
m = MemberInfo(
|
|
user_id="@bob:example.org",
|
|
display_name="Bob",
|
|
avatar_url="mxc://example.org/bob",
|
|
membership="join",
|
|
power_level=50,
|
|
)
|
|
assert m.power_level == 50
|
|
assert m.membership == "join"
|
|
|
|
|
|
class TestPowerLevels:
|
|
def test_defaults(self) -> None:
|
|
pl = PowerLevels()
|
|
assert pl.users_default == 0
|
|
assert pl.ban == 50
|
|
assert pl.invite == 0
|
|
|
|
def test_custom(self) -> None:
|
|
pl = PowerLevels(
|
|
users={"@admin:example.org": 100},
|
|
users_default=0,
|
|
ban=60,
|
|
invite=50,
|
|
)
|
|
assert pl.users["@admin:example.org"] == 100
|
|
assert pl.ban == 60
|
|
assert pl.invite == 50
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Room operations tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCreateRoom:
|
|
@pytest.mark.asyncio
|
|
async def test_create_room_basic(self, mock_client) -> None:
|
|
result = await create_room(mock_client, name="My Room")
|
|
assert result.success is True
|
|
assert result.room_id == "!newroom:example.org"
|
|
mock_client.create_room.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_room_with_all_options(self, mock_client) -> None:
|
|
result = await create_room(
|
|
mock_client,
|
|
name="Public Room",
|
|
topic="Hello world",
|
|
alias_localpart="myroom",
|
|
preset="public_chat",
|
|
visibility="public",
|
|
invitees=["@alice:example.org"],
|
|
is_direct=False,
|
|
room_version="10",
|
|
)
|
|
assert result.success is True
|
|
call_kwargs = mock_client.create_room.call_args[1]
|
|
assert call_kwargs["name"] == "Public Room"
|
|
assert call_kwargs["topic"] == "Hello world"
|
|
assert call_kwargs["preset"] == "public_chat"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_room_failure(self, mock_client) -> None:
|
|
mock_client.create_room.side_effect = Exception("Server error")
|
|
result = await create_room(mock_client, name="Fail")
|
|
assert result.success is False
|
|
assert "Server error" in result.error
|
|
|
|
|
|
class TestJoinRoom:
|
|
@pytest.mark.asyncio
|
|
async def test_join_success(self, mock_client) -> None:
|
|
ok = await join_room(mock_client, "!room:example.org")
|
|
assert ok is True
|
|
mock_client.join_room.assert_called_once_with("!room:example.org", servers=None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_with_servers(self, mock_client) -> None:
|
|
ok = await join_room(mock_client, "#room:example.org", servers=["matrix.org"])
|
|
assert ok is True
|
|
mock_client.join_room.assert_called_once_with("#room:example.org", servers=["matrix.org"])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_failure(self, mock_client) -> None:
|
|
mock_client.join_room.side_effect = Exception("Room not found")
|
|
ok = await join_room(mock_client, "!bad:example.org")
|
|
assert ok is False
|
|
|
|
|
|
class TestLeaveRoom:
|
|
@pytest.mark.asyncio
|
|
async def test_leave_success(self, mock_client) -> None:
|
|
ok = await leave_room(mock_client, "!room:example.org")
|
|
assert ok is True
|
|
mock_client.leave_room.assert_called_once_with("!room:example.org")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leave_failure(self, mock_client) -> None:
|
|
mock_client.leave_room.side_effect = Exception("Not joined")
|
|
ok = await leave_room(mock_client, "!room:example.org")
|
|
assert ok is False
|
|
|
|
|
|
class TestForgetRoom:
|
|
@pytest.mark.asyncio
|
|
async def test_forget_success(self, mock_client) -> None:
|
|
ok = await forget_room(mock_client, "!room:example.org")
|
|
assert ok is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_forget_failure(self, mock_client) -> None:
|
|
mock_client.forget_room.side_effect = Exception("Error")
|
|
ok = await forget_room(mock_client, "!room:example.org")
|
|
assert ok is False
|
|
|
|
|
|
class TestGetJoinedRooms:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_rooms(self, mock_client) -> None:
|
|
rooms = await get_joined_rooms(mock_client)
|
|
assert rooms == ["!room1:example.org", "!room2:example.org"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_empty_on_error(self, mock_client) -> None:
|
|
mock_client.get_joined_rooms.side_effect = Exception("Network error")
|
|
rooms = await get_joined_rooms(mock_client)
|
|
assert rooms == []
|
|
|
|
|
|
class TestResolveRoomAlias:
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_success(self, mock_client) -> None:
|
|
room_id, servers = await resolve_room_alias(mock_client, "#test:example.org")
|
|
assert room_id == "!room1:example.org"
|
|
assert servers == ["example.org"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_failure(self, mock_client) -> None:
|
|
mock_client.resolve_room_alias.side_effect = Exception("Not found")
|
|
room_id, servers = await resolve_room_alias(mock_client, "#bad:example.org")
|
|
assert room_id == ""
|
|
assert servers == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Room info & members
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetRoomInfo:
|
|
@pytest.mark.asyncio
|
|
async def test_get_room_info_basic(self, mock_client) -> None:
|
|
mock_client.get_state_event = AsyncMock(side_effect=lambda room_id, event_type: {
|
|
"m.room.name": {"name": "Test Room"},
|
|
"m.room.topic": {"topic": "Testing stuff"},
|
|
"m.room.avatar": {"url": "mxc://example.org/avatar"},
|
|
"m.room.canonical_alias": {"alias": "#test:example.org"},
|
|
"m.room.join_rules": {"join_rule": "invite"},
|
|
"m.room.encryption": {"algorithm": "m.megolm.v1.aes-sha2"},
|
|
}.get(event_type, None))
|
|
|
|
mock_client.get_members = AsyncMock(return_value=[
|
|
{"state_key": "@a:example.org", "content": {"membership": "join", "displayname": "A"}},
|
|
{"state_key": "@b:example.org", "content": {"membership": "join", "displayname": "B"}},
|
|
{"state_key": "@c:example.org", "content": {"membership": "invite", "displayname": "C"}},
|
|
])
|
|
|
|
info = await get_room_info(mock_client, "!room:example.org")
|
|
assert info is not None
|
|
assert info.name == "Test Room"
|
|
assert info.topic == "Testing stuff"
|
|
assert info.avatar_url == "mxc://example.org/avatar"
|
|
assert info.canonical_alias == "#test:example.org"
|
|
assert info.joined_members == 2
|
|
assert info.invited_members == 1
|
|
assert info.member_count == 3
|
|
assert info.encryption is True
|
|
assert info.is_public is False
|
|
# joined==2 and not is_public → heuristically direct
|
|
assert info.is_direct is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_room_info_public(self, mock_client) -> None:
|
|
mock_client.get_state_event = AsyncMock(side_effect=lambda room_id, event_type: {
|
|
"m.room.join_rules": {"join_rule": "public"},
|
|
}.get(event_type, None))
|
|
mock_client.get_members = AsyncMock(return_value=[
|
|
{"state_key": "@a:example.org", "content": {"membership": "join"}},
|
|
{"state_key": "@b:example.org", "content": {"membership": "join"}},
|
|
])
|
|
|
|
info = await get_room_info(mock_client, "!room:example.org")
|
|
assert info is not None
|
|
assert info.is_public is True
|
|
# public room + not direct since is_public → not direct
|
|
assert info.is_direct is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_room_info_error(self, mock_client) -> None:
|
|
# Individual state event exceptions are caught internally,
|
|
# so we get a RoomInfo with defaults. Only a total failure
|
|
# (e.g. get_members failing) returns None.
|
|
mock_client.get_state_event = AsyncMock(side_effect=Exception("Failed"))
|
|
mock_client.get_members = AsyncMock(return_value=[])
|
|
info = await get_room_info(mock_client, "!room:example.org")
|
|
assert info is not None
|
|
assert info.room_id == "!room:example.org"
|
|
assert info.name == "" # failed state events → empty defaults
|
|
|
|
|
|
class TestGetRoomMembers:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_members(self, mock_client) -> None:
|
|
mock_client.get_members = AsyncMock(return_value=[
|
|
{
|
|
"state_key": "@alice:example.org",
|
|
"content": {"membership": "join", "displayname": "Alice", "avatar_url": "mxc://a"},
|
|
},
|
|
{
|
|
"state_key": "@bob:example.org",
|
|
"content": {"membership": "join", "displayname": "Bob"},
|
|
},
|
|
])
|
|
members = await get_room_members(mock_client, "!room:example.org")
|
|
assert len(members) == 2
|
|
assert members[0].user_id == "@alice:example.org"
|
|
assert members[0].display_name == "Alice"
|
|
assert members[0].avatar_url == "mxc://a"
|
|
assert members[0].membership == "join"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_empty_on_error(self, mock_client) -> None:
|
|
mock_client.get_members.side_effect = Exception("Error")
|
|
members = await get_room_members(mock_client, "!room:example.org")
|
|
assert members == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Member management
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestInviteUser:
|
|
@pytest.mark.asyncio
|
|
async def test_invite_success(self, mock_client) -> None:
|
|
ok = await invite_user(mock_client, "!room:example.org", "@alice:example.org")
|
|
assert ok is True
|
|
mock_client.invite_user.assert_called_once_with(
|
|
"!room:example.org", "@alice:example.org", reason=None
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invite_with_reason(self, mock_client) -> None:
|
|
ok = await invite_user(
|
|
mock_client, "!room:example.org", "@bob:example.org", reason="Welcome!"
|
|
)
|
|
assert ok is True
|
|
mock_client.invite_user.assert_called_once_with(
|
|
"!room:example.org", "@bob:example.org", reason="Welcome!"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invite_failure(self, mock_client) -> None:
|
|
mock_client.invite_user.side_effect = Exception("User blocked")
|
|
ok = await invite_user(mock_client, "!room:example.org", "@bad:example.org")
|
|
assert ok is False
|
|
|
|
|
|
class TestKickUser:
|
|
@pytest.mark.asyncio
|
|
async def test_kick_success(self, mock_client) -> None:
|
|
ok = await kick_user(mock_client, "!room:example.org", "@spammer:example.org")
|
|
assert ok is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_kick_failure(self, mock_client) -> None:
|
|
mock_client.kick_user.side_effect = Exception("No permission")
|
|
ok = await kick_user(mock_client, "!room:example.org", "@admin:example.org")
|
|
assert ok is False
|
|
|
|
|
|
class TestBanUser:
|
|
@pytest.mark.asyncio
|
|
async def test_ban_success(self, mock_client) -> None:
|
|
ok = await ban_user(mock_client, "!room:example.org", "@spammer:example.org", reason="Spam")
|
|
assert ok is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ban_failure(self, mock_client) -> None:
|
|
mock_client.ban_user.side_effect = Exception("No permission")
|
|
ok = await ban_user(mock_client, "!room:example.org", "@admin:example.org")
|
|
assert ok is False
|
|
|
|
|
|
class TestUnbanUser:
|
|
@pytest.mark.asyncio
|
|
async def test_unban_success(self, mock_client) -> None:
|
|
ok = await unban_user(mock_client, "!room:example.org", "@former_spammer:example.org")
|
|
assert ok is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unban_failure(self, mock_client) -> None:
|
|
mock_client.unban_user.side_effect = Exception("Error")
|
|
ok = await unban_user(mock_client, "!room:example.org", "@user:example.org")
|
|
assert ok is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Room state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSetRoomName:
|
|
@pytest.mark.asyncio
|
|
async def test_set_name_success(self, mock_client) -> None:
|
|
ok = await set_room_name(mock_client, "!room:example.org", "New Name")
|
|
assert ok is True
|
|
mock_client.send_state_event.assert_called_once_with(
|
|
"!room:example.org", "m.room.name", {"name": "New Name"}
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_name_failure(self, mock_client) -> None:
|
|
mock_client.send_state_event.side_effect = Exception("No permission")
|
|
ok = await set_room_name(mock_client, "!room:example.org", "Name")
|
|
assert ok is False
|
|
|
|
|
|
class TestSetRoomTopic:
|
|
@pytest.mark.asyncio
|
|
async def test_set_topic_success(self, mock_client) -> None:
|
|
ok = await set_room_topic(mock_client, "!room:example.org", "New Topic")
|
|
assert ok is True
|
|
mock_client.send_state_event.assert_called_once_with(
|
|
"!room:example.org", "m.room.topic", {"topic": "New Topic"}
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_topic_failure(self, mock_client) -> None:
|
|
mock_client.send_state_event.side_effect = Exception("No permission")
|
|
ok = await set_room_topic(mock_client, "!room:example.org", "Topic")
|
|
assert ok is False
|