app-catalog-felhom.eu/scripts/upgrade_fixtures_box.py

# PORTED 2026-09-23 (night shift, R-462) VERBATIM from felhom.eu/documentation/audits/update-night-2026-09-21/
# fixtures.py (as carried forward in night-2026-09-23/). The box walk and the test bench now read the
# SAME seed/verify code; upgrade_boxport.py adapts it to the bench. Edit here, not in the audit copy.
#!/usr/bin/env python3
"""Box-side seed/verify fixtures for walk.py, guest 9202.

THE ONE RULE (R-156), carried verbatim from `app-catalog-felhom.eu/scripts/upgrade_fixtures.py`:
*nothing is ever seeded into a volume by hand.* Every seed here goes in through the app's OWN
interface — its HTTP API through the household's real front door (traefik, `Host: <sub>.<domain>`),
or its own CLI running inside its own container. A raw SQL INSERT or a planted file is never used.

If an app has no non-browser route, its fixture returns None and the edge is recorded
`inconclusive — no non-browser seed route`, WITH WHAT WAS TRIED. That is a result, not a gap.

Each fixture:
    seed(w, sub, say)   -> an opaque token, or None
    verify(w, sub, tok, say) -> True / False
verify() must ask the APP, never the filesystem: a migration is supposed to rewrite files.
Where a fixture can prove itself (a negative control that must read as absent) it does so on EVERY
call, so a readback that has broken into always saying "found" fails instead of passing everything.
"""
import base64, json, re, secrets, time


def _gx(w, container, *cmd, timeout=240):
    """Run a command inside the app's OWN container on 9202 (its own CLI, not our SQL)."""
    import shlex
    line = " ".join(shlex.quote(c) for c in cmd)
    return w.guest(f"docker exec {container} {line} 2>&1", timeout=timeout)


# =============================================================================================
class PrivateBin:
    """PrivateBin's own JSON API. A paste is a POST and reading it back is a GET — an
    application-level round trip. File-backed, no database: this single seed IS the file half."""
    sub = "paste"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200",)):
            return None
        marker = "upg-" + secrets.token_hex(8)
        ct = base64.b64encode(marker.encode()).decode()
        body = json.dumps({
            "v": 2,
            "adata": [[base64.b64encode(secrets.token_bytes(16)).decode(),
                       base64.b64encode(secrets.token_bytes(8)).decode(),
                       100000, 256, 128, "aes", "gcm", "none"], "plaintext", 0, 0],
            "ct": ct, "meta": {"expire": "never"}})
        rc, code, out = w.app_curl(sub, "/", "-H", "X-Requested-With: JSONHttpRequest",
                                   "-H", "Content-Type: application/json",
                                   data=body, method="POST")
        try:
            j = json.loads(out)
        except Exception:
            say(f"    privatebin: POST returned non-JSON (http {code}): {out[:200]}")
            return None
        if j.get("status") != 0 or not j.get("id"):
            say(f"    privatebin: POST refused: {out[:250]}")
            return None
        say(f"    privatebin: seeded paste id={j['id']}")
        return {"id": j["id"], "marker": ct}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200",), tries=36):
            return False
        # negative control, every call: a paste id that cannot exist must NOT read back
        rc, code, out = w.app_curl(sub, "/?pasteid=" + secrets.token_hex(8),
                                   "-H", "X-Requested-With: JSONHttpRequest")
        if t["marker"] in out:
            say("    privatebin: READBACK UNUSABLE — a paste id that cannot exist returned the marker")
            return False
        rc, code, out = w.app_curl(sub, "/?pasteid=" + t["id"],
                                   "-H", "X-Requested-With: JSONHttpRequest")
        got = code == "200" and t["marker"] in out
        say(f"    privatebin: readback http={code} marker_present={got}")
        return got


# =============================================================================================
class Docmost:
    """Docmost's own REST API: create the first workspace+user, then prove the account survives by
    asking the app to AUTHENTICATE it. Login is version-stable across the API churn."""
    sub = "docs"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302", "404")):
            return None
        email = f"drill-{secrets.token_hex(4)}@gate.invalid"
        pw = "Drill-" + secrets.token_hex(10)
        body = json.dumps({"workspaceName": "drill", "name": "drill", "email": email, "password": pw})
        rc, code, out = w.app_curl(sub, "/api/auth/setup", "-H", "Content-Type: application/json",
                                   data=body, method="POST")
        say(f"    docmost: /api/auth/setup http={code} rc={rc}")
        if code not in ("200", "201"):
            say(f"    docmost: setup refused: {out[:250]}")
            return None
        return {"email": email, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302", "404"), tries=36):
            return False
        # negative control: a password that was never set must NOT authenticate
        bad = json.dumps({"email": t["email"], "password": "definitely-" + secrets.token_hex(8)})
        rc, code, _ = w.app_curl(sub, "/api/auth/login", "-H", "Content-Type: application/json",
                                 data=bad, method="POST")
        if code in ("200", "201"):
            say("    docmost: READBACK UNUSABLE — a wrong password authenticated")
            return False
        body = json.dumps({"email": t["email"], "password": t["pw"]})
        rc, code, out = w.app_curl(sub, "/api/auth/login", "-H", "Content-Type: application/json",
                                   data=body, method="POST")
        ok = code in ("200", "201")
        say(f"    docmost: login as the seeded user http={code} ok={ok}")
        if not ok:
            say(f"    docmost: login body {out[:200]}")
        return ok


# =============================================================================================
class BookStack:
    """BookStack mints no API token without a browser, so BOTH halves go through `php artisan` —
    BookStack's OWN CLI, inside its own container, against its own User model.

    The exit code carries no information here (`bookstack:reset-mfa` exits 1 for a user it FOUND
    and for one it did not), so the discriminator is the OUTPUT: the positive sentence required and
    the not-found sentence required absent. The negative control runs on every verify.

    LIMITATION (R-460): this seeds the DATABASE half only. The FILE half needs the API token the
    app cannot mint headlessly — so a bookstack edge is at best HALF-proven here.
    """
    sub = "wiki"

    def _artisan(self, w, *args):
        for path in ("/app/www/artisan", "/var/www/html/artisan"):
            out = _gx(w, "bookstack", "php", path, *args)
            if "Could not open input file" not in out:
                return " ".join(out.split())
        return " ".join(out.split())

    def _lookup(self, w, email):
        out = self._artisan(w, "bookstack:reset-mfa", f"--email={email}")
        found = f"Email: {email}" in out
        missing = "could not be found" in out
        if found == missing:
            return None, out
        return found, out

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/login", want=("200",), tries=72):
            return None
        email = f"drill-{secrets.token_hex(4)}@gate.invalid"
        pw = "Drill-" + secrets.token_hex(10)
        out = self._artisan(w, "bookstack:create-admin", f"--email={email}",
                            f"--name=drill-{secrets.token_hex(3)}", f"--password={pw}")
        say(f"    bookstack: artisan create-admin :: {out[:140]}")
        if "successfully created" not in out:
            return None
        return {"email": email, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/login", want=("200",), tries=72):
            say("    bookstack: the app never served /login")
            return False
        absent, _ = self._lookup(w, f"nobody-{secrets.token_hex(6)}@gate.invalid")
        if absent is not False:
            say(f"    bookstack: READBACK UNUSABLE — an email that cannot exist did not read absent ({absent})")
            return False
        found, out = self._lookup(w, t["email"])
        say(f"    bookstack: readback of the seeded account found={found} :: {out[:140]}")
        return found is True


# =============================================================================================
class Gitea:
    """Gitea's own admin CLI creates the first user; its own REST API (basic auth) then creates a
    repository and reads it back. Both are the app's own interfaces."""
    sub = "git"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302")):
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        out = _gx(w, "gitea", "su", "git", "-c",
                  f"gitea admin user create --username {user} --password {pw} "
                  f"--email {user}@gate.invalid --admin --must-change-password=false")
        say(f"    gitea: admin user create :: {' '.join(out.split())[:140]}")
        if "has been successfully created" not in out and "successfully created" not in out:
            return None
        repo = "drillrepo" + secrets.token_hex(3)
        rc, code, body = w.app_curl(sub, "/api/v1/user/repos", "-u", f"{user}:{pw}",
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"name": repo, "private": True}), method="POST")
        say(f"    gitea: create repo http={code}")
        if code not in ("201", "200"):
            say(f"    gitea: repo refused {body[:200]}")
            return None
        return {"user": user, "pw": pw, "repo": repo}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=36):
            return False
        rc, code, _ = w.app_curl(sub, f"/api/v1/repos/{t['user']}/nope{secrets.token_hex(4)}",
                                 "-u", f"{t['user']}:{t['pw']}")
        if code == "200":
            say("    gitea: READBACK UNUSABLE — a repo that cannot exist returned 200")
            return False
        rc, code, body = w.app_curl(sub, f"/api/v1/repos/{t['user']}/{t['repo']}",
                                    "-u", f"{t['user']}:{t['pw']}")
        ok = code == "200" and t["repo"] in body
        say(f"    gitea: readback of the seeded repo http={code} ok={ok}")
        return ok


