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, os, 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:
    """The first user comes from gitea's own FIRST-RUN INSTALLER (R-624, 2026-09-30): the template sets no
    INSTALL_LOCK, so a fresh instance serves the installer form at `/`, and `gitea admin user create`
    refuses (`MustInstalled()`). The household fills that form in; so does this fixture — `POST /` with the
    form's OWN default values (read from the page, never typed) plus an admin account. Measured on the bench
    2026-09-30 (1.27.0): no CSRF field on the install form; 200, gitea restarts its web server in-process, and
    the admin's Basic auth answers `/api/v1/user` 200 within seconds. On a box the setup gate (decision 46) is
    in front; walk.app_curl passes it as the household does. An instance that is ALREADY installed falls back
    to the admin CLI (the old route). Its own REST API (basic auth) then creates a repository and reads it
    back. All of these are the app's own interfaces."""
    sub = "git"
    FORM_DEFAULTS = ("db_path", "app_name", "repo_root_path", "lfs_root_path", "run_user", "domain",
                     "ssh_port", "http_port", "app_url", "log_root_path")

    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, page = w.app_curl(sub, "/")
        if 'name="db_type"' in (page or "") and 'name="admin_name"' in (page or ""):
            args = ["--data-urlencode", "db_type=sqlite3", "--data-urlencode", "password_algorithm=pbkdf2"]
            for n in self.FORM_DEFAULTS:
                m = re.search(r'name="%s" value="([^"]*)"' % n, page)
                if m:
                    args += ["--data-urlencode", f"{n}={m.group(1)}"]
            args += ["--data-urlencode", f"admin_name={user}", "--data-urlencode", f"admin_passwd={pw}",
                     "--data-urlencode", f"admin_confirm_passwd={pw}",
                     "--data-urlencode", f"admin_email={user}@gate.invalid"]
            rc, code, body = w.app_curl(sub, "/", *args, method="POST", timeout=120)
            say(f"    gitea: first-run installer POST / http={code}")
            ok = False
            for _ in range(40):              # the installer restarts gitea's web server in-process
                rc, c2, _ = w.app_curl(sub, "/api/v1/user", "-u", f"{user}:{pw}", timeout=15)
                if c2 == "200":
                    ok = True
                    break
                time.sleep(3)
            if not ok:
                self.tried = f"installer POST / -> {code}; the admin never authenticated (last {c2})"
                say(f"    gitea: {self.tried} :: {' '.join((body or '').split())[:160]}")
                return None
            say("    gitea: installed through its own first-run form; the admin authenticates")
        else:
            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: already installed — admin user create :: {' '.join(out.split())[:140]}")
            if "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)
        # /api/setup is Zipline 4's first-run route (measured on the bench 2026-09-30, v4.6.1: GET -> {"firstSetup":true},
        # POST {username,password} -> 200 with a SUPERADMIN, the seeded user logs in). The two others were guesses
        # (R-624 read this app as "no route" because of them); they stay as fallbacks.
        for path in ("/api/setup", "/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):
        # 2026-09-30: `/` answers 301 on 9202 once set up — the old wait (200/302/307) never saw the app and the
        # readback returned False with no line in the log. The seed's own health route is the wait now.
        if not w.wait_app(sub, "/api/healthcheck", want=("200",), tries=60):
            say("    zipline: /api/healthcheck never answered 200")
            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")
        # The RIGHT password first (2026-09-30): zipline rate-limits logins, and a wrong attempt first made the
        # right one answer 429 on the bench and on 9202. The wrong one after — any non-200 (401/429) is a refusal.
        code = None
        for _ in range(6):
            rc, code, body = login(t["pw"])
            if code != "429":
                break
            time.sleep(20)
        ok = code == "200"
        say(f"    zipline: login as the seeded user http={code} ok={ok}")
        rc, bad, _ = login("wrong-" + secrets.token_hex(6))
        if bad == "200":
            say("    zipline: READBACK UNUSABLE — a wrong password authenticated")
            return False
        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}")
        # R-744: Outline 1.10 names the cookie `__Host-csrfToken` on a secure request (server/utils/csrf.ts getCookieName,
        # shared/constants.ts CSRF.secureCookieName) and `csrfToken` over plain HTTP — 1.9.1 always said `csrfToken`.
        # The double-submit check compares the cookie under the request's own name with the x-csrf-token header.
        csrf = re.search(r"(?im)^set-cookie:\s*((?:__Host-)?csrfToken)=([^;\r\n]+)", out or "")
        if not csrf:
            self.tried = "no csrfToken / __Host-csrfToken cookie from GET /home"
            return None
        cn, cs = csrf.group(1), csrf.group(2)
        say(f"    outline: CSRF cookie {cn}")
        rc, code, out = w.app_curl(sub, "/api/apiKeys.create", "-H", f"Cookie: {ck}; {cn}={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


# =============================================================================================
class CalibreWeb:
    """Calibre-Web Automated (2026-09-30, R-462). THE FRONT DOOR is the household's own „Upload" button: log in
    through the login form (Flask-WTF CSRF token read from the page), then `POST /upload` with the book, exactly
    the form the page posts. The other door — dropping a file into the ingest folder (`${IMPORT_PATH}/calibre`,
    reached through FileBrowser) — is NOT used: from here it would be a file planted in a mount (R-156).
    Measured on the bench 2026-09-30: uploads are ON in a fresh v4.0.6; the book lands in the household's
    library folder (`${USERDATA_PATH}/media/books/<author>/<title> (<id>)/`, backup class `mandatory`).

    THE ADMIN PASSWORD. On a box the product's `after_install` (the app's own CLI, `cps.py -s admin:<pw>`) sets
    it from the deploy field, and the walk holds the value it generated. The bench runs no `after_install`, so
    when the generated password does not log in, the fixture runs THE SAME command the product runs — the
    app's own CLI inside its own container — and says so.

    READBACK: the app's OPDS feed (HTTP Basic, the route e-readers use): the search must list the title, and the
    book's own download must be the uploaded book (the marker read out of the EPUB the app serves). Negative
    controls on every call: a wrong password must answer 401, and a title that cannot exist must not be found."""
    sub = "books"

    @staticmethod
    def _epub(marker):
        import io, zipfile
        buf = io.BytesIO()
        z = zipfile.ZipFile(buf, "w")
        z.writestr(zipfile.ZipInfo("mimetype"), "application/epub+zip")
        z.writestr("META-INF/container.xml",
                   '<?xml version="1.0"?><container version="1.0" xmlns="urn:oasis:names:tc:opendocument:xmlns:container">'
                   '<rootfiles><rootfile full-path="OEBPS/content.opf" media-type="application/oebps-package+xml"/>'
                   '</rootfiles></container>')
        z.writestr("OEBPS/content.opf",
                   '<?xml version="1.0" encoding="UTF-8"?><package xmlns="http://www.idpf.org/2007/opf" '
                   'unique-identifier="bid" version="2.0"><metadata xmlns:dc="http://purl.org/dc/elements/1.1/" '
                   f'xmlns:opf="http://www.idpf.org/2007/opf"><dc:title>{marker}</dc:title>'
                   '<dc:creator opf:role="aut">Drill Author</dc:creator><dc:language>en</dc:language>'
                   f'<dc:identifier id="bid">urn:uuid:{marker}</dc:identifier></metadata><manifest>'
                   '<item id="c1" href="c1.xhtml" media-type="application/xhtml+xml"/>'
                   '<item id="ncx" href="toc.ncx" media-type="application/x-dtbncx+xml"/></manifest>'
                   '<spine toc="ncx"><itemref idref="c1"/></spine></package>')
        z.writestr("OEBPS/toc.ncx",
                   '<?xml version="1.0"?><ncx xmlns="http://www.daisy.org/z3986/2005/ncx/" version="2005-1">'
                   f'<head><meta name="dtb:uid" content="urn:uuid:{marker}"/></head><docTitle><text>{marker}</text>'
                   '</docTitle><navMap><navPoint id="n1" playOrder="1"><navLabel><text>One</text></navLabel>'
                   '<content src="c1.xhtml"/></navPoint></navMap></ncx>')
        z.writestr("OEBPS/c1.xhtml",
                   '<?xml version="1.0" encoding="UTF-8"?><html xmlns="http://www.w3.org/1999/xhtml"><head>'
                   f'<title>{marker}</title></head><body><p>{marker}-body the household keeps this.</p></body></html>')
        z.close()
        return buf.getvalue()

    def _opds(self, w, sub, pw, path):
        return w.app_curl(sub, path, "-u", f"admin:{pw}")

    def seed(self, w, sub, say):
        import os, tempfile
        if not w.wait_app(sub, "/login", want=("200",), tries=72):
            return None
        pw = (w.GENERATED.get("calibre-web") or {}).get("ADMIN_PASSWORD") or ""
        rc, code, _ = self._opds(w, sub, pw, "/opds")
        if code != "200":
            say(f"    calibre-web: the generated admin password does not log in (opds http={code}) — running the "
                "template's own after_install command (the app's CLI, as the product does after an install)")
            import shlex   # as the template's after_install: `user: abc` (docker exec -u keeps the image's PATH; `su` does not)
            out = w.guest("docker exec -u abc calibre-web python3 /app/calibre-web-automated/cps.py -p /config/app.db -s "
                          + shlex.quote(f"admin:{pw}") + " 2>&1")
            say(f"    calibre-web: after_install :: {' '.join(out.split())[-80:]}")
            rc, code, _ = self._opds(w, sub, pw, "/opds")
            if code != "200":
                self.tried = f"admin login after after_install -> {code}"
                return None
        jar = tempfile.mktemp(prefix="cw-jar-")
        try:
            rc, code, page = w.app_curl(sub, "/login", "-c", jar, "-b", jar)
            m = re.search(r'name="csrf_token" value="([^"]+)"', page or "")
            if not m:
                self.tried = f"GET /login -> {code}, no csrf_token"
                return None
            rc, code, _ = w.app_curl(sub, "/login", "-c", jar, "-b", jar, "--data-urlencode", f"csrf_token={m.group(1)}",
                                     "--data-urlencode", "username=admin", "--data-urlencode", f"password={pw}",
                                     "--data-urlencode", "remember_me=on", method="POST")
            rc, code, home = w.app_curl(sub, "/", "-c", jar, "-b", jar)
            m = re.search(r'name="csrf_token" value="([^"]+)"', home or "")
            has_form = 'action="/upload"' in (home or "")
            if code != "200" or not has_form or not m:
                self.tried = f"login -> home {code}, upload form present={has_form}"
                say(f"    calibre-web: {self.tried}")
                return None
            marker = "upg" + secrets.token_hex(6)
            book = tempfile.mktemp(prefix="cw-", suffix=".epub")
            open(book, "wb").write(self._epub(marker))
            rc, code, out = w.app_curl(sub, "/upload", "-c", jar, "-b", jar, "-H", f"X-CSRFToken: {m.group(1)}",
                                       "-F", f"csrf_token={m.group(1)}",
                                       "-F", f"btn-upload=@{book};type=application/epub+zip", method="POST")
            os.unlink(book)
            say(f"    calibre-web: POST /upload http={code} {out[:80]}")
            if code != "200":
                self.tried = f"POST /upload -> {code} {out[:120]}"
                return None
        finally:
            if os.path.exists(jar):
                os.unlink(jar)
        t = {"pw": pw, "marker": marker}
        for _ in range(24):                 # the upload is a task; the library entry follows within seconds
            rc, code, feed = self._opds(w, sub, pw, f"/opds/search/{marker}")
            if marker in (feed or ""):
                say(f"    calibre-web: seeded book {marker} is in the library")
                return t
            time.sleep(5)
        self.tried = "uploaded, but the book never appeared in the OPDS search"
        return None

    def verify(self, w, sub, t, say):
        import io, os, tempfile, zipfile
        if not w.wait_app(sub, "/login", want=("200",), tries=72):
            say("    calibre-web: the app never served /login")
            return False
        rc, code, _ = self._opds(w, sub, "wrong-" + secrets.token_hex(6), "/opds")
        if code != "401":
            say(f"    calibre-web: READBACK UNUSABLE — a wrong password answered {code}")
            return False
        ghost = "upg" + secrets.token_hex(6)
        rc, code, feed = self._opds(w, sub, t["pw"], f"/opds/search/{ghost}")
        if code != "200" or "<entry>" in (feed or ""):
            say(f"    calibre-web: READBACK UNUSABLE — a title that cannot exist: http={code}, entries={(feed or '').count('<entry>')}")
            return False
        feed = ""
        for _ in range(12):
            rc, code, feed = self._opds(w, sub, t["pw"], f"/opds/search/{t['marker']}")
            if code == "200" and f"<title>{t['marker']}</title>" in (feed or ""):
                break
            time.sleep(5)
        listed = f"<title>{t['marker']}</title>" in (feed or "")
        m = re.search(r'href="(/opds/download/\d+/epub/?)"', feed or "")
        served = False
        if listed and m:
            f = tempfile.mktemp(prefix="cw-dl-")
            rc, code, _ = w.app_curl(sub, m.group(1), "-u", f"admin:{t['pw']}", "-o", f)
            try:
                z = zipfile.ZipFile(io.BytesIO(open(f, "rb").read()))
                served = any(f"{t['marker']}-body" in z.read(n).decode("utf-8", "replace") for n in z.namelist())
            except Exception as e:
                say(f"    calibre-web: the download is not a readable EPUB (http={code}): {e}")
            finally:
                if os.path.exists(f):
                    os.unlink(f)
        say(f"    calibre-web: readback — listed={listed} download_link={bool(m)} book_content_served={served}")
        return listed and served


# =============================================================================================
class Wger:
    """wger (2026-09-30, R-462). THE FRONT DOOR is the web login form (`/en/user/login`, Django CSRF), then the
    app's own REST API with that session: `POST /api/v2/weightentry/` (a weight on a date — household data in its
    SQLite), read back with `GET /api/v2/weightentry/?weight=<w>`. Measured on the bench 2026-09-30 (2.6).
    NOT the app-API login (`/allauth/app/v1/auth/login`): it answers 500 on a CORRECT password on this template
    (no JWT_PRIVATE_KEY — filed, R-737). THE ADMIN PASSWORD: as calibre-web — the box's `after_install` sets it;
    the bench runs the template's own command (password as the last argument) when the generated one does not
    log in. Negative controls on every readback: no session answers 403, a weight never entered counts 0."""
    sub = "fitness"
    SET_PW = ("import os, sys; sys.path.insert(0, '/home/wger/src'); os.chdir('/home/wger/src'); "
              "os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings.main'); import django; django.setup(); "
              "from django.contrib.auth.models import User; u = User.objects.get(username='admin'); "
              "u.set_password(sys.argv[1]); u.save(); print('FELHOM_AFTER_INSTALL_OK')")

    @staticmethod
    def _host(w, sub):
        return w.host(sub) if hasattr(w, "host") else f"{sub}.{w.DOMAIN}"

    def _login(self, w, sub, pw, jar):
        o = f"https://{self._host(w, sub)}"
        hdr = ["-H", f"Origin: {o}", "-H", f"Referer: {o}/en/user/login", "-c", jar, "-b", jar]
        rc, code, page = w.app_curl(sub, "/en/user/login", *hdr)
        m = re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', page or "")
        if not m:
            return None, f"GET /en/user/login -> {code}, no csrf"
        rc, code, _ = w.app_curl(sub, "/en/user/login", *hdr, "--data-urlencode", f"csrfmiddlewaretoken={m.group(1)}",
                                 "--data-urlencode", "login=admin", "--data-urlencode", f"password={pw}", method="POST")
        jt = open(jar).read() if os.path.exists(jar) else ""
        if code != "302" or "sessionid" not in jt:
            return None, f"POST /en/user/login -> {code}"
        csrf = re.search(r"csrftoken\s+(\S+)", jt)
        return hdr + ["-H", f"X-CSRFToken: {csrf.group(1) if csrf else ''}"], "ok"

    def seed(self, w, sub, say):
        import shlex, tempfile
        if not w.wait_app(sub, "/en/user/login", want=("200",), tries=72):
            return None
        pw = (w.GENERATED.get("wger") or {}).get("ADMIN_PASSWORD") or ""
        jar = tempfile.mktemp(prefix="wger-jar-")
        try:
            hdr, why = self._login(w, sub, pw, jar)
            if hdr is None:
                say(f"    wger: the generated admin password does not log in ({why}) — running the template's own "
                    "after_install command (the app's CLI, as the product does after an install)")
                out = w.guest("docker exec wger python3 -c " + shlex.quote(self.SET_PW) + " " + shlex.quote(pw) + " 2>&1")
                say(f"    wger: after_install :: {' '.join(out.split())[-60:]}")
                hdr, why = self._login(w, sub, pw, jar)
                if hdr is None:
                    self.tried = f"web login after after_install: {why}"
                    return None
            weight = "%d.%02d" % (60 + secrets.randbelow(60), secrets.randbelow(100))
            rc, code, out = w.app_curl(sub, "/api/v2/weightentry/", *hdr, "-H", "Content-Type: application/json",
                                       data=json.dumps({"date": "2026-09-30T10:00:00Z", "weight": weight}), method="POST")
            say(f"    wger: POST /api/v2/weightentry/ http={code}")
            if code != "201":
                self.tried = f"POST /api/v2/weightentry/ -> {code} {out[:120]}"
                return None
            return {"pw": pw, "weight": weight}
        finally:
            if os.path.exists(jar):
                os.unlink(jar)

    def verify(self, w, sub, t, say):
        import tempfile
        if not w.wait_app(sub, "/en/user/login", want=("200",), tries=72):
            say("    wger: the app never served /en/user/login")
            return False
        rc, code, _ = w.app_curl(sub, f"/api/v2/weightentry/?weight={t['weight']}")
        if code not in ("401", "403"):
            say(f"    wger: READBACK UNUSABLE — no session answered {code}")
            return False
        jar = tempfile.mktemp(prefix="wger-jar-")
        try:
            hdr, why = self._login(w, sub, t["pw"], jar)
            if hdr is None:
                say(f"    wger: login failed: {why}")
                return False
            rc, code, out = w.app_curl(sub, "/api/v2/weightentry/?weight=199.99", *hdr)
            if code != "200" or '"count":0' not in (out or "").replace(" ", ""):
                say(f"    wger: READBACK UNUSABLE — a weight never entered: http={code} {out[:80]}")
                return False
            rc, code, out = w.app_curl(sub, f"/api/v2/weightentry/?weight={t['weight']}", *hdr)
            ok = code == "200" and f'"weight":"{t["weight"]}"' in (out or "").replace(" ", "")
            say(f"    wger: readback of the seeded weight entry http={code} found={ok}")
            return ok
        finally:
            if os.path.exists(jar):
                os.unlink(jar)


# =============================================================================================
class Crafty:
    """Crafty Controller 4 (2026-09-30, R-462). Its own REST API v2, behind its own HTTPS port (the template tells
    traefik `scheme=https`; the bench adapter reads that label): `POST /api/v2/auth/login` as `admin` with the
    deploy's CRAFTY_PASSWORD (the compose entrypoint writes it into default.json), then `POST /api/v2/roles` — a
    role is a record in crafty's own database, created the way its panel creates one — read back with
    `GET /api/v2/roles`. Measured on the bench 2026-09-30 (4.10.7): the POST needs `mfa_required` (500 without).
    Negative control on every readback: a wrong token must answer 401/403."""
    sub = "minecraft"

    def _token(self, w, sub, pw):
        rc, code, out = w.app_curl(sub, "/api/v2/auth/login", "-H", "Content-Type: application/json",
                                   data=json.dumps({"username": "admin", "password": pw}), method="POST")
        try:
            return json.loads(out)["data"]["token"], code
        except Exception:
            return None, f"{code} {(out or '')[:120]}"

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=90):
            return None
        pw = (w.GENERATED.get("crafty-controller") or {}).get("CRAFTY_PASSWORD") or ""
        tok, why = None, None
        for _ in range(12):                  # crafty answers `/` before its API accepts the seeded admin
            tok, why = self._token(w, sub, pw)
            if tok:
                break
            time.sleep(5)
        if not tok:
            self.tried = f"POST /api/v2/auth/login -> {why}"
            return None
        role = "upg" + secrets.token_hex(5)
        rc, code, out = w.app_curl(sub, "/api/v2/roles", "-H", f"Authorization: Bearer {tok}",
                                   "-H", "Content-Type: application/json",
                                   data=json.dumps({"name": role, "servers": [], "mfa_required": False}), method="POST")
        say(f"    crafty: POST /api/v2/roles http={code} {out[:60]}")
        if code != "200" or '"ok"' not in (out or ""):
            self.tried = f"POST /api/v2/roles -> {code} {out[:120]}"
            return None
        return {"pw": pw, "role": role}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=90):
            say("    crafty: the app never answered /")
            return False
        tok = None
        for _ in range(12):
            tok, why = self._token(w, sub, t["pw"])
            if tok:
                break
            time.sleep(5)
        if not tok:
            say(f"    crafty: login failed: {why}")
            return False
        rc, code, _ = w.app_curl(sub, "/api/v2/roles", "-H", f"Authorization: Bearer x{tok}")
        if code not in ("401", "403"):
            say(f"    crafty: READBACK UNUSABLE — a wrong token answered {code}")
            return False
        rc, code, out = w.app_curl(sub, "/api/v2/roles", "-H", f"Authorization: Bearer {tok}")
        names = [r.get("role_name") for r in (json.loads(out).get("data") or [])] if code == "200" else []
        ok = t["role"] in names
        say(f"    crafty: readback of the seeded role http={code} found={ok} (roles listed: {len(names)})")
        return ok


