feat(2): mattermost-lts create-message round-trip (§4.3 P3) — first-user→login→team→channel→post→read-back; harness http.post_with_headers (returns response headers, for mattermost login Token)

runner/harness/http.py

@@ -132,6 +132,40 @@ def http_request(
         return 0, None
 
 
+def post_with_headers(
+    url: str,
+    data: dict | bytes | None = None,
+    headers: dict[str, str] | None = None,
+    content_type: str = "application/json",
+    timeout: int = 15,
+) -> tuple[int, object | None, dict[str, str]]:
+    """Like http_post but ALSO returns the response headers as a dict — for APIs that hand back an
+    auth token in a response header rather than the body (e.g. mattermost login → `Token` header).
+    Returns (status, parsed_json_or_None, response_headers). status=0 + {} on transport failure."""
+    if isinstance(data, (bytes, bytearray)):
+        body: bytes | None = bytes(data)
+    elif content_type == "application/json" and data is not None:
+        body = json.dumps(data).encode()
+    elif content_type == "application/x-www-form-urlencoded" and data is not None:
+        body = urllib.parse.urlencode(data).encode()
+    else:
+        body = None
+    req = urllib.request.Request(url, data=body, method="POST")
+    req.add_header("Content-Type", content_type)
+    for k, v in (headers or {}).items():
+        req.add_header(k, v)
+    try:
+        with urllib.request.urlopen(req, timeout=timeout, context=_CTX) as resp:
+            return resp.getcode(), _parse_body(resp.read()), dict(resp.headers)
+    except urllib.error.HTTPError as e:
+        try:
+            return e.code, _parse_body(e.read()), dict(e.headers or {})
+        except Exception:  # noqa: BLE001
+            return e.code, None, dict(getattr(e, "headers", {}) or {})
+    except Exception:  # noqa: BLE001
+        return 0, None, {}
+
+
 def assert_converges(
     fn,
     description: str,

tests/mattermost-lts/functional/test_create_message.py

@@ -0,0 +1,102 @@
+"""mattermost-lts — Q4.5 recipe-specific functional test (plan §4.3: "create the app's primary
+object — a message — and read it back").
+
+Exercises mattermost's core function end-to-end against the live per-run deploy, via the REST API:
+  1. Bootstrap the FIRST user (a fresh mattermost server lets the first user be created unauthenticated
+     and makes them system admin).
+  2. Log in → capture the session token from the `Token` response header.
+  3. Create a team, then an open channel in it.
+  4. POST a message (a unique marker) to the channel.
+  5. GET the post back by id and assert the message text round-trips intact.
+
+NOT health-only: a mattermost whose DB/API/posting path is broken fails here even though `/` and
+`/api/v4/system/ping` return 200. The marker is unique per run so a stale/echoed response can't pass.
+"""
+
+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
+
+# Password must satisfy mattermost's default policy (>=5 chars; use a clearly strong one).
+_ADMIN_PW = "Ccci-Test-Pw-2026!"
+
+
+def _bearer(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def test_create_message_roundtrip(live_app):
+    base = f"https://{live_app}/api/v4"
+    uniq = uuid.uuid4().hex[:10]
+
+    # 1) Bootstrap first user (system admin on a fresh server). username: lowercase alnum.
+    username = f"ccci{uniq}"
+    email = f"{username}@ccci.example.com"
+    status, user = harness_http.http_post(
+        f"{base}/users",
+        data={"email": email, "username": username, "password": _ADMIN_PW},
+        timeout=30,
+    )
+    assert status in (200, 201) and isinstance(user, dict) and user.get("id"), (
+        f"first-user creation failed: HTTP {status}, body={user!r}"
+    )
+
+    # 2) Login → token in the `Token` response header.
+    status, _, hdrs = harness_http.post_with_headers(
+        f"{base}/users/login",
+        data={"login_id": email, "password": _ADMIN_PW},
+        timeout=30,
+    )
+    assert status == 200, f"login failed: HTTP {status}"
+    token = hdrs.get("Token") or hdrs.get("token")
+    assert token, f"login returned no Token header; headers={list(hdrs.keys())}"
+    auth = _bearer(token)
+
+    # 3) Create a team, then an open channel in it.
+    status, team = harness_http.http_post(
+        f"{base}/teams",
+        data={"name": f"t{uniq}", "display_name": f"ccci {uniq}", "type": "O"},
+        headers=auth,
+        timeout=30,
+    )
+    assert status in (200, 201) and isinstance(team, dict) and team.get("id"), (
+        f"team creation failed: HTTP {status}, body={team!r}"
+    )
+    status, chan = harness_http.http_post(
+        f"{base}/channels",
+        data={
+            "team_id": team["id"],
+            "name": f"c{uniq}",
+            "display_name": f"chan {uniq}",
+            "type": "O",
+        },
+        headers=auth,
+        timeout=30,
+    )
+    assert status in (200, 201) and isinstance(chan, dict) and chan.get("id"), (
+        f"channel creation failed: HTTP {status}, body={chan!r}"
+    )
+
+    # 4) POST a unique marker message.
+    marker = f"ccci-marker-{uniq}-roundtrip"
+    status, post = harness_http.http_post(
+        f"{base}/posts",
+        data={"channel_id": chan["id"], "message": marker},
+        headers=auth,
+        timeout=30,
+    )
+    assert status in (200, 201) and isinstance(post, dict) and post.get("id"), (
+        f"post creation failed: HTTP {status}, body={post!r}"
+    )
+
+    # 5) Read it back by id and assert the message survived the round-trip.
+    status, got = harness_http.http_get(f"{base}/posts/{post['id']}", headers=auth, timeout=30)
+    assert status == 200 and isinstance(got, dict), f"read-back failed: HTTP {status}, body={got!r}"
+    assert got.get("message") == marker, (
+        f"message did not round-trip: sent {marker!r}, got {got.get('message')!r}"
+    )