diff --git a/examples/member.py b/examples/member.py new file mode 100644 index 0000000..ba97c5c --- /dev/null +++ b/examples/member.py @@ -0,0 +1,28 @@ +from matrix import Bot + +bot = Bot() + + +@bot.command() +async def profile(ctx): + member = ctx.member + + name = await member.get_display_name() + avatar = await member.get_avatar_url() + level = await member.get_room_power_level(ctx.room) + presence = await member.get_presence() + has_permission = await member.has_room_permission(ctx.room, "ban") + has_event_permission = await member.has_event_permission(ctx.room, "m.room.message") + + await ctx.reply( + f"Welcome {member.mention()}!\n" + f"Display name: {name}\n" + f"Avatar URL: {avatar}\n" + f"Power level: {level}\n" + f"Presence: {presence}\n" + f"Has permission to ban users: {has_permission}\n" + f"Has permission to send messages: {has_event_permission}" + ) + + +bot.start(config="config.yaml") diff --git a/matrix/context.py b/matrix/context.py index 185b5f0..ab4d3cb 100644 --- a/matrix/context.py +++ b/matrix/context.py @@ -7,6 +7,7 @@ from .message import Message from .room import Room from .types import File, Image +from .member import Member if TYPE_CHECKING: from .bot import Bot # pragma: no cover @@ -23,6 +24,7 @@ def __init__(self, bot: "Bot", room: Room, event: Event): self.bot = bot self.room = room self.event = event + self.member = Member(event.sender, bot.client) self.body: str = getattr(event, "body", "") self.sender: str = event.sender diff --git a/matrix/member.py b/matrix/member.py new file mode 100644 index 0000000..d698602 --- /dev/null +++ b/matrix/member.py @@ -0,0 +1,170 @@ +from matrix.room import Room +from nio import AsyncClient +from matrix.api import matrix_call +from dataclasses import dataclass +from typing import Any + + +@dataclass +class MemberProfile: + displayname: str | None + avatar_url: str | None + other_info: dict[Any, Any] + + +class Member: + def __init__(self, user_id: str, client: AsyncClient) -> None: + self._user_id: str = user_id + self._client: AsyncClient = client + + def __str__(self) -> str: + return self.mention() + + @property + def user_id(self) -> str: + return self._user_id + + def mention(self) -> str: + """Get a Markdown-formatted mention link (matrix.to pill) for this member. + + ## Example + + ```python + await ctx.reply(f"Welcome {member.mention()}!") + ``` + """ + return f"[{self._user_id}](https://matrix.to/#/{self._user_id})" + + async def get_profile(self) -> MemberProfile | None: + """Get the profile information for this member. + + ## Example + + ```python + profile = await member.get_profile() + for key, value in profile.other_info.items(): + print(f"{key}: {value}") + ``` + """ + profile = await matrix_call( + self._client.get_profile(self._user_id), + error_message=f"Failed to get profile for user {self._user_id}", + ) + if profile: + return MemberProfile( + displayname=profile.displayname, + avatar_url=profile.avatar_url, + other_info=profile.other_info, + ) + return None + + async def get_display_name(self) -> str | None: + """Get the display name for this member. + + ## Example + + ```python + display_name = await member.get_display_name() + print(f"Display name: {display_name}") + ``` + """ + display_name = await matrix_call( + self._client.get_displayname(self._user_id), + error_message=f"Failed to get display name for user {self._user_id}", + ) + return display_name.displayname if display_name else None + + async def get_avatar_url(self) -> str | None: + """Get the avatar URL for this member. + + ## Example + + ```python + avatar_url = await member.get_avatar_url() + print(f"Avatar URL: {avatar_url}") + ``` + """ + avatar = await matrix_call( + self._client.get_avatar(self._user_id), + error_message=f"Failed to get avatar for user {self._user_id}", + ) + return await self._client.mxc_to_http(avatar.avatar_url) if avatar else None + + async def get_presence(self) -> str | None: + """Get the presence status for this member. + + ## Example + + ```python + presence = await member.get_presence() + print(f"Presence status: {presence}") + ``` + """ + presence = await matrix_call( + self._client.get_presence(self._user_id), + error_message=f"Failed to get presence for user {self._user_id}", + ) + return presence.presence if presence else None + + async def get_room_power_level(self, room: Room) -> int: + """Get the power level for this member in a specific room. + + ## Example + + ```python + level = await member.get_room_power_level(ctx.room) + print(f"Power level in room {room.room_id}: {level}") + ``` + """ + power_level = await room.get_state_event("m.room.power_levels", "") + if power_level is None: + return 0 + + content = power_level.content + users = content.get("users", {}) + default = content.get("users_default", 0) + + return int(users.get(self._user_id, default)) + + async def has_room_permission(self, room: Room, permission: str) -> bool: + """Check if this member has a specific permission in a room. + + ## Example + + ```python + has_permission = await member.has_room_permission(ctx.room, "ban") + print(f"Has ban permission: {has_permission}") + ``` + """ + power_level = await self.get_room_power_level(room) + power_levels_event = await room.get_state_event("m.room.power_levels", "") + if power_levels_event is None: + return False + + content = power_levels_event.content + required_level = content.get(permission) + if required_level is None: + return False # Permission not defined in the room's power levels + + return bool(power_level >= required_level) + + async def has_event_permission(self, room: Room, event_type: str) -> bool: + """Check if this member has permission to send a specific event type in a room. + + ## Example + + ```python + can_send_message = await member.has_event_permission(ctx.room, "m.room.message") + print(f"Can send messages: {can_send_message}") + ``` + """ + power_level = await self.get_room_power_level(room) + power_levels_event = await room.get_state_event("m.room.power_levels", "") + if power_levels_event is None: + return False + + content = power_levels_event.content + events = content.get("events", {}) + required_level = events.get(event_type, 0) + + return bool(power_level >= required_level) diff --git a/matrix/room.py b/matrix/room.py index 1d1fc0c..5587211 100644 --- a/matrix/room.py +++ b/matrix/room.py @@ -1,6 +1,12 @@ -from typing import Any - -from nio import AsyncClient, MatrixRoom, Event +from typing import Any, Union + +from nio import ( + AsyncClient, + MatrixRoom, + Event, + RoomGetStateEventResponse, + RoomGetStateEventError, +) from matrix.api import matrix_call from matrix.message import Message @@ -97,6 +103,14 @@ def encrypted(self) -> bool: """Whether the room is encrypted.""" return self._matrix_room.encrypted # type: ignore[no-any-return] + async def get_state_event( + self, event_type: str, state_key: str = "" + ) -> Union[RoomGetStateEventResponse, RoomGetStateEventError]: + return await matrix_call( + self._client.room_get_state_event(self.room_id, event_type, state_key), + error_message=f"Failed to get state event for room {self._matrix_room.room_id}", + ) + async def send( self, content: str | None = None, diff --git a/tests/test_member.py b/tests/test_member.py new file mode 100644 index 0000000..6927f76 --- /dev/null +++ b/tests/test_member.py @@ -0,0 +1,309 @@ +import pytest +from unittest.mock import AsyncMock, Mock + +from nio import ( + AsyncClient, + PresenceGetResponse, + ProfileGetAvatarResponse, + ProfileGetDisplayNameResponse, + ProfileGetResponse, + RoomGetStateEventResponse, +) +from matrix.member import Member, MemberProfile +from matrix.room import Room + + +@pytest.fixture +def client(): + return AsyncMock(spec=AsyncClient) + + +@pytest.fixture +def room(): + room = Mock(spec=Room) + room.room_id = "!room:example.com" + return room + + +@pytest.fixture +def member(client): + return Member("@user:matrix.org", client) + + +def power_levels_response(content): + return RoomGetStateEventResponse( + content=content, + event_type="m.room.power_levels", + state_key="", + room_id="!room:example.com", + ) + + +def test_member_user_id__expect_configured_user_id(member): + assert member.user_id == "@user:matrix.org" + + +def test_member_mention__expect_markdown_matrix_to_link(member): + assert ( + member.mention() == "[@user:matrix.org](https://matrix.to/#/@user:matrix.org)" + ) + + +def test_member_str__expect_same_value_as_mention(member): + assert str(member) == member.mention() + + +@pytest.mark.asyncio +async def test_get_profile__with_profile_response__expect_member_profile( + member, client +): + response = ProfileGetResponse( + displayname="Alice", + avatar_url="mxc://matrix.org/avatar", + other_info={"chat.commet.profile_status": "Online"}, + ) + client.get_profile = AsyncMock(return_value=response) + + result = await member.get_profile() + + client.get_profile.assert_awaited_once_with("@user:matrix.org") + assert result == MemberProfile( + displayname="Alice", + avatar_url="mxc://matrix.org/avatar", + other_info={"chat.commet.profile_status": "Online"}, + ) + + +@pytest.mark.asyncio +async def test_get_profile__when_client_returns_none__expect_none(member, client): + client.get_profile = AsyncMock(return_value=None) + + result = await member.get_profile() + + assert result is None + + +@pytest.mark.asyncio +async def test_get_display_name__with_display_name__expect_value(member, client): + response = ProfileGetDisplayNameResponse(displayname="Alice") + client.get_displayname = AsyncMock(return_value=response) + + result = await member.get_display_name() + + client.get_displayname.assert_awaited_once_with("@user:matrix.org") + assert result == "Alice" + + +@pytest.mark.asyncio +async def test_get_display_name__without_display_name__expect_none(member, client): + client.get_displayname = AsyncMock(return_value=ProfileGetDisplayNameResponse()) + + result = await member.get_display_name() + + assert result is None + + +@pytest.mark.asyncio +async def test_get_display_name__when_client_returns_none__expect_none(member, client): + client.get_displayname = AsyncMock(return_value=None) + + result = await member.get_display_name() + + assert result is None + + +@pytest.mark.asyncio +async def test_get_avatar_url__with_mxc_avatar__expect_http_url(member, client): + response = ProfileGetAvatarResponse(avatar_url="mxc://matrix.org/avatar") + client.get_avatar = AsyncMock(return_value=response) + client.mxc_to_http = AsyncMock(return_value="https://matrix.org/avatar") + + result = await member.get_avatar_url() + + client.get_avatar.assert_awaited_once_with("@user:matrix.org") + client.mxc_to_http.assert_awaited_once_with("mxc://matrix.org/avatar") + assert result == "https://matrix.org/avatar" + + +@pytest.mark.asyncio +async def test_get_avatar_url__when_client_returns_none__expect_none(member, client): + client.get_avatar = AsyncMock(return_value=None) + + result = await member.get_avatar_url() + + client.mxc_to_http.assert_not_awaited() + assert result is None + + +@pytest.mark.asyncio +async def test_get_room_power_level__expect_user_power_level(member, room): + response = power_levels_response( + {"users": {"@user:matrix.org": 50}, "users_default": 0} + ) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.get_room_power_level(room) + + room.get_state_event.assert_awaited_once_with("m.room.power_levels", "") + assert result == 50 + + +@pytest.mark.asyncio +async def test_get_room_power_level__when_user_not_listed__expect_users_default( + member, room +): + response = power_levels_response({"users": {}, "users_default": 10}) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.get_room_power_level(room) + + assert result == 10 + + +@pytest.mark.asyncio +async def test_get_room_power_level__when_content_empty__expect_zero(member, room): + response = power_levels_response({}) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.get_room_power_level(room) + + assert result == 0 + + +@pytest.mark.asyncio +async def test_get_room_power_level__when_event_is_none__expect_zero(member, room): + room.get_state_event = AsyncMock(return_value=None) + + result = await member.get_room_power_level(room) + + assert result == 0 + + +@pytest.mark.asyncio +async def test_get_presence__with_online_presence__expect_online(member, client): + response = PresenceGetResponse( + user_id="@user:matrix.org", + presence="online", + last_active_ago=None, + currently_active=True, + status_msg=None, + ) + client.get_presence = AsyncMock(return_value=response) + + result = await member.get_presence() + + client.get_presence.assert_awaited_once_with("@user:matrix.org") + assert result == "online" + + +@pytest.mark.asyncio +async def test_get_presence__when_client_returns_none__expect_none(member, client): + client.get_presence = AsyncMock(return_value=None) + + result = await member.get_presence() + + assert result is None + + +@pytest.mark.asyncio +async def test_has_room_permission__with_user_level_meeting_requirement__expect_true( + member, room +): + response = power_levels_response( + { + "users": {"@user:matrix.org": 50}, + "users_default": 0, + "ban": 50, + } + ) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.has_room_permission(room, "ban") + + assert room.get_state_event.await_count == 2 + assert result is True + + +@pytest.mark.asyncio +async def test_has_room_permission__with_user_level_too_low__expect_false(member, room): + response = power_levels_response( + { + "users": {"@user:matrix.org": 10}, + "users_default": 0, + "ban": 50, + } + ) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.has_room_permission(room, "ban") + + assert result is False + + +@pytest.mark.asyncio +async def test_has_room_permission__when_permission_not_defined__expect_false( + member, room +): + response = power_levels_response( + {"users": {"@user:matrix.org": 100}, "users_default": 0} + ) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.has_room_permission(room, "ban") + + assert result is False + + +@pytest.mark.asyncio +async def test_has_event_permission__with_user_level_meeting_requirement__expect_true( + member, room +): + response = power_levels_response( + { + "users": {"@user:matrix.org": 50}, + "users_default": 0, + "events": {"m.room.message": 0}, + } + ) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.has_event_permission(room, "m.room.message") + + assert room.get_state_event.await_count == 2 + assert result is True + + +@pytest.mark.asyncio +async def test_has_event_permission__with_user_level_too_low__expect_false( + member, room +): + response = power_levels_response( + { + "users": {"@user:matrix.org": 10}, + "users_default": 0, + "events": {"m.room.power_levels": 100}, + } + ) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.has_event_permission(room, "m.room.power_levels") + + assert result is False + + +@pytest.mark.asyncio +async def test_has_event_permission__when_event_level_not_defined__expect_true( + member, room +): + response = power_levels_response( + { + "users": {"@user:matrix.org": 0}, + "users_default": 0, + "events": {}, + } + ) + room.get_state_event = AsyncMock(return_value=response) + + result = await member.has_event_permission(room, "m.room.message") + + assert result is True