# =============================================================================================
class UptimeKuma:
    """Uptime Kuma 2 (2026-09-30, R-462). Its web page talks to the server over socket.io, and its first-run setup,
    login and every edit exist ONLY there — so the fixture speaks socket.io's plain HTTP long-polling transport
    (Engine.IO v4: `GET/POST /socket.io/?EIO=4&transport=polling`) through the app's own front door, exactly the
    messages the page sends: `setup` (the first admin), `login`, `addStatusPage` (a status page is a record in its
    own SQLite). READBACK through its public REST route `GET /api/status-page/<slug>` (the page households share),
    which must carry the seeded title; negative control: a slug never made must not answer 200 with a title."""
    sub = "status"
    RS = "\x1e"

    class _IO:
        def __init__(self, w, sub):
            self.w, self.sub, self.sid, self.buf, self.ack = w, sub, None, [], 0

        def _get(self):
            rc, code, out = self.w.app_curl(self.sub, f"/socket.io/?EIO=4&transport=polling&sid={self.sid}", timeout=30)
            if code != "200":
                raise RuntimeError(f"poll http={code} {out[:80]}")
            for p in (out or "").split(UptimeKuma.RS):
                if p == "2":            # ping → pong
                    self._post("3")
                elif p:
                    self.buf.append(p)

        def _post(self, body):
            rc, code, out = self.w.app_curl(self.sub, f"/socket.io/?EIO=4&transport=polling&sid={self.sid}",
                                            "-H", "Content-Type: text/plain;charset=UTF-8", data=body, method="POST")
            if code != "200":
                raise RuntimeError(f"post http={code} {out[:80]}")

        def open(self):
            rc, code, out = self.w.app_curl(self.sub, "/socket.io/?EIO=4&transport=polling")
            m = re.search(r'"sid":"([^"]+)"', out or "")
            if code != "200" or not m:
                raise RuntimeError(f"handshake http={code} {out[:80]}")
            self.sid = m.group(1)
            self._post("40")
            for _ in range(10):
                self._get()
                if any(p.startswith("40") for p in self.buf):
                    return self
            raise RuntimeError("no socket.io connect")

        def call(self, event, *args):
            n = self.ack
            self.ack += 1
            self._post(f"42{n}" + json.dumps([event, *args]))
            for _ in range(30):
                for p in list(self.buf):
                    if p.startswith(f"43{n}["):
                        self.buf.remove(p)
                        return json.loads(p[len(f"43{n}"):])[0]
                self._get()
            raise RuntimeError(f"no answer to {event}")

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=90):
            return None
        user, pw = "drill" + secrets.token_hex(3), "Drill-" + secrets.token_hex(10) + "A1"
        try:
            io = self._IO(w, sub).open()
            r = io.call("setup", user, pw)
            say(f"    uptime-kuma: setup ok={r.get('ok')} {str(r.get('msg'))[:60]}")
            if not r.get("ok"):
                self.tried = f"socket.io setup -> {r}"
                return None
            io = self._IO(w, sub).open()
            r = io.call("login", {"username": user, "password": pw, "token": ""})
            if not r.get("ok"):
                self.tried = f"socket.io login -> {str(r)[:120]}"
                return None
            title, slug = "Drill " + secrets.token_hex(4), "upg" + secrets.token_hex(4)
            r = io.call("addStatusPage", title, slug)
            say(f"    uptime-kuma: addStatusPage ok={r.get('ok')} {str(r.get('msg'))[:60]}")
            if not r.get("ok"):
                self.tried = f"socket.io addStatusPage -> {str(r)[:120]}"
                return None
        except Exception as e:
            self.tried = f"socket.io: {e}"
            say(f"    uptime-kuma: {self.tried}")
            return None
        return {"user": user, "pw": pw, "title": title, "slug": slug}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/", want=("200", "302"), tries=90):
            say("    uptime-kuma: the app never answered /")
            return False
        rc, code, out = w.app_curl(sub, f"/api/status-page/upgnever{secrets.token_hex(4)}")
        if code == "200" and '"title"' in (out or ""):
            say(f"    uptime-kuma: READBACK UNUSABLE — a slug never made answered {code} with a title")
            return False
        found = False
        for _ in range(12):
            rc, code, out = w.app_curl(sub, f"/api/status-page/{t['slug']}")
            try:
                found = code == "200" and json.loads(out)["config"]["title"] == t["title"]
            except Exception:
                found = False
            if found:
                break
            time.sleep(5)
        say(f"    uptime-kuma: readback of the seeded status page http={code} found={found}")
        return found