# =============================================================================================
class Navidrome:
    """Navidrome's own REST API: create the first admin through /auth/createAdmin, then prove the
    account survives by logging in through the same door."""
    sub = "music"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302")):
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        rc, code, out = w.app_curl(sub, "/auth/createAdmin", "-H", "Content-Type: application/json",
                                   data=json.dumps({"username": user, "password": pw}), method="POST")
        say(f"    navidrome: createAdmin http={code}")
        if code not in ("200", "201"):
            say(f"    navidrome: refused {out[:200]}")
            return None
        return {"user": user, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=36):
            return False
        bad = json.dumps({"username": t["user"], "password": "wrong-" + secrets.token_hex(6)})
        rc, code, _ = w.app_curl(sub, "/auth/login", "-H", "Content-Type: application/json",
                                 data=bad, method="POST")
        if code in ("200", "201"):
            say("    navidrome: READBACK UNUSABLE — a wrong password authenticated")
            return False
        body = json.dumps({"username": t["user"], "password": t["pw"]})
        rc, code, out = w.app_curl(sub, "/auth/login", "-H", "Content-Type: application/json",
                                   data=body, method="POST")
        ok = code in ("200", "201")
        say(f"    navidrome: login as the seeded user http={code} ok={ok}")
        return ok


# =============================================================================================
class Vaultwarden:
    """Vaultwarden's own account API: register an account, then prove it survives by asking the app
    to issue a token for it (its own login endpoint, the household's own route)."""
    sub = "vault"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/alive", want=("200",)):
            return None
        email = f"drill-{secrets.token_hex(4)}@gate.invalid"
        # Vaultwarden stores an already-hashed master key; the value is opaque to the server.
        key = base64.b64encode(secrets.token_bytes(32)).decode()
        body = json.dumps({"email": email, "name": "drill", "masterPasswordHash": key,
                           "key": "0." + base64.b64encode(secrets.token_bytes(48)).decode(),
                           "kdf": 0, "kdfIterations": 600000})
        rc, code, out = w.app_curl(sub, "/api/accounts/register",
                                   "-H", "Content-Type: application/json",
                                   data=body, method="POST")
        say(f"    vaultwarden: register http={code}")
        if code not in ("200", "204"):
            say(f"    vaultwarden: refused {out[:250]}")
            return None
        return {"email": email, "key": key}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/alive", want=("200",), tries=36):
            return False
        def login(pwhash):
            return w.app_curl(sub, "/identity/connect/token",
                              "-H", "Content-Type: application/x-www-form-urlencoded",
                              data=("grant_type=password&scope=api%20offline_access"
                                    f"&client_id=web&deviceType=9&deviceIdentifier=drill"
                                    f"&deviceName=drill&username={t['email']}&password={pwhash}"),
                              method="POST")
        rc, code, _ = login(base64.b64encode(secrets.token_bytes(32)).decode())
        if code == "200":
            say("    vaultwarden: READBACK UNUSABLE — a wrong master key authenticated")
            return False
        rc, code, out = login(t["key"].replace("+", "%2B").replace("=", "%3D").replace("/", "%2F"))
        ok = code == "200" and "access_token" in out
        say(f"    vaultwarden: token for the seeded account http={code} ok={ok}")
        if not ok:
            say(f"    vaultwarden: body {out[:200]}")
        return ok


# =============================================================================================
class Django:
    """A Django app's OWN management CLI, inside its own container, against its own User model.

    Same category as BookStack's `php artisan`: the app's own code and its own ORM, never a raw SQL
    INSERT and never a planted file (R-156). `createsuperuser --noinput` is Django's own documented
    non-interactive route, and the readback asks the SAME ORM whether the account exists.

    THE FIXTURE PROVES ITSELF ON EVERY CALL: each verify() also asks for a username that cannot
    exist and requires the answer False. A readback that has broken into always saying True
    therefore fails instead of passing everything.

    LIMITATION, recorded rather than papered over: this seeds the DATABASE half only. An app whose
    data is also FILES (adventurelog's images) has a file half this fixture does not touch.
    """

    def __init__(self, container, sub, ready_path="/", ready=("200", "302", "301", "404"),
                 python="python", workdir=None):
        # `python` and `workdir` are per-app because the image decides them: adventurelog's
        # interpreter is on PATH, tandoor ships a VENV and the bare `python` cannot import Django
        # at all ("Couldn't import Django. Are you sure it's installed…"). Measured, not guessed.
        self.container = container
        self.sub = sub
        self.ready_path = ready_path
        self.ready = ready
        self.python = python
        self.workdir = workdir

    def _wd(self):
        return f"-w {self.workdir} " if self.workdir else ""

    def _manage(self, w, code):
        # -c is passed to `manage.py shell`; the app's own shell, its own ORM.
        return w.guest(
            f"docker exec {self._wd()}{self.container} {self.python} manage.py shell "
            f"-c {json.dumps(code)} 2>&1", timeout=300)

    def _exists(self, w, username):
        # ONE LINE, semicolon-separated. A `\n` inside a double-quoted shell argument reaches
        # python as a literal backslash-n and is a SyntaxError — which is exactly how the first
        # adventurelog run read as `inconclusive`. The fixture refused to guess, which is right,
        # but the instrument was the thing that was broken.
        out = self._manage(w, (
            "from django.contrib.auth import get_user_model; "
            f"print('DRILL_ANSWER=' + str(get_user_model().objects.filter(username={username!r}).exists()))"
        ))
        m = re.search(r"DRILL_ANSWER=(True|False)", out)
        return (m.group(1) == "True") if m else None, " ".join(out.split())[-300:]

    def seed(self, w, sub, say):
        if not w.wait_app(sub, self.ready_path, want=self.ready, tries=90):
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        out = w.guest(
            f"docker exec -e DJANGO_SUPERUSER_PASSWORD={pw} {self._wd()}{self.container} "
            f"{self.python} manage.py createsuperuser --noinput "
            f"--username {user} --email {user}@gate.invalid 2>&1", timeout=300)
        say(f"    {self.container}: createsuperuser :: {' '.join(out.split())[:160]}")
        got, detail = self._exists(w, user)
        if got is not True:
            say(f"    {self.container}: the account did not appear in the app's own ORM :: {detail[:200]}")
            return None
        say(f"    {self.container}: seeded superuser {user}")
        return {"user": user, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, self.ready_path, want=self.ready, tries=90):
            say(f"    {self.container}: the app never served {self.ready_path}")
            return False
        absent, detail = self._exists(w, "nobody" + secrets.token_hex(6))
        if absent is not False:
            say(f"    {self.container}: READBACK UNUSABLE — a username that cannot exist did not "
                f"read as absent ({absent}) :: {detail[:200]}")
            return False
        found, detail = self._exists(w, t["user"])
        say(f"    {self.container}: readback of the seeded account found={found}")
        if found is not True:
            say(f"    {self.container}: :: {detail[:250]}")
        return found is True


