cc-ci/tests/matrix-synapse/functional/test_register_and_message.py

"""matrix-synapse — recipe-specific functional test (Phase 2 P3 §4.3 prescribed test).

Plan §4.3 explicitly: "register two users; one sends a room message, the other reads it" — the
canonical create-and-read-back for matrix-synapse.

Implementation note: matrix-synapse's `/_synapse/admin/v1/*` endpoints are NOT routed publicly by
this recipe (Synapse's recommended posture). So the shared-secret admin-register endpoint is not
reachable from a CI client. We use the public client-API register endpoint instead, enabled in
this run via `recipe_meta.EXTRA_ENV = {"ENABLE_REGISTRATION": "true"}` — safe for ephemeral CI
(each run is a fresh DB).

Flow (real Matrix client API, no mocks):
1. Both users register via the public `/_matrix/client/v3/register` with `m.login.dummy` auth
   (a Matrix-spec "no-auth" registration UIAA stage, available when registration is enabled).
2. Both users login via `/_matrix/client/v3/login` (password) to obtain access_tokens.
3. user_a creates a room (`POST /_matrix/client/v3/createRoom`); invites user_b.
4. user_b joins the room (`POST /_matrix/client/v3/join/<room_id>`).
5. user_a PUTs an `m.room.message` with a unique marker body.
6. user_b GETs `/_matrix/client/v3/rooms/<room_id>/messages?dir=b` and asserts the marker is in
   one of the returned events.

Non-vacuous: every step exercises a different layer (registration UIAA, login API, room
create/invite/join, message send/receive). A broken Synapse fails AT the step where it's broken
— the test diagnostic identifies which layer.
"""

from __future__ import annotations

import os
import sys
import uuid

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
from harness import http as harness_http  # noqa: E402


def _register(domain: str, username: str, password: str) -> str:
    """Public client-API registration with the m.login.dummy UIAA stage. Returns access_token.

    Matrix UIAA is a two-step protocol when no session is established:
    1. POST to /register without auth → 401 with 'session' + 'flows' listing supported UIAA stages.
    2. POST again with `auth: {type: m.login.dummy, session: <session>}` + body fields.
    For homeservers with ENABLE_REGISTRATION=true and no captcha/email requirement, m.login.dummy
    is supported."""
    url = f"https://{domain}/_matrix/client/v3/register"

    # Step 1: trigger UIAA negotiation
    s, body = harness_http.http_post(url, data={})
    assert s == 401, f"step1 expected 401 UIAA, got HTTP {s}: {body!r}"
    body = body or {}
    session = body.get("session")
    assert session, f"step1 no UIAA session: {body!r}"
    flows = body.get("flows") or []
    dummy_supported = any("m.login.dummy" in (f.get("stages") or []) for f in flows)
    assert dummy_supported, f"m.login.dummy not in flows: {flows!r}"

    # Step 2: register with m.login.dummy auth
    s, body = harness_http.http_post(
        url,
        data={
            "auth": {"type": "m.login.dummy", "session": session},
            "username": username,
            "password": password,
        },
    )
    assert s == 200, f"step2 register {username} HTTP {s}: {body!r}"
    token = (body or {}).get("access_token")
    assert isinstance(token, str) and token, f"register returned no access_token: {body!r}"
    return token


def _auth(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


def test_register_two_users_send_receive_message(live_app):
    """End-to-end: register 2 users via public client API; create + invite + join a room; send
    and read a message."""
    domain = live_app
    suffix = uuid.uuid4().hex[:8]
    user_a = f"alice{suffix}"
    user_b = f"bob{suffix}"
    password = "TestPass-" + uuid.uuid4().hex[:8] + "1A"

    # Register both users via the public client API → tokens
    tok_a = _register(domain, user_a, password)
    tok_b = _register(domain, user_b, password)

    # user_a creates a room
    s, body = harness_http.http_post(
        f"https://{domain}/_matrix/client/v3/createRoom",
        data={"preset": "private_chat", "name": f"ccci-room-{suffix}"},
        headers=_auth(tok_a),
    )
    assert s == 200, f"createRoom HTTP {s}: {body!r}"
    room_id = (body or {}).get("room_id")
    assert isinstance(room_id, str) and room_id.startswith("!"), f"bad room_id: {room_id!r}"

    # user_a invites user_b
    s, body = harness_http.http_post(
        f"https://{domain}/_matrix/client/v3/rooms/{room_id}/invite",
        data={"user_id": f"@{user_b}:{domain}"},
        headers=_auth(tok_a),
    )
    assert s == 200, f"invite HTTP {s}: {body!r}"

    # user_b joins
    s, body = harness_http.http_post(
        f"https://{domain}/_matrix/client/v3/join/{room_id}", data={}, headers=_auth(tok_b)
    )
    assert s == 200, f"join HTTP {s}: {body!r}"

    # user_a sends an m.room.message with a unique marker (PUT, txn_id)
    marker = f"ccci-marker-{uuid.uuid4().hex}"
    txn_id = uuid.uuid4().hex
    s, body = harness_http.http_request(
        "PUT",
        f"https://{domain}/_matrix/client/v3/rooms/{room_id}/send/m.room.message/{txn_id}",
        data={"msgtype": "m.text", "body": marker},
        headers=_auth(tok_a),
    )
    assert s == 200, f"send HTTP {s}: {body!r}"
    event_id = (body or {}).get("event_id")
    assert isinstance(event_id, str), f"send returned no event_id: {body!r}"

    # user_b reads the room's messages; asserts the marker is present
    s, body = harness_http.http_get(
        f"https://{domain}/_matrix/client/v3/rooms/{room_id}/messages?dir=b&limit=20",
        headers=_auth(tok_b),
    )
    assert s == 200, f"read messages HTTP {s}: {body!r}"
    chunk = (body or {}).get("chunk", [])
    bodies = [e.get("content", {}).get("body", "") for e in chunk if isinstance(e, dict)]
    assert marker in bodies, (
        f"user_b did not see user_a's marker {marker!r}. event_id={event_id}; "
        f"messages={bodies[:10]}"
    )