# =============================================================================================
class Radicale:
    """Radicale (2026-10-01, new app through NEW-APP-CHECKLIST.md). THE FRONT DOOR is CalDAV itself, the route every
    phone and desktop calendar uses: HTTP basic auth with the box's generated login (`RADICALE_USER`, default
    "csalad"; `RADICALE_PASSWORD`). Seed: MKCALENDAR a calendar, PUT one event (a unique UID and SUMMARY); read back
    with GET. Negative controls on every readback: no credentials 401, a wrong password 401, an event never created
    404 — so a read that says "found" cannot be an open door or a catch-all."""
    sub = "calendar"

    @staticmethod
    def _creds(w):
        g = w.GENERATED.get("radicale") or {}
        return g.get("RADICALE_USER") or "csalad", g.get("RADICALE_PASSWORD") or ""

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/.web/", want=("200",), tries=60):
            self.tried = "/.web/ never answered 200"
            return None
        user, pw = self._creds(w)
        cal = "felhom-" + secrets.token_hex(4)
        uid = "ev-" + secrets.token_hex(6) + "@felhom"
        summary = "Felhom teszt " + secrets.token_hex(3)
        rc, code, _ = w.app_curl(sub, f"/{user}/{cal}/", "-u", f"{user}:{pw}", method="MKCALENDAR")
        say(f"    radicale: MKCALENDAR /{user}/{cal}/ http={code}")
        if code != "201":
            self.tried = f"MKCALENDAR -> {code}"
            return None
        ics = ("BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//felhom//fixture//EN\r\nBEGIN:VEVENT\r\n"
               f"UID:{uid}\r\nDTSTAMP:20261001T100000Z\r\nDTSTART:20261002T100000Z\r\nSUMMARY:{summary}\r\n"
               "END:VEVENT\r\nEND:VCALENDAR\r\n")
        rc, code, _ = w.app_curl(sub, f"/{user}/{cal}/{uid}.ics", "-u", f"{user}:{pw}", "-H", "Content-Type: text/calendar",
                                 data=ics, method="PUT")
        say(f"    radicale: PUT event http={code}")
        if code != "201":
            self.tried = f"PUT event -> {code}"
            return None
        return {"user": user, "pw": pw, "cal": cal, "uid": uid, "summary": summary}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/.web/", want=("200",), tries=60):
            say("    radicale: /.web/ never answered")
            return False
        path = f"/{t['user']}/{t['cal']}/{t['uid']}.ics"
        rc, c_none, _ = w.app_curl(sub, path)
        rc, c_wrong, _ = w.app_curl(sub, path, "-u", f"{t['user']}:{t['pw']}x")
        rc, c_absent, _ = w.app_curl(sub, f"/{t['user']}/{t['cal']}/never-{secrets.token_hex(3)}.ics", "-u", f"{t['user']}:{t['pw']}")
        if c_none != "401" or c_wrong != "401" or c_absent != "404":
            say(f"    radicale: READBACK UNUSABLE — no auth {c_none}, wrong password {c_wrong}, absent event {c_absent}")
            return False
        rc, code, out = w.app_curl(sub, path, "-u", f"{t['user']}:{t['pw']}")
        ok = code == "200" and f"SUMMARY:{t['summary']}" in (out or "")
        say(f"    radicale: readback http={code} found={ok} (controls: no auth {c_none}, wrong {c_wrong}, absent {c_absent})")
        return ok