# =============================================================================================
class Calcom:
    """Cal.com's OWN first-run API: `POST /api/auth/setup` creates the first (admin) user while the
    instance has none — the route its own setup wizard calls (upstream v6.2.0
    `apps/web/app/api/auth/setup/route.ts`). The readback is the user's PUBLIC booking page, `GET
    /<username>`, rendered by the app from its own database. Measured on 9202 2026-09-28 (v6.2.0,
    PostgreSQL 16, memory raised to 2048M in the DRILL catalog only — R-703): setup → 200, a second
    setup → 400 "No setup needed.", the page → 200, an unknown name → 404.

    THE FIXTURE PROVES ITSELF ON EVERY CALL: verify() also asks for a name that cannot exist and
    requires 404 — a readback that has broken into "always 200" fails instead of passing everything.
    """
    sub = "cal"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/auth/providers", want=("200",), tries=120):
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(6) + "Aa9x"   # >= 15 chars, a digit, both cases (its own rule)
        body = json.dumps({"username": user, "full_name": "Drill Gate", "email_address": f"{user}@gate.invalid",
                           "password": pw})
        rc, code, out = w.app_curl(sub, "/api/auth/setup", "-H", "Content-Type: application/json",
                                   data=body, method="POST")
        say(f"    calcom: /api/auth/setup http={code} :: {out[:120]}")
        if code not in ("200", "201"):
            return None
        rc, code, _ = w.app_curl(sub, "/" + user, timeout=60)
        if code != "200":
            say(f"    calcom: the seeded user's page answered {code}, not 200")
            return None
        say(f"    calcom: seeded user {user}")
        return {"user": user, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/api/auth/providers", want=("200",), tries=120):
            say("    calcom: the app never served /api/auth/providers")
            return False
        rc, code, _ = w.app_curl(sub, "/nobody" + secrets.token_hex(6), timeout=60)
        if code != "404":
            say(f"    calcom: READBACK UNUSABLE — a name that cannot exist answered {code}, not 404")
            return False
        rc, code, _ = w.app_curl(sub, "/" + t["user"], timeout=60)
        say(f"    calcom: readback — the seeded user's page http={code}")
        return code == "200"


# =============================================================================================
class Claper:
    """Claper's OWN release CLI inside its own container: `bin/claper rpc` runs Elixir code in the
    RUNNING node, against the app's own `Claper.Accounts` context (its changeset, its password hash).
    Not SQL, not a planted file (R-156). `eval` would boot a second, app-less VM — `rpc` asks the
    live one. Measured on 9202 2026-09-28 (claper 2.5.1, PostgreSQL 16): register_user → {:ok, id=2};
    the readback authenticated with the right password and REFUSED a wrong one.

    THE FIXTURE PROVES ITSELF ON EVERY CALL: verify() also asks the same function with a password
    that was never set and requires False — a readback that has broken into "always found" fails.
    """
    container = "claper"

    def _rpc(self, w, code):
        # ELIXIR_ERL_OPTIONS=+fnu: the release otherwise warns about latin1 on every call (noise only).
        out = w.guest(f"docker exec -e ELIXIR_ERL_OPTIONS=+fnu {self.container} /app/bin/claper rpc "
                      f"{json.dumps(code)} 2>&1", timeout=240)
        return " ".join(out.split())

    def _auth(self, w, email, pw):
        out = self._rpc(w, f'IO.puts("DRILL_ANSWER=#{{Claper.Accounts.get_user_by_email_and_password({json.dumps(email)}, {json.dumps(pw)}) != nil}}")')
        m = re.search(r"DRILL_ANSWER=(true|false)", out)
        return (m.group(1) == "true") if m else None, out[-300:]

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=90):
            return None
        email = f"drill-{secrets.token_hex(4)}@gate.invalid"
        pw = "Drill-" + secrets.token_hex(10)
        out = self._rpc(w, ('case Claper.Accounts.register_user(%{email: ' + json.dumps(email) + ', password: '
                            + json.dumps(pw) + '}) do {:ok, u} -> IO.puts("DRILL_SEEDED=#{u.id}"); '
                            '{:error, cs} -> IO.inspect(cs.errors, label: "DRILL_REFUSED") end'))
        say(f"    claper: register_user :: {out[-160:]}")
        got, detail = self._auth(w, email, pw)
        if got is not True:
            say(f"    claper: the account does not authenticate in the app's own context :: {detail[:200]}")
            return None
        say(f"    claper: seeded user {email}")
        return {"email": email, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=90):
            say("    claper: the app never served /")
            return False
        wrong, detail = self._auth(w, t["email"], "definitely-" + secrets.token_hex(8))
        if wrong is not False:
            say(f"    claper: READBACK UNUSABLE — a wrong password did not read as refused ({wrong}) :: {detail[:200]}")
            return False
        ok, detail = self._auth(w, t["email"], t["pw"])
        say(f"    claper: readback — the seeded account authenticates={ok}")
        if ok is not True:
            say(f"    claper: :: {detail[:250]}")
        return ok is True


# =============================================================================================
class Nextcloud:
    """Nextcloud's OWN admin CLI, `occ`, inside its own container: its own code, its own user
    backend. Not a SQL INSERT and not a planted file (R-156).

    `occ user:info` is the readback, and it PROVES ITSELF on every call: a uid that cannot exist
    must answer "user not found". A readback that has broken into always succeeding therefore
    fails instead of passing everything.

    This is the app chosen for the MariaDB engine-major edge (`09` §3 decision 5, R-469 lifted):
    the app image does NOT move, only the `mariadb:` sidecar, so the edge carries exactly one
    migration and a failure is readable.
    """
    sub = "cloud"

    def _occ(self, w, *args, timeout=420):
        import shlex
        line = " ".join(shlex.quote(a) for a in args)
        return w.guest(f"docker exec -u www-data nextcloud php occ {line} 2>&1", timeout=timeout)

    def _info(self, w, uid):
        out = self._occ(w, "user:info", uid)
        flat = " ".join(out.split())
        if "user not found" in flat.lower() or "could not be found" in flat.lower():
            return False, flat
        if f"user_id: {uid}" in flat or f"- user_id: {uid}" in flat or f"user_id: {uid}" in out:
            return True, flat
        return None, flat

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/status.php", want=("200",), tries=120):
            return None
        # /status.php answers 200 while the image's own first-run install is still going — measured
        # 2026-09-23 night on a loaded bench: `occ` then says "Nextcloud is not installed". Ask occ.
        for i in range(60):
            st = self._occ(w, "status", timeout=120)
            if "installed: true" in st:
                break
            time.sleep(5)
        else:
            say(f"    nextcloud: occ status never said installed: {' '.join(st.split())[:160]}")
            return None
        uid = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        out = w.guest(
            f"docker exec -u www-data -e OC_PASS={pw} nextcloud php occ user:add "
            f"--password-from-env --display-name={uid} {uid} 2>&1", timeout=420)
        say(f"    nextcloud: occ user:add :: {' '.join(out.split())[:160]}")
        got, flat = self._info(w, uid)
        if got is not True:
            say(f"    nextcloud: the account did not appear via occ user:info :: {flat[:220]}")
            return None
        say(f"    nextcloud: seeded user {uid}")
        return {"uid": uid, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/status.php", want=("200",), tries=120):
            say("    nextcloud: the app never served /status.php")
            return False
        absent, flat = self._info(w, "nobody" + secrets.token_hex(6))
        if absent is not False:
            say(f"    nextcloud: READBACK UNUSABLE — a uid that cannot exist did not read absent "
                f"({absent}) :: {flat[:200]}")
            return False
        found, flat = self._info(w, t["uid"])
        say(f"    nextcloud: readback of the seeded user found={found}")
        if found is not True:
            say(f"    nextcloud: :: {flat[:250]}")
        return found is True


# =============================================================================================
class Grafana:
    """Grafana's own HTTP API as the admin the DEPLOY created. The password is the one the
    controller showed the household — read from the app's own `app.yaml`, not invented — and the
    data (a folder) goes in and comes back through the app's own REST API."""
    sub = "grafana"

    def _auth(self, w, name="grafana"):
        # app.yaml stores this ENCRYPTED (`ENC:…`), so it cannot be read back off the box — which
        # is correct, and is why the harness uses the value IT generated for the deploy.
        pw = (w.GENERATED.get(name) or {}).get("GF_SECURITY_ADMIN_PASSWORD") or "admin"
        return f"admin:{pw}"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/health", want=("200",), tries=72):
            return None
        au = self._auth(w)
        title = "drill-" + secrets.token_hex(5)
        rc, code, body = w.app_curl(sub, "/api/folders", "-u", au,
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"title": title}), method="POST")
        say(f"    grafana: create folder http={code}")
        if code not in ("200", "201"):
            say(f"    grafana: refused {body[:220]}")
            return None
        try:
            uid = json.loads(body)["uid"]
        except Exception:
            say(f"    grafana: no uid in {body[:200]}")
            return None
        return {"uid": uid, "title": title}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/api/health", want=("200",), tries=72):
            return False
        au = self._auth(w)
        rc, code, _ = w.app_curl(sub, "/api/folders/nope" + secrets.token_hex(5), "-u", au)
        if code == "200":
            say("    grafana: READBACK UNUSABLE — a folder uid that cannot exist returned 200")
            return False
        rc, code, body = w.app_curl(sub, f"/api/folders/{t['uid']}", "-u", au)
        ok = code == "200" and t["title"] in body
        say(f"    grafana: readback of the seeded folder http={code} ok={ok}")
        return ok


