Add pre-defined check can_redact

matrix/checks.py

@@ -120,3 +120,34 @@ def wrapper(cmd: "Command") -> "Command":
         return cmd
 
     return wrapper
+
+
+def can_redact() -> Callable:
+    """
+    Decorator to restrict a command to users allowed to redact other
+    users' events (power level >= the room's `redact` power level).
+
+    ## Example
+
+    ```python
+    @can_redact()
+    @bot.command("purge")
+    async def purge(ctx: Context, event_id: str) -> None:
+        message = await ctx.room.fetch_message(event_id)
+        await message.delete()
+
+    @purge.error(CheckError)
+    async def purge_error(ctx: Context, error: CheckError) -> None:
+        await ctx.reply("You are not allowed to redact messages in this room.")
+    ```
+    """
+
+    async def _can_redact(ctx: "Context") -> bool:
+        levels = ctx.room.power_levels
+        return levels.can_user_redact(ctx.sender)  # type: ignore[no-any-return]
+
+    def wrapper(cmd: "Command") -> "Command":
+        cmd.check(_can_redact)
+        return cmd
+
+    return wrapper

tests/test_checks.py

@@ -1,11 +1,12 @@
 import pytest
 
 from unittest.mock import AsyncMock, Mock
-from nio import MatrixRoom, RoomMessageText, AsyncClient, PowerLevels
+from nio import MatrixRoom, RoomMessageText, AsyncClient, PowerLevels, DefaultLevels
 
 from matrix.checks import (
     ADMIN_POWER_LEVEL,
     MODERATOR_POWER_LEVEL,
+    can_redact,
     is_admin,
     is_moderator,
     is_room_encrypted,
@@ -44,12 +45,19 @@ def make_event(sender: str) -> RoomMessageText:
 
 
 def make_context(
-    bot, client, sender: str, level: int, encrypted: bool = False
+    bot,
+    client,
+    sender: str,
+    level: int,
+    encrypted: bool = False,
+    defaults: DefaultLevels | None = None,
 ) -> Context:
     """Builds a real Context whose room reports `level` as the sender's power level."""
     matrix_room = Mock(spec=MatrixRoom)
     matrix_room.room_id = "!room:example.com"
-    matrix_room.power_levels = PowerLevels(users={sender: level})
+    matrix_room.power_levels = PowerLevels(
+        defaults=defaults or DefaultLevels(), users={sender: level}
+    )
     matrix_room.encrypted = encrypted
 
     room = Room(matrix_room, client)
@@ -252,6 +260,106 @@ async def restricted(ctx):
     assert called is True
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "level, expected",
+    [
+        (50, True),  # nio's default `redact` power level is 50
+        (100, True),
+        (49, False),
+        (0, False),
+    ],
+)
+async def test_can_redact_check__respects_redact_power_level(
+    bot, client, level, expected
+):
+    async def my_command(ctx):
+        pass
+
+    cmd = Command(my_command)
+    can_redact()(cmd)
+
+    check = cmd.checks[-1]
+    ctx = make_context(bot, client, "@user:example.com", level)
+
+    assert await check(ctx) is expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("level, expected", [(74, False), (75, True)])
+async def test_can_redact_check__respects_custom_redact_level(
+    bot, client, level, expected
+):
+    """The check follows the room's configured `redact` power level."""
+
+    async def my_command(ctx):
+        pass
+
+    cmd = Command(my_command)
+    can_redact()(cmd)
+
+    check = cmd.checks[-1]
+    ctx = make_context(
+        bot,
+        client,
+        "@user:example.com",
+        level,
+        defaults=DefaultLevels(redact=75),
+    )
+
+    assert await check(ctx) is expected
+
+
+def test_can_redact__returns_the_same_command():
+    async def my_command(ctx):
+        pass
+
+    cmd = Command(my_command)
+    assert can_redact()(cmd) is cmd
+
+
+@pytest.mark.asyncio
+async def test_can_redact__raises_check_error_when_cannot_redact(bot, client):
+    called = False
+
+    async def restricted(ctx):
+        nonlocal called
+        called = True
+
+    cmd = Command(restricted)
+    can_redact()(cmd)
+
+    caught: list[Exception] = []
+
+    @cmd.error(CheckError)
+    async def on_check_error(ctx, error):
+        caught.append(error)
+
+    ctx = make_context(bot, client, "@user:example.com", 0)
+    await cmd(ctx)
+
+    assert called is False
+    assert len(caught) == 1
+    assert isinstance(caught[0], CheckError)
+
+
+@pytest.mark.asyncio
+async def test_can_redact__allows_command_when_can_redact(bot, client):
+    called = False
+
+    async def restricted(ctx):
+        nonlocal called
+        called = True
+
+    cmd = Command(restricted)
+    can_redact()(cmd)
+
+    ctx = make_context(bot, client, "@user:example.com", 50)
+    await cmd(ctx)
+
+    assert called is True
+
+
 def test_power_level_constants__match_matrix_defaults():
     """The thresholds follow the defaults of `m.room.power_levels`."""
     assert ADMIN_POWER_LEVEL == 100