# =============================================================================================
class Karakeep:
    """Karakeep (2026-10-01, new app through NEW-APP-CHECKLIST.md). THE FRONT DOOR is the app's own: the first account
    is made by its sign-up call (tRPC `users.create` — the first account becomes the admin; through the setup gate on the
    box, as the household), then the route its phone app uses to sign in — `apiKeys.exchange` (email + password -> an API
    key) — and the REST API: `POST /api/v1/bookmarks` (a text bookmark with a unique title), read back with
    `GET /api/v1/bookmarks/<id>`. Negative controls on every readback: no key 401, a wrong key 401, an id never made 404.
    A text bookmark, not a URL: the crawler's success depends on the internet, the read-back must not."""
    sub = "bookmarks"

    @staticmethod
    def _trpc(w, sub, proc, payload):
        rc, code, out = w.app_curl(sub, f"/api/trpc/{proc}?batch=1", "-H", "Content-Type: application/json",
                                   data=json.dumps({"0": {"json": payload}}), method="POST")
        try:
            return code, json.loads(out)[0]
        except Exception:
            return code, {"_raw": (out or "")[:200]}

    def _key(self, w, sub, t):
        code, r = self._trpc(w, sub, "apiKeys.exchange", {"keyName": "felhom-" + secrets.token_hex(3),
                                                           "email": t["email"], "password": t["pw"]})
        return ((r.get("result") or {}).get("data") or {}).get("json", {}).get("key") if code == "200" else None

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/health", want=("200",), tries=72):
            self.tried = "/api/health never answered 200"
            return None
        email = "admin-" + secrets.token_hex(3) + "@felhom.invalid"
        pw = "Kk-" + secrets.token_hex(10)
        code, r = self._trpc(w, sub, "users.create", {"name": "Felhom", "email": email, "password": pw, "confirmPassword": pw})
        role = ((r.get("result") or {}).get("data") or {}).get("json", {}).get("role")
        say(f"    karakeep: users.create (the first account) http={code} role={role}")
        if code != "200":
            self.tried = f"users.create -> {code} {str(r)[:120]}"
            return None
        t = {"email": email, "pw": pw}
        key = self._key(w, sub, t)
        if not key:
            self.tried = "apiKeys.exchange gave no key"
            return None
        title = "felhom-seed-" + secrets.token_hex(4)
        rc, code, out = w.app_curl(sub, "/api/v1/bookmarks", "-H", f"Authorization: Bearer {key}", "-H", "Content-Type: application/json",
                                   data=json.dumps({"type": "text", "text": "Felhom teszt jegyzet " + title, "title": title}), method="POST")
        say(f"    karakeep: POST /api/v1/bookmarks http={code}")
        try:
            bid = json.loads(out)["id"]
        except Exception:
            self.tried = f"POST bookmark -> {code} {(out or '')[:120]}"
            return None
        t.update({"id": bid, "title": title, "role": role})
        return t

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/api/health", want=("200",), tries=72):
            say("    karakeep: /api/health never answered")
            return False
        key = self._key(w, sub, t)
        if not key:
            say("    karakeep: the household's email + password no longer give a key")
            return False
        rc, c_none, _ = w.app_curl(sub, f"/api/v1/bookmarks/{t['id']}")
        rc, c_wrong, _ = w.app_curl(sub, f"/api/v1/bookmarks/{t['id']}", "-H", "Authorization: Bearer ak1_felhom_wrong")
        rc, c_absent, _ = w.app_curl(sub, "/api/v1/bookmarks/never" + secrets.token_hex(4), "-H", f"Authorization: Bearer {key}")
        if c_none != "401" or c_wrong != "401" or c_absent != "404":
            say(f"    karakeep: READBACK UNUSABLE — no key {c_none}, wrong key {c_wrong}, absent id {c_absent}")
            return False
        rc, code, out = w.app_curl(sub, f"/api/v1/bookmarks/{t['id']}", "-H", f"Authorization: Bearer {key}")
        ok = code == "200" and t["title"] in (out or "")
        say(f"    karakeep: readback http={code} found={ok} (controls: no key {c_none}, wrong {c_wrong}, absent {c_absent})")
        return ok