# =============================================================================================
class AudiobookShelf:
    """audiobookshelf's own /init endpoint creates the first root account; its own /login proves
    the account survived. Both are the app's own API."""
    sub = "audiobooks"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/status", want=("200",), tries=72):
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        rc, code, body = w.app_curl(sub, "/init", "-H", "Content-Type: application/json",
                                    data=json.dumps({"newRoot": {"username": user, "password": pw}}),
                                    method="POST")
        say(f"    audiobookshelf: /init http={code}")
        if code not in ("200", "204"):
            say(f"    audiobookshelf: refused {body[:220]}")
            return None
        return {"user": user, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/status", want=("200",), tries=72):
            return False
        bad = json.dumps({"username": t["user"], "password": "wrong-" + secrets.token_hex(6)})
        rc, code, _ = w.app_curl(sub, "/login", "-H", "Content-Type: application/json",
                                 data=bad, method="POST")
        if code == "200":
            say("    audiobookshelf: READBACK UNUSABLE — a wrong password authenticated")
            return False
        rc, code, body = w.app_curl(sub, "/login", "-H", "Content-Type: application/json",
                                    data=json.dumps({"username": t["user"], "password": t["pw"]}),
                                    method="POST")
        ok = code == "200" and t["user"] in body
        say(f"    audiobookshelf: login as the seeded root http={code} ok={ok}")
        return ok


# =============================================================================================
class ActualBudget:
    """Actual's own bootstrap API sets the server password; its own login proves it survived."""
    sub = "budget"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=72):
            return None
        pw = "Drill-" + secrets.token_hex(10)
        rc, code, body = w.app_curl(sub, "/account/bootstrap",
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"password": pw}), method="POST")
        say(f"    actualbudget: /account/bootstrap http={code} :: {body[:140]}")
        if code not in ("200", "201") or '"status":"ok"' not in body:
            return None
        return {"pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=72):
            return False
        def login(p):
            return w.app_curl(sub, "/account/login", "-H", "Content-Type: application/json",
                              data=json.dumps({"loginMethod": "password", "password": p}),
                              method="POST")
        rc, code, body = login("wrong-" + secrets.token_hex(6))
        if '"status":"ok"' in body:
            say("    actualbudget: READBACK UNUSABLE — a wrong password authenticated")
            return False
        rc, code, body = login(t["pw"])
        ok = '"status":"ok"' in body
        say(f"    actualbudget: login with the seeded password http={code} ok={ok}")
        if not ok:
            say(f"    actualbudget: body {body[:200]}")
        return ok


# =============================================================================================
class Mealie:
    """Mealie ships a documented first-run admin. We log in as it through the app's own OAuth-style
    token endpoint, create a recipe through the app's own API, and read the recipe back."""
    sub = "recipes"

    def _token(self, w, sub, pw="MyPassword"):
        rc, code, body = w.app_curl(
            sub, "/api/auth/token", "-H", "Content-Type: application/x-www-form-urlencoded",
            data=f"username=changeme%40example.com&password={pw}", method="POST")
        if code != "200":
            return None, f"http={code} {body[:200]}"
        try:
            return json.loads(body)["access_token"], ""
        except Exception:
            return None, body[:200]

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/app/about", want=("200",), tries=90):
            return None
        tok, why = self._token(w, sub)
        if not tok:
            say(f"    mealie: could not authenticate as the first-run admin :: {why}")
            return None
        name = "drill-" + secrets.token_hex(5)
        rc, code, body = w.app_curl(sub, "/api/recipes", "-H", f"Authorization: Bearer {tok}",
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"name": name}), method="POST")
        say(f"    mealie: create recipe http={code}")
        if code not in ("200", "201"):
            say(f"    mealie: refused {body[:220]}")
            return None
        slug = body.strip().strip('"')
        return {"slug": slug, "name": name}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/api/app/about", want=("200",), tries=90):
            return False
        tok, why = self._token(w, sub)
        if not tok:
            say(f"    mealie: could not authenticate after the update :: {why}")
            return False
        rc, code, _ = w.app_curl(sub, "/api/recipes/nope" + secrets.token_hex(5),
                                 "-H", f"Authorization: Bearer {tok}")
        if code == "200":
            say("    mealie: READBACK UNUSABLE — a slug that cannot exist returned 200")
            return False
        rc, code, body = w.app_curl(sub, f"/api/recipes/{t['slug']}",
                                    "-H", f"Authorization: Bearer {tok}")
        ok = code == "200" and t["name"] in body
        say(f"    mealie: readback of the seeded recipe http={code} ok={ok}")
        return ok


