checkbot/tests/test_matrix.py

import json

import httpx
import pytest

from app.chat import MatrixAdapter, MatrixError, MessageRef
from app.chat.matrix import MAX_BODY_BYTES, _truncate_bytes
from app.chat.matrix_format import to_matrix

TOKEN = "syt_SECRET-token.abc"
USER = "@bot:example.de"
ROOM = "!abc:example.de"
BASE = "https://matrix.example.de"


def make(handler, room=ROOM, **kwargs):
    return MatrixAdapter(BASE, USER, TOKEN, room, transport=httpx.MockTransport(handler), **kwargs)


def sent_event(event_id="$ev1"):
    return httpx.Response(200, json={"event_id": event_id})


async def test_send_payload_url_and_ref():
    seen = {}

    def handler(request):
        seen["method"] = request.method
        seen["raw_path"] = request.url.raw_path.decode()
        seen["auth"] = request.headers["authorization"]
        seen["body"] = json.loads(request.content)
        return sent_event()

    ref = await make(handler).send(ROOM, "<b>h1</b>\nDisk: a &lt;b&gt; &amp; c")
    assert seen["method"] == "PUT"
    assert seen["raw_path"].startswith("/_matrix/client/v3/rooms/%21abc%3Aexample.de/send/m.room.message/")
    assert len(seen["raw_path"].rsplit("/", 1)[1]) >= 16
    assert seen["auth"] == f"Bearer {TOKEN}"
    body = seen["body"]
    assert body["msgtype"] == "m.text" and body["format"] == "org.matrix.custom.html"
    assert body["body"] == "h1\nDisk: a <b> & c"
    assert body["formatted_body"] == "<b>h1</b><br>Disk: a &lt;b&gt; &amp; c"
    assert ref == MessageRef("matrix", ROOM, "$ev1")


async def test_each_send_uses_new_txn_id():
    paths = []

    def handler(request):
        paths.append(request.url.path)
        return sent_event()

    adapter = make(handler)
    await adapter.send(ROOM, "a")
    await adapter.send(ROOM, "a")
    assert paths[0] != paths[1]


async def test_edit_uses_replace_relation():
    seen = {}

    def handler(request):
        seen["body"] = json.loads(request.content)
        return sent_event("$ev2")

    await make(handler).edit(MessageRef("matrix", ROOM, "$ev1"), "<b>x</b>\nAnalyse")
    body = seen["body"]
    assert body["body"].startswith("* ") and body["formatted_body"].startswith("* ")
    assert body["m.relates_to"] == {"rel_type": "m.replace", "event_id": "$ev1"}
    assert body["m.new_content"]["body"] == "x\nAnalyse"
    assert body["m.new_content"]["formatted_body"] == "<b>x</b><br>Analyse"


async def test_429_retry_after_respected_once():
    calls, slept = [], []

    def handler(request):
        calls.append(request.url.path)
        if len(calls) == 1:
            return httpx.Response(429, json={"errcode": "M_LIMIT_EXCEEDED", "retry_after_ms": 2500})
        return sent_event()

    async def sleep(seconds):
        slept.append(seconds)

    ref = await make(handler, sleep=sleep).send(ROOM, "x")
    assert ref.message_id == "$ev1" and slept == [2.5]
    assert calls[0] == calls[1]


async def test_429_twice_raises():
    async def sleep(seconds):
        pass

    def handler(request):
        return httpx.Response(429, json={"retry_after_ms": 10})

    with pytest.raises(MatrixError):
        await make(handler, sleep=sleep).send(ROOM, "x")


async def test_check_ok():
    def handler(request):
        if request.url.path.endswith("/account/whoami"):
            return httpx.Response(200, json={"user_id": USER})
        return httpx.Response(200, json={"joined_rooms": [ROOM]})

    assert "Raum beigetreten" in await make(handler).check()


async def test_check_wrong_user():
    def handler(request):
        return httpx.Response(200, json={"user_id": "@other:example.de"})

    with pytest.raises(MatrixError, match="@other:example.de"):
        await make(handler).check()


async def test_check_not_in_room():
    def handler(request):
        if request.url.path.endswith("/account/whoami"):
            return httpx.Response(200, json={"user_id": USER})
        return httpx.Response(200, json={"joined_rooms": ["!other:example.de"]})

    with pytest.raises(MatrixError, match="nicht beigetreten"):
        await make(handler).check()


async def test_errors_never_contain_token():
    def http_error(request):
        return httpx.Response(401, json={"errcode": "M_UNKNOWN_TOKEN", "error": f"bad {TOKEN}"})

    def network_error(request):
        raise httpx.ConnectError(f"boom {TOKEN}")

    for handler in (http_error, network_error):
        with pytest.raises(MatrixError) as info:
            await make(handler).send(ROOM, "x")
        assert TOKEN not in str(info.value)
        assert info.value.__cause__ is None and info.value.__suppress_context__


async def test_missing_event_id_raises():
    with pytest.raises(MatrixError):
        await make(lambda request: httpx.Response(200, json={})).send(ROOM, "x")


def test_to_matrix_escapes_foreign_markup():
    body, formatted = to_matrix("<b>x</b> &lt;script&gt;")
    assert body == "x <script>"
    assert formatted == "<b>x</b> &lt;script&gt;"


def test_truncate_bytes_keeps_short_text():
    assert _truncate_bytes("hallo") == "hallo"


def test_truncate_bytes_cuts_at_utf8_boundary_and_marks():
    text = "ä" * 20000
    truncated = _truncate_bytes(text, limit=100)
    assert len(truncated.encode("utf-8")) <= 100
    assert truncated.endswith("…")
    truncated.encode("utf-8").decode("utf-8")


async def test_send_truncates_body_and_formatted_body():
    seen = {}

    def handler(request):
        seen["body"] = json.loads(request.content)
        return sent_event()

    long_text = "A" * (MAX_BODY_BYTES + 5000)
    await make(handler).send(ROOM, long_text)
    body = seen["body"]
    assert len(body["body"].encode("utf-8")) <= MAX_BODY_BYTES
    assert len(body["formatted_body"].encode("utf-8")) <= MAX_BODY_BYTES
    assert body["body"].endswith("…")


async def test_edit_truncates_new_content_too():
    seen = {}

    def handler(request):
        seen["body"] = json.loads(request.content)
        return sent_event("$ev2")

    long_text = "B" * (MAX_BODY_BYTES + 5000)
    await make(handler).edit(MessageRef("matrix", ROOM, "$ev1"), long_text)
    new_content = seen["body"]["m.new_content"]
    assert len(new_content["body"].encode("utf-8")) <= MAX_BODY_BYTES
    assert len(new_content["formatted_body"].encode("utf-8")) <= MAX_BODY_BYTES