# =============================================================================================
class Dawarich:
    """Dawarich (2026-10-01, new app through NEW-APP-CHECKLIST.md). THE FRONT DOOR is the route the phone apps use:
    `POST /api/v1/overland/batches?api_key=<key>` (Overland's GeoJSON — one point with a unique place and time), read back
    with `GET /api/v1/points?api_key=…&start_at=…&end_at=…`. The API key is the account's own (the household copies it
    from its settings page); the fixture reads it with the app's own CLI (`bin/rails runner` in the app's container —
    R-156's allowed route, nothing planted). Negative controls on every readback: no key 401, a wrong key 401, a range
    with nothing in it answers []."""
    sub = "timeline"

    @staticmethod
    def _key(w):
        out = w.guest("docker exec dawarich bin/rails runner 'puts User.order(:id).first.api_key' 2>/dev/null | tail -1")
        k = (out or "").strip()
        return k if re.fullmatch(r"[A-Za-z0-9_-]{20,}", k) else None

    def seed(self, w, sub, say):
        if not w.wait_app(sub, "/api/v1/health", want=("200",), tries=90):
            self.tried = "/api/v1/health never answered 200"
            return None
        key = self._key(w)
        if not key:
            self.tried = "no API key from the app's own CLI"
            return None
        lat = round(47.40 + secrets.randbelow(2000) / 10000.0, 4)
        lon = round(19.00 + secrets.randbelow(2000) / 10000.0, 4)
        ts = "2026-09-%02dT%02d:%02d:00Z" % (1 + secrets.randbelow(28), secrets.randbelow(24), secrets.randbelow(60))
        body = {"locations": [{"type": "Feature", "geometry": {"type": "Point", "coordinates": [lon, lat]},
                               "properties": {"timestamp": ts, "altitude": 110, "speed": 0, "horizontal_accuracy": 5,
                                              "device_id": "felhom-fixture"}}]}
        rc, code, out = w.app_curl(sub, f"/api/v1/overland/batches?api_key={key}", "-H", "Content-Type: application/json",
                                   data=json.dumps(body), method="POST")
        say(f"    dawarich: POST /api/v1/overland/batches http={code} ({lat},{lon} at {ts})")
        if code not in ("200", "201"):
            self.tried = f"overland batch -> {code} {(out or '')[:120]}"
            return None
        return {"lat": lat, "lon": lon, "ts": ts}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/api/v1/health", want=("200",), tries=90):
            say("    dawarich: /api/v1/health never answered")
            return False
        key = self._key(w)
        day = t["ts"][:10]
        rng = f"start_at={day}T00:00:00Z&end_at={day}T23:59:59Z"
        rc, c_none, _ = w.app_curl(sub, f"/api/v1/points?{rng}")
        rc, c_wrong, _ = w.app_curl(sub, f"/api/v1/points?api_key=felhom-wrong-key&{rng}")
        rc, c_empty, o_empty = w.app_curl(sub, f"/api/v1/points?api_key={key}&start_at=2001-01-01T00:00:00Z&end_at=2001-01-02T00:00:00Z")
        if c_none != "401" or c_wrong != "401" or (o_empty or "").strip() != "[]":
            say(f"    dawarich: READBACK UNUSABLE — no key {c_none}, wrong key {c_wrong}, empty range {(o_empty or '')[:40]!r}")
            return False
        found = False
        for _ in range(12):   # the batch is stored by a background job
            rc, code, out = w.app_curl(sub, f"/api/v1/points?api_key={key}&{rng}")
            # the point's fields are read as text: the unique coordinates must both appear in the answer
            found = code == "200" and str(t["lat"]) in (out or "") and str(t["lon"]) in (out or "")
            if found:
                break
            time.sleep(5)
        say(f"    dawarich: readback http={code} found={found} (controls: no key {c_none}, wrong {c_wrong}, empty range [])")
        return found