# =============================================================================================
class N8n:
    """n8n's own owner-setup API creates the first account; its own login proves it survived."""
    sub = "auto"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/healthz", want=("200",), tries=90):
            return None
        email = f"drill-{secrets.token_hex(4)}@gate.invalid"
        pw = "Drill" + secrets.token_hex(8) + "1"
        rc, code, body = w.app_curl(sub, "/rest/owner/setup", "-H", "Content-Type: application/json",
                                    data=json.dumps({"email": email, "firstName": "drill",
                                                     "lastName": "drill", "password": pw}),
                                    method="POST")
        say(f"    n8n: /rest/owner/setup http={code}")
        if code not in ("200", "201"):
            say(f"    n8n: refused {body[:220]}")
            return None
        return {"email": email, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/healthz", want=("200",), tries=90):
            return False
        def login(p):
            return w.app_curl(sub, "/rest/login", "-H", "Content-Type: application/json",
                              data=json.dumps({"emailOrLdapLoginId": t["email"], "password": p}),
                              method="POST")
        rc, code, _ = login("wrong-" + secrets.token_hex(6))
        if code == "200":
            say("    n8n: READBACK UNUSABLE — a wrong password authenticated")
            return False
        rc, code, body = login(t["pw"])
        ok = code == "200" and t["email"] in body
        say(f"    n8n: login as the seeded owner http={code} ok={ok}")
        return ok


# =============================================================================================
class Zipline:
    """Zipline's own setup/login API. Zipline 4 creates the first user through its own endpoint."""
    sub = "img"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/healthcheck", want=("200",), tries=90):
            if not w.wait_app(sub, "/", want=("200", "302", "307"), tries=30):
                return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        for path in ("/api/auth/register", "/api/auth/setup"):
            rc, code, body = w.app_curl(sub, path, "-H", "Content-Type: application/json",
                                        data=json.dumps({"username": user, "password": pw}),
                                        method="POST")
            say(f"    zipline: {path} http={code} :: {body[:160]}")
            if code in ("200", "201"):
                return {"user": user, "pw": pw}
        say("    zipline: neither register nor setup accepted a first user")
        return None

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302", "307"), tries=60):
            return False
        def login(p):
            return w.app_curl(sub, "/api/auth/login", "-H", "Content-Type: application/json",
                              data=json.dumps({"username": t["user"], "password": p}),
                              method="POST")
        rc, code, _ = login("wrong-" + secrets.token_hex(6))
        if code == "200":
            say("    zipline: READBACK UNUSABLE — a wrong password authenticated")
            return False
        rc, code, body = login(t["pw"])
        ok = code == "200"
        say(f"    zipline: login as the seeded user http={code} ok={ok}")
        return ok


# =============================================================================================
class Vikunja:
    """Vikunja's own REST API: register a user, log in, create a project, read the project back.
    Four calls, all the app's own front door."""
    sub = "tasks"

    def _token(self, w, sub, t, pw=None):
        rc, code, body = w.app_curl(sub, "/api/v1/login", "-H", "Content-Type: application/json",
                                    data=json.dumps({"username": t["user"],
                                                     "password": pw or t["pw"]}), method="POST")
        if code != "200":
            return None, f"http={code} {body[:160]}"
        try:
            return json.loads(body)["token"], ""
        except Exception:
            return None, body[:160]

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/v1/info", want=("200",), tries=72):
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        rc, code, body = w.app_curl(sub, "/api/v1/register", "-H", "Content-Type: application/json",
                                    data=json.dumps({"username": user, "password": pw,
                                                     "email": f"{user}@gate.invalid"}),
                                    method="POST")
        say(f"    vikunja: register http={code}")
        if code not in ("200", "201"):
            say(f"    vikunja: refused {body[:220]}")
            return None
        t = {"user": user, "pw": pw}
        tok, why = self._token(w, sub, t)
        if not tok:
            say(f"    vikunja: could not log in after registering :: {why}")
            return None
        title = "drill-" + secrets.token_hex(5)
        # Vikunja CREATES with PUT, not POST — a POST answers `405 Method Not Allowed`, which
        # reads like a broken fixture and is really the wrong verb. Measured 2026-09-21.
        rc, code, body = w.app_curl(sub, "/api/v1/projects", "-H", f"Authorization: Bearer {tok}",
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"title": title}), method="PUT")
        say(f"    vikunja: create project http={code}")
        if code not in ("200", "201"):
            say(f"    vikunja: project refused {body[:220]}")
            return None
        t["title"] = title
        t["pid"] = json.loads(body).get("id")
        return t

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/api/v1/info", want=("200",), tries=72):
            return False
        bad, why = self._token(w, sub, t, pw="wrong-" + secrets.token_hex(6))
        if bad:
            say("    vikunja: READBACK UNUSABLE — a wrong password authenticated")
            return False
        tok, why = self._token(w, sub, t)
        if not tok:
            say(f"    vikunja: the seeded account no longer authenticates :: {why}")
            return False
        rc, code, body = w.app_curl(sub, f"/api/v1/projects/{t['pid']}",
                                    "-H", f"Authorization: Bearer {tok}")
        ok = code == "200" and t["title"] in body
        say(f"    vikunja: readback of the seeded project http={code} ok={ok}")
        return ok


# =============================================================================================
class OpenGist:
    """Opengist's own sign-up and sign-in FORMS.

    Two things had to be measured. Its sign-up is CSRF-protected: a bare POST answers 500 with an
    HTML page, which reads like a broken app and is really a missing token — fetch the form, keep
    its cookie, send its `_csrf` back. And its REST API refuses the account's own password
    (`401 {"message":"Bad crendentials"}`) because it wants a token the app will not mint without a
    browser. So the SEEDED DATA is the account itself and the READBACK is a real sign-in, which is
    the same shape the docmost and navidrome fixtures use.

    LIMITATION, recorded rather than papered over: this is the DATABASE half. A gist's CONTENT is
    not seeded, because that needs the API token above.
    """
    sub = "gist"

    def _form(self, w, sub, path, jar, fields):
        rc, code, html = w.app_curl(sub, path, "-b", jar, "-c", jar)
        m = re.search(r'name="_csrf"[^>]*value="([^"]+)"', html or "")
        if not m:
            return None, f"no _csrf on {path} (http={code})"
        body = "&".join([f"_csrf={m.group(1)}"] + [f"{k}={v}" for k, v in fields.items()])
        rc, code, out = w.app_curl(sub, path, "-b", jar, "-c", jar,
                                   "-H", "Content-Type: application/x-www-form-urlencoded",
                                   data=body, method="POST")
        return code, out

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=72):
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        jar = f"/tmp/og-{secrets.token_hex(4)}.jar"
        code, out = self._form(w, sub, "/register", jar, {"username": user, "password": pw})
        say(f"    opengist: /register (with its own _csrf) http={code}")
        if code not in ("200", "302", "303"):
            say(f"    opengist: refused {str(out)[:200]}")
            return None
        return {"user": user, "pw": pw}

    def verify(self, w, sub, t, say):
        # Wait for the LOGIN FORM, not for the root page. Measured 2026-09-21: immediately after a
        # successful update the root answers while /login does not yet carry its `_csrf`, so the
        # sign-in silently fails and the app looks like it lost the account. It had not.
        # 1.15 moved every page under `/-/` (`/-/login`, `/-/all`; `/login` answers 404) — measured
        # 2026-09-23 night. Ask the app which shape it serves instead of assuming one.
        pre, home_path = "", "/"
        for _ in range(72):
            if w.app_curl(sub, "/-/login")[1] == "200":
                pre, home_path = "/-", "/-/all"
                break
            if w.app_curl(sub, "/login")[1] == "200":
                break
            time.sleep(5)
        else:
            say("    opengist: neither /login nor /-/login came back after the update")
            return False
        say(f"    opengist: sign-in form at {pre}/login")
        for _ in range(24):
            rc, code, html = w.app_curl(sub, pre + "/login")
            if code == "200" and '_csrf' in (html or ""):
                break
            time.sleep(5)
        jar = f"/tmp/og-{secrets.token_hex(4)}.jar"
        code, _ = self._form(w, sub, pre + "/login", jar,
                             {"username": t["user"], "password": "wrong-" + secrets.token_hex(5)})
        rc, c2, home = w.app_curl(sub, home_path, "-b", jar)
        if t["user"] in (home or ""):
            say("    opengist: READBACK UNUSABLE — a wrong password signed in")
            return False
        jar2 = f"/tmp/og-{secrets.token_hex(4)}.jar"
        code, _ = self._form(w, sub, pre + "/login", jar2, {"username": t["user"], "password": t["pw"]})
        rc, c2, home = w.app_curl(sub, home_path, "-b", jar2)
        signed_in = t["user"] in (home or "")
        # The ACCOUNT's own public page is the readback that does not depend on a cookie: 1.15 marks its
        # session cookie Secure, so a plain-HTTP bench cannot send it back (measured 2026-09-23 night).
        # A user that was never created must 404 on the same call, or the readback proves nothing.
        rc, pc, prof = w.app_curl(sub, "/" + t["user"])
        rc, nc, _ = w.app_curl(sub, "/nobody" + secrets.token_hex(4))
        profile = pc == "200" and t["user"] in (prof or "")
        if nc == "200":
            say("    opengist: READBACK UNUSABLE — a never-created user's page answered 200")
            return None
        say(f"    opengist: account page /{t['user']} http={pc} found={profile} (never-created user {nc}); "
            f"sign-in http={code} name_on_page={signed_in}")
        return profile


