From fc6e35d617b1f3ad9dbe30fbac4d642576f75a90 Mon Sep 17 00:00:00 2001 From: autonomic-bot Date: Fri, 29 May 2026 08:31:37 +0100 Subject: [PATCH] =?UTF-8?q?feat(2):=20mattermost-lts=20create-message=20ro?= =?UTF-8?q?und-trip=20(=C2=A74.3=20P3)=20=E2=80=94=20first-user=E2=86=92lo?= =?UTF-8?q?gin=E2=86=92team=E2=86=92channel=E2=86=92post=E2=86=92read-back?= =?UTF-8?q?;=20harness=20http.post=5Fwith=5Fheaders=20(returns=20response?= =?UTF-8?q?=20headers,=20for=20mattermost=20login=20Token)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- runner/harness/http.py | 34 ++++++ .../functional/test_create_message.py | 102 ++++++++++++++++++ 2 files changed, 136 insertions(+) create mode 100644 tests/mattermost-lts/functional/test_create_message.py diff --git a/runner/harness/http.py b/runner/harness/http.py index e07eb4e..002d9b6 100644 --- a/runner/harness/http.py +++ b/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, diff --git a/tests/mattermost-lts/functional/test_create_message.py b/tests/mattermost-lts/functional/test_create_message.py new file mode 100644 index 0000000..9174964 --- /dev/null +++ b/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}" + )