# =============================================================================================
class Grimmory:
    """Grimmory (2026-10-01, new app through NEW-APP-CHECKLIST.md). THE FRONT DOOR is the app's own API, the one its web
    client calls: the first admin by `POST /api/v1/setup` (through the setup gate on the box, as the household), sign in
    with `POST /api/v1/auth/login` (a bearer token), a library on `/books`, and a real EPUB (built here, with a unique
    title) uploaded with `POST /api/v1/files/upload` — the app writes it into the household's book folder. Read back:
    `GET /api/v1/books` lists the title. Negative controls on every readback: no token 401, a wrong password 401, and a
    second `POST /api/v1/setup` refused (403) — the readback is not an open door."""
    sub = "library"

    @staticmethod
    def _epub(title):
        import io as _io, zipfile
        b = _io.BytesIO()
        z = zipfile.ZipFile(b, "w")
        z.writestr(zipfile.ZipInfo("mimetype"), "application/epub+zip")
        z.writestr("META-INF/container.xml", '<?xml version="1.0"?><container version="1.0" xmlns="urn:oasis:names:tc:opendocument:xmlns:container"><rootfiles><rootfile full-path="OEBPS/content.opf" media-type="application/oebps-package+xml"/></rootfiles></container>')
        z.writestr("OEBPS/content.opf", f'<?xml version="1.0"?><package xmlns="http://www.idpf.org/2007/opf" unique-identifier="id" version="2.0"><metadata xmlns:dc="http://purl.org/dc/elements/1.1/"><dc:title>{title}</dc:title><dc:creator>Felhom</dc:creator><dc:identifier id="id">{title}</dc:identifier><dc:language>hu</dc:language></metadata><manifest><item id="c1" href="c1.xhtml" media-type="application/xhtml+xml"/></manifest><spine><itemref idref="c1"/></spine></package>')
        z.writestr("OEBPS/c1.xhtml", '<?xml version="1.0"?><html xmlns="http://www.w3.org/1999/xhtml"><head><title>c</title></head><body><p>Szia!</p></body></html>')
        z.close()
        return b.getvalue()

    def _token(self, w, sub, user, pw):
        rc, code, out = w.app_curl(sub, "/api/v1/auth/login", "-H", "Content-Type: application/json",
                                   data=json.dumps({"username": user, "password": pw}), method="POST")
        try:
            return code, json.loads(out).get("accessToken")
        except Exception:
            return code, None

    def seed(self, w, sub, say):
        import tempfile
        if not w.wait_app(sub, "/api/v1/healthcheck", want=("200",), tries=90):
            self.tried = "/api/v1/healthcheck never answered 200"
            return None
        user, pw = "admin", "Gm-" + secrets.token_hex(10)
        rc, code, out = w.app_curl(sub, "/api/v1/setup", "-H", "Content-Type: application/json",
                                   data=json.dumps({"username": user, "password": pw, "email": "admin@felhom.invalid", "name": "Felhom"}), method="POST")
        say(f"    grimmory: POST /api/v1/setup (the first admin) http={code}")
        if code != "200":
            self.tried = f"setup -> {code} {(out or '')[:120]}"
            return None
        code, tok = self._token(w, sub, user, pw)
        if not tok:
            self.tried = f"login -> {code}"
            return None
        auth = ["-H", f"Authorization: Bearer {tok}"]
        rc, code, out = w.app_curl(sub, "/api/v1/libraries", *auth, "-H", "Content-Type: application/json",
                                   data=json.dumps({"name": "Konyvek", "icon": "book", "iconType": "LUCIDE", "paths": [{"path": "/books"}],
                                                    "watch": True, "metadataSource": "EMBEDDED", "organizationMode": "BOOK_PER_FILE"}), method="POST")
        try:
            lib = json.loads(out)
            lid, pid = lib["id"], lib["paths"][0]["id"]
        except Exception:
            self.tried = f"create library -> {code} {(out or '')[:120]}"
            return None
        title = "Felhom Teszt " + secrets.token_hex(4)
        fn = tempfile.mktemp(suffix=".epub")
        open(fn, "wb").write(self._epub(title))
        if hasattr(w, "put_file"):
            fn = w.put_file(fn)
        rc, code, out = w.app_curl(sub, "/api/v1/files/upload", *auth, "-F", f"file=@{fn};type=application/epub+zip",
                                   "-F", f"libraryId={lid}", "-F", f"pathId={pid}", method="POST")
        say(f"    grimmory: library {lid}, upload of an EPUB http={code}")
        if code not in ("200", "201", "204"):
            self.tried = f"upload -> {code} {(out or '')[:120]}"
            return None
        return {"user": user, "pw": pw, "title": title}

    def verify(self, w, sub, t, say):
        if not w.wait_app(sub, "/api/v1/healthcheck", want=("200",), tries=90):
            say("    grimmory: /api/v1/healthcheck never answered")
            return False
        rc, c_none, _ = w.app_curl(sub, "/api/v1/books")
        c_wrong, _ = self._token(w, sub, t["user"], t["pw"] + "x")
        rc, c_setup, _ = w.app_curl(sub, "/api/v1/setup", "-H", "Content-Type: application/json",
                                    data=json.dumps({"username": "x" + secrets.token_hex(2), "password": "Gm-" + secrets.token_hex(8),
                                                     "email": "x@felhom.invalid", "name": "X"}), method="POST")
        if c_none != "401" or c_wrong != "401" or c_setup != "403":
            say(f"    grimmory: READBACK UNUSABLE — no token {c_none}, wrong password {c_wrong}, second setup {c_setup}")
            return False
        code, tok = self._token(w, sub, t["user"], t["pw"])
        found = False
        for _ in range(12):
            rc, code, out = w.app_curl(sub, "/api/v1/books", "-H", f"Authorization: Bearer {tok}")
            found = code == "200" and t["title"] in (out or "")
            if found:
                break
            time.sleep(5)
        say(f"    grimmory: readback http={code} found={found} (controls: no token {c_none}, wrong {c_wrong}, second setup {c_setup})")
        return found


FIXTURES = {
    "uptime-kuma": UptimeKuma(),
    "crafty-controller": Crafty(),
    "wger": Wger(),
    "calibre-web": CalibreWeb(),
    "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(),
    "radicale": Radicale(),
    "karakeep": Karakeep(),
    "dawarich": Dawarich(),
    "grimmory": Grimmory(),
}