# =============================================================================================
class Papra:
    """Papra's own e-mail sign-up and sign-in endpoints."""
    sub = "papra"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/health", want=("200",), tries=72):
            if not w.wait_app(sub, "/", want=("200", "302"), tries=30):
                return None
        email = f"drill-{secrets.token_hex(4)}@gate.invalid"
        pw = "Drill-" + secrets.token_hex(10)
        rc, code, body = w.app_curl(sub, "/api/auth/sign-up/email",
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"email": email, "password": pw,
                                                     "name": "drill"}), method="POST")
        say(f"    papra: sign-up http={code}")
        if code not in ("200", "201"):
            say(f"    papra: refused {body[:220]}")
            return None
        return {"email": email, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=72):
            return False
        def signin(p):
            return w.app_curl(sub, "/api/auth/sign-in/email",
                              "-H", "Content-Type: application/json",
                              data=json.dumps({"email": t["email"], "password": p}), method="POST")
        rc, code, _ = signin("wrong-" + secrets.token_hex(6))
        if code == "200":
            say("    papra: READBACK UNUSABLE — a wrong password authenticated")
            return False
        rc, code, body = signin(t["pw"])
        ok = code == "200"
        say(f"    papra: sign-in as the seeded account http={code} ok={ok}")
        return ok


# =============================================================================================
class HomeAssistant:
    """Home Assistant's own onboarding API creates the owner account and hands back a code the
    same API exchanges for a token. Both are the app's own documented non-browser route."""
    sub = "ha"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=120):
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        rc, code, body = w.app_curl(sub, "/api/onboarding/users",
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"client_id": f"https://{sub}.felhom.invalid/",
                                                     "name": "drill", "username": user,
                                                     "password": pw, "language": "en"}),
                                    method="POST")
        say(f"    home-assistant: /api/onboarding/users http={code}")
        if code not in ("200", "201"):
            say(f"    home-assistant: refused {body[:220]}")
            return None
        return {"user": user, "pw": pw}

    def _login(self, w, sub, user, pw):
        """The app's own login flow: start it, then answer it. A 200 with a step_id of
        `mfa`/`init` means the credentials were REFUSED; only `create_entry` is a pass."""
        rc, code, body = w.app_curl(sub, "/auth/login_flow",
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"client_id": f"https://{sub}.felhom.invalid/",
                                                     "handler": ["homeassistant", None],
                                                     "redirect_uri": f"https://{sub}.felhom.invalid/",
                                                     "type": "authorize"}), method="POST")
        if code not in ("200", "201"):
            return None, f"flow start http={code} {body[:160]}"
        try:
            fid = json.loads(body)["flow_id"]
        except Exception:
            return None, body[:160]
        rc, code, body = w.app_curl(sub, f"/auth/login_flow/{fid}",
                                    "-H", "Content-Type: application/json",
                                    data=json.dumps({"client_id": f"https://{sub}.felhom.invalid/",
                                                     "username": user, "password": pw}),
                                    method="POST")
        try:
            j = json.loads(body)
        except Exception:
            return None, body[:160]
        return (j.get("result") if j.get("type") == "create_entry" else None), body[:200]

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=120):
            return False
        bad, why = self._login(w, sub, t["user"], "wrong-" + secrets.token_hex(6))
        if bad:
            say("    home-assistant: READBACK UNUSABLE — a wrong password authenticated")
            return False
        good, why = self._login(w, sub, t["user"], t["pw"])
        ok = bool(good)
        say(f"    home-assistant: login as the seeded owner ok={ok}")
        if not ok:
            say(f"    home-assistant: {why}")
        return ok


# =============================================================================================
class Romm:
    """RomM's own user API, driven the way RomM's own front end drives it.

    Three things had to be measured rather than guessed, and each one answered a 403 or a 422 that
    looked like a different fault: RomM sets a **`romm_csrftoken` cookie** on any GET and requires
    it back in an **`x-csrftoken` header** (a bare POST is `403 CSRF token verification failed`,
    which reads like an auth problem); the fields go in the **JSON body**, not the query string (a
    query-string POST is `422 Field required` for every field it was just given); and `email` is
    required alongside username, password and role.

    On a fresh install with no admin the first `POST /api/users` is accepted unauthenticated;
    afterwards it is not — which is what makes the readback (`POST /api/login` as that user) a real
    authentication rather than a repeat of the seed.

    LIMITATION: this is the DATABASE half. RomM's other half is the ROM library on the drive, which
    this does not populate.
    """
    sub = "arcade"

    def _csrf(self, w, sub):
        jar = f"/tmp/romm-{secrets.token_hex(4)}.jar"
        w.app_curl(sub, "/api/heartbeat", "-c", jar)
        out = w.sh(["bash", "-lc", f"grep -i csrf {jar} | awk '{{print $7}}'"]).stdout or ""
        return jar, out.strip()

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/heartbeat", want=("200",), tries=120):
            if not w.wait_app(sub, "/", want=("200", "302"), tries=30):
                return None
        jar, tok = self._csrf(w, sub)
        if not tok:
            say("    romm: no romm_csrftoken cookie was set on /api/heartbeat")
            return None
        user = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(10)
        rc, code, body = w.app_curl(
            sub, "/api/users", "-b", jar, "-H", f"x-csrftoken: {tok}",
            "-H", "Content-Type: application/json",
            data=json.dumps({"username": user, "email": f"{user}@gate.invalid",
                             "password": pw, "role": "admin"}), method="POST")
        say(f"    romm: POST /api/users http={code}")
        if code not in ("200", "201"):
            say(f"    romm: refused {body[:220]}")
            return None
        return {"user": user, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/api/heartbeat", want=("200",), tries=120):
            return False
        jar, tok = self._csrf(w, sub)
        rc, code, _ = w.app_curl(sub, "/api/login", "-b", jar, "-H", f"x-csrftoken: {tok}",
                                 "-u", f"{t['user']}:wrong-{secrets.token_hex(5)}", method="POST")
        if code == "200":
            say("    romm: READBACK UNUSABLE — a wrong password authenticated")
            return False
        rc, code, body = w.app_curl(sub, "/api/login", "-b", jar, "-H", f"x-csrftoken: {tok}",
                                    "-u", f"{t['user']}:{t['pw']}", method="POST")
        ok = code == "200"
        say(f"    romm: login as the seeded user http={code} ok={ok}")
        if not ok:
            say(f"    romm: body {body[:200]}")
        return ok



# =============================================================================================
class Wishlist:
    """Wishlist's own SvelteKit FORM actions (added night 2026-09-23, R-612's app). Sign-up at
    /signup, then prove the account survived by signing in at /login — and by a wrong password being
    REFUSED on the same call, so a readback that always says "ok" fails instead of passing.
    SvelteKit refuses a cross-site form post: the Origin must be the app's own https origin."""
    sub = "wishlist"

    def _post(self, w, sub, path, body):
        origin = "https://" + getattr(w, "host", lambda s: f"{s}.{w.DOMAIN}")(sub)  # the Host the request carries
        rc, code, out = w.app_curl(sub, path, "-H", f"Origin: {origin}", "-H", "x-sveltekit-action: true",
                                   "-H", "Content-Type: application/x-www-form-urlencoded",
                                   data=body, method="POST")
        try:
            return code, json.loads(out)
        except Exception:
            return code, {"type": "unparsed", "raw": (out or "")[:200]}

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/signup", want=("200",), tries=72):
            return None
        u = "drill" + secrets.token_hex(3)
        pw = "Drill-" + secrets.token_hex(8)
        code, j = self._post(w, sub, "/signup",
                             f"name=Drill&username={u}&email={u}%40example.invalid&password={pw}&tokenId=")
        say(f"    wishlist: /signup http={code} type={j.get('type')}")
        if j.get("type") not in ("success", "redirect"):
            self.tried = f"POST /signup -> {code} {str(j)[:150]}"
            return None
        return {"u": u, "pw": pw}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/login", want=("200",), tries=72):
            say("    wishlist: /login never came back")
            return False
        c1, bad = self._post(w, sub, "/login", f"username={t['u']}&password=wrong-{secrets.token_hex(5)}")
        if bad.get("type") != "failure":
            say(f"    wishlist: READBACK UNUSABLE — a wrong password was not refused ({bad.get('type')})")
            return None
        c2, good = self._post(w, sub, "/login", f"username={t['u']}&password={t['pw']}")
        ok = good.get("type") in ("success", "redirect")
        say(f"    wishlist: sign-in as the seeded user type={good.get('type')} ok={ok} (wrong password refused)")
        return ok

# =============================================================================================
def _set_cookies(out):
    """The `name=value` pairs of every Set-Cookie in a `curl -D -` answer (headers + body), joined for
    a Cookie header. Kept in the fixture's own memory only; never printed."""
    pairs = re.findall(r"(?im)^set-cookie:\s*([^=;\s]+=[^;\r\n]*)", out or "")
    return "; ".join(pairs)


class Sparkyfitness:
    """SparkyFitness's OWN better-auth API: `POST /api/auth/sign-up/email` makes the first account (the
    household's own first-run route — the box's sign-up block goes up only AFTER the setup, decision 47),
    `POST /api/auth/sign-in/email` gives the session cookie, `POST /api/measurements/check-in` stores a
    weight for one date, `GET /api/measurements/check-in/<date>` reads it back. Measured on the bench
    2026-09-30 (v0.17.3, PostgreSQL 15): sign-up 200, check-in 200, readback equal, an empty date → `{}`,
    a wrong password → 401.

    THE FIXTURE PROVES ITSELF ON EVERY CALL: verify() also requires a wrong password to be refused and a
    date with no check-in to read back without the weight.
    """
    sub = "sparky"

    def _signin(self, w, sub, email, pw):
        # better-auth rate-limits sign-in per client address (a box's traefik is ONE address): a 429 is waited
        # out, never read as a verdict (measured on 9202 2026-09-30: seed sign-in + wrong + right within 1 s → 429).
        for _ in range(4):
            rc, code, out = w.app_curl(sub, "/api/auth/sign-in/email", "-D", "-", "-H", "Content-Type: application/json",
                                       "-H", f"Origin: https://{sub}.{w.DOMAIN}",
                                       data=json.dumps({"email": email, "password": pw}), method="POST")
            if code != "429":
                break
            time.sleep(15)
        return code, _set_cookies(out)

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/health", want=("200",), tries=120):
            if not w.wait_app(sub, "/", want=("200",), tries=30):
                return None
        email = "drill" + secrets.token_hex(3) + "@gate.invalid"
        pw = "Drill-" + secrets.token_hex(10)
        weight = round(50 + secrets.randbelow(4000) / 100, 2)
        rc, code, out = w.app_curl(sub, "/api/auth/sign-up/email", "-H", "Content-Type: application/json",
                                   "-H", f"Origin: https://{sub}.{w.DOMAIN}",
                                   data=json.dumps({"email": email, "password": pw, "name": "Drill"}), method="POST")
        say(f"    sparkyfitness: sign-up http={code}")
        if code not in ("200", "201"):
            self.tried = f"POST /api/auth/sign-up/email -> {code} {out[:120]}"
            return None
        code, ck = self._signin(w, sub, email, pw)
        if code != "200" or not ck:
            self.tried = f"POST /api/auth/sign-in/email -> {code}"
            return None
        rc, code, out = w.app_curl(sub, "/api/measurements/check-in", "-H", f"Cookie: {ck}",
                                   "-H", "Content-Type: application/json", "-H", f"Origin: https://{sub}.{w.DOMAIN}",
                                   data=json.dumps({"entry_date": "2026-09-01", "weight": weight}), method="POST")
        say(f"    sparkyfitness: check-in http={code}")
        if code != "200":
            self.tried = f"POST /api/measurements/check-in -> {code} {out[:120]}"
            return None
        say(f"    sparkyfitness: seeded user {email.split('@')[0]} with a weight of {weight} on 2026-09-01")
        return {"email": email, "pw": pw, "weight": weight}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200",), tries=120):
            say("    sparkyfitness: the app never served /")
            return False
        code, _ = self._signin(w, sub, t["email"], "wrong-" + secrets.token_hex(6))
        if code == "200":
            say("    sparkyfitness: READBACK UNUSABLE — a wrong password signed in")
            return False
        code, ck = self._signin(w, sub, t["email"], t["pw"])
        if code != "200" or not ck:
            say(f"    sparkyfitness: the seeded user could not sign in (http {code})")
            return False
        rc, code, out = w.app_curl(sub, "/api/measurements/check-in/1999-01-01", "-H", f"Cookie: {ck}")
        if code != "200" or '"weight"' in (out or ""):
            say(f"    sparkyfitness: READBACK UNUSABLE — an empty date answered {code} {out[:80]}")
            return False
        rc, code, out = w.app_curl(sub, "/api/measurements/check-in/2026-09-01", "-H", f"Cookie: {ck}")
        try:
            got = json.loads(out).get("weight")
        except Exception:
            got = None
        say(f"    sparkyfitness: readback of the seeded weight http={code} equal={got == t['weight']}")
        return got == t["weight"]


class Rallly:
    """Rallly's OWN better-auth API and its own tRPC, the household's route: `POST /api/better-auth/sign-up/email`
    makes the account, which then needs the six-digit code Rallly e-mails. **The one step that is not the front
    door:** the code is READ (a SELECT, never a write) from the app's own `verifications` row, standing in for the
    household's mailbox — this venue has none. The code is then given back through the front door
    (`POST /api/better-auth/email-otp/verify-email`), the session comes from `sign-in/email`, and a poll is made
    with `polls.make` (tRPC). The readback is the public `polls.get` by the poll's id. Nothing is written by hand
    (R-156). Measured on the bench 2026-09-30 (v4.11.1, PostgreSQL 16): sign-up 200 (the mail send fails —
    ESOCKET — and the code is stored anyway), verify 200, `polls.make` 200, `polls.get` 200 with the title, an
    unknown id → 404 "Poll not found".

    THE FIXTURE PROVES ITSELF ON EVERY CALL: verify() also requires an id that cannot exist to be not found.
    """
    sub = "poll"

    def _auth(self, w, sub, path, body):
        return w.app_curl(sub, "/api/better-auth/" + path, "-D", "-", "-H", "Content-Type: application/json",
                          "-H", f"Origin: https://{sub}.{w.DOMAIN}", data=json.dumps(body), method="POST")

    def _get(self, w, sub, poll_id):
        q = json.dumps({"json": {"urlId": poll_id}}, separators=(",", ":"))
        import urllib.parse
        return w.app_curl(sub, "/api/trpc/polls.get?input=" + urllib.parse.quote(q))

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/login", want=("200",), tries=90):
            return None
        email = "drill" + secrets.token_hex(3) + "@gate.invalid"
        pw = "Drill-" + secrets.token_hex(10)
        rc, code, out = self._auth(w, sub, "sign-up/email", {"email": email, "password": pw, "name": "Drill"})
        say(f"    rallly: sign-up http={code}")
        if code != "200":
            self.tried = f"POST /api/better-auth/sign-up/email -> {code}"
            return None
        otp = ""
        for _ in range(10):
            otp = w.guest("docker exec rallly-postgres psql -U rallly -d rallly -Atc "
                          f"\"select split_part(value,':',1) from verifications where identifier="
                          f"'email-verification-otp-{email}' order by created_at desc limit 1\" 2>&1").strip()
            if re.fullmatch(r"\d{6}", otp):
                break
            time.sleep(2)
        if not re.fullmatch(r"\d{6}", otp):
            self.tried = "the e-mail code was not in the app's own verifications table"
            say("    rallly: no e-mail code found in the app's own table")
            return None
        rc, code, out = self._auth(w, sub, "email-otp/verify-email", {"email": email, "otp": otp})
        say(f"    rallly: verify-email with the code http={code}")
        if code != "200":
            self.tried = f"POST /api/better-auth/email-otp/verify-email -> {code}"
            return None
        rc, code, out = self._auth(w, sub, "sign-in/email", {"email": email, "password": pw})
        ck = _set_cookies(out)
        if code != "200" or "session_token" not in ck:
            self.tried = f"POST /api/better-auth/sign-in/email -> {code}"
            return None
        title = "drillpoll-" + secrets.token_hex(4)
        rc, code, out = w.app_curl(sub, "/api/trpc/polls.make", "-H", f"Cookie: {ck}", "-H", "Content-Type: application/json",
                                   "-H", f"Origin: https://{sub}.{w.DOMAIN}",
                                   data=json.dumps({"json": {"title": title, "timeZone": "Europe/Budapest",
                                                             "options": [{"startDate": "2026-12-01"}]}}), method="POST")
        try:
            pid = json.loads(out)["result"]["data"]["json"]["data"]["id"]
        except Exception:
            self.tried = f"POST /api/trpc/polls.make -> {code} {out[:120]}"
            say(f"    rallly: polls.make -> {code} {out[:120]}")
            return None
        say(f"    rallly: seeded poll {title} ({pid})")
        return {"id": pid, "title": title}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/login", want=("200",), tries=90):
            say("    rallly: the app never served /login")
            return False
        rc, code, out = self._get(w, sub, "Nope" + secrets.token_hex(4))
        if code != "404":
            say(f"    rallly: READBACK UNUSABLE — an id that cannot exist answered {code}")
            return False
        rc, code, out = self._get(w, sub, t["id"])
        found = code == "200" and t["title"] in (out or "")
        say(f"    rallly: readback of the seeded poll http={code} found={found}")
        return found


class Outline:
    """Outline's OWN first-run API, the household's route while the app holds no workspace:
    `POST /api/installation.create` makes the workspace and its admin and signs them in (Outline mounts it
    only on self-hosted installs, and refuses it once a team exists). The session makes an API key
    (`apiKeys.create`, with Outline's own CSRF cookie + header), the key makes a collection and a published
    document, and the readback is `documents.info` with the key. Measured on the bench 2026-09-30 (v1.9.1,
    PostgreSQL 16): every call 200, the title and body read back, an unknown id → 404, a wrong key → 401.

    THE FIXTURE PROVES ITSELF ON EVERY CALL: verify() also requires an unknown document id to be not found
    and a wrong key to be refused.
    """
    sub = "kb"
    ZERO = "00000000-0000-4000-8000-000000000000"

    def _api(self, w, sub, call, body, key):
        return w.app_curl(sub, "/api/" + call, "-H", f"Authorization: Bearer {key}", "-H",
                          "Content-Type: application/json", data=json.dumps(body), method="POST")

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/_health", want=("200",), tries=90):
            return None
        o = f"https://{sub}.{w.DOMAIN}"
        rc, code, out = w.app_curl(sub, "/api/installation.create", "-D", "-", "-H", "Content-Type: application/json",
                                   "-H", f"Origin: {o}",
                                   data=json.dumps({"teamName": "Drill", "userName": "Drill",
                                                    "userEmail": "drill" + secrets.token_hex(3) + "@gate.invalid"}),
                                   method="POST")
        ck = _set_cookies(out)
        say(f"    outline: installation.create http={code}")
        if code not in ("200", "302") or "accessToken=" not in ck:
            self.tried = f"POST /api/installation.create -> {code}"
            return None
        rc, code, out = w.app_curl(sub, "/home", "-D", "-", "-o", "/dev/null", "-H", f"Cookie: {ck}")
        csrf = re.search(r"(?im)^set-cookie:\s*csrfToken=([^;\r\n]+)", out or "")
        if not csrf:
            self.tried = "no csrfToken cookie from GET /home"
            return None
        cs = csrf.group(1)
        rc, code, out = w.app_curl(sub, "/api/apiKeys.create", "-H", f"Cookie: {ck}; csrfToken={cs}", "-H", f"x-csrf-token: {cs}",
                                   "-H", "Content-Type: application/json", "-H", f"Origin: {o}",
                                   data=json.dumps({"name": "drill"}), method="POST")
        try:
            key = json.loads(out)["data"]["value"]
        except Exception:
            self.tried = f"POST /api/apiKeys.create -> {code} {out[:120]}"
            return None
        rc, code, out = self._api(w, sub, "collections.create", {"name": "Drill"}, key)
        try:
            col = json.loads(out)["data"]["id"]
        except Exception:
            self.tried = f"POST /api/collections.create -> {code} {out[:120]}"
            return None
        title, body = "drilldoc-" + secrets.token_hex(4), "drill body " + secrets.token_hex(6)
        rc, code, out = self._api(w, sub, "documents.create",
                                  {"title": title, "text": body, "collectionId": col, "publish": True}, key)
        try:
            did = json.loads(out)["data"]["id"]
        except Exception:
            self.tried = f"POST /api/documents.create -> {code} {out[:120]}"
            return None
        say(f"    outline: seeded document {title}")
        return {"key": key, "id": did, "title": title, "body": body}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/_health", want=("200",), tries=90):
            say("    outline: the app never served /_health")
            return False
        rc, code, _ = self._api(w, sub, "documents.info", {"id": self.ZERO}, t["key"])
        if code != "404":
            say(f"    outline: READBACK UNUSABLE — an unknown document answered {code}")
            return False
        rc, code, _ = self._api(w, sub, "documents.info", {"id": t["id"]}, "ol_api_" + secrets.token_hex(19))
        if code != "401":
            say(f"    outline: READBACK UNUSABLE — a wrong key answered {code}")
            return False
        rc, code, out = self._api(w, sub, "documents.info", {"id": t["id"]}, t["key"])
        found = code == "200" and t["title"] in (out or "") and t["body"] in (out or "")
        say(f"    outline: readback of the seeded document http={code} found={found}")
        return found


FIXTURES = {
    "sparkyfitness": Sparkyfitness(),
    "rallly": Rallly(),
    "outline": Outline(),
    "home-assistant": HomeAssistant(),
    "romm": Romm(),
    "vikunja": Vikunja(),
    "opengist": OpenGist(),
    "papra": Papra(),
    "mealie": Mealie(),
    "n8n": N8n(),
    "zipline": Zipline(),
    "grafana": Grafana(),
    "audiobookshelf": AudiobookShelf(),
    "actualbudget": ActualBudget(),
    "nextcloud": Nextcloud(),
    "adventurelog": Django("adventurelog", "travel", "/admin/login/"),
    "tandoor": Django("tandoor", "recipes", "/accounts/login/",
                      python="/opt/recipes/venv/bin/python", workdir="/opt/recipes"),
    # 2026-09-26 (version-travel Part B): paperless-ngx is Django too — `manage.py` in its WORKDIR
    # (/usr/src/paperless/src), python3 on PATH; measured on 9202 (the readback answered False for a
    # username that cannot exist) before this line was written.
    "paperless-ngx": Django("paperless-webserver", "paperless", "/accounts/login/"),
    "privatebin": PrivateBin(),
    "docmost": Docmost(),
    "bookstack": BookStack(),
    "gitea": Gitea(),
    "navidrome": Navidrome(),
    "vaultwarden": Vaultwarden(),
    "wishlist": Wishlist(),
    "claper": Claper(),
    "calcom": Calcom(),
}