clanker-saddle/test/bundles/canon_deploy.py

"""The canon deploy-hardening bundle (goal 27, after the third review
rejection "the canon never gets installed"): the class of miss was NOT
the install rules or the endpoints — a `meson install --destdir` image
boots and answers the canon API — it was that the running instance had
the canon CODE but no canon DIR (a stale cached package: MANIFEST.usm
1.0 pre-dated canon/, so `usm manifest deploy` satisfied the deploy from
an old artifact). This bundle pins the deploy-time truths:

- the packaged archive REALLY carries the canon files (a tar inspection
  of the .usm package -- skipped when usm is absent)
- the DEPLOYED layout resolves: `meson install --destdir` with a /usr
  prefix, then the app booted FROM THAT TREE (binary at
  <tmp>/usr/bin/clanker-saddle) in a cwd with NO ./canon and no
  CLANKER_CANON_DIR answers /api/saddle/canon with the 15 texts and
  serves one document by slug — the exe-relative installed share is the
  resolver's first non-override candidate
- the startup health line names the resolved directory and count (and
  the empty-shelf shape logs every candidate + the override)
- MANIFEST.usm/meson.build versions moved past 1.0 (the stale-package
  class: a re-deploy must build a fresh artifact)

Mind the machine: XDG_DATA_DIRS must include /usr/share (GLib schemas),
web-config.json exists only through `statum-genkeys --out` (a bare
redirect corrupts it), and the deployed app runs OUTSIDE the harness's
WORK dir with its own scratch db and its own port so nothing collides
with the primary shard app.
"""

import os
import shutil
import subprocess
import tarfile
import tempfile
import urllib.error
import urllib.request

from harness import *  # noqa: F401,F403 — the shared wire helpers


def http_raw(base, method, path, token=None):
    """A raw call against an arbitrary base URL (the deployed tree runs
    on its own port, not the harness's WEB)."""
    request = urllib.request.Request(base + path, method=method)
    if token is not None:
        request.add_header("Authorization", "Bearer " + token)
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return response.status, json.loads(response.read().decode() or "{}")
    except urllib.error.HTTPError as error:
        body = error.read().decode()
        try:
            return error.code, json.loads(body)
        except ValueError:
            return error.code, {"raw": body}


def deploy_tree(target):
    """`meson install` the existing build with a /usr prefix into
    <target>/usr; returns (ok, log tail). The builddir may carry meson's
    default /usr/local prefix (a fresh `meson setup` does) — pin /usr so
    the deployed layout is exactly an image-style install: binary at
    <target>/usr/bin, canon at <target>/usr/share/clanker-saddle/canon.
    The reconfigure regenerates but never rebuilds, and the prefix never
    affects the test-run binary's location."""
    configure = subprocess.run(
        ["meson", "configure", os.path.join(REPO, "builddir"), "-Dprefix=/usr"],
        capture_output=True, text=True, timeout=120)
    if configure.returncode != 0:
        print(configure.stdout[-1000:])
        print(configure.stderr[-1000:])
        return False, "meson configure -Dprefix=/usr failed"
    command = ["meson", "install", "-C", os.path.join(REPO, "builddir"),
               "--destdir", target]
    done = subprocess.run(command, capture_output=True, text=True, timeout=300)
    if done.returncode != 0:
        print(done.stdout[-2000:])
        print(done.stderr[-2000:])
        return False, "meson install failed"
    return True, done.stdout[-500:]


def boot_deployed(root, port):
    """Boots the DEPLOYED binary (root/usr/bin/clanker-saddle) in a cwd
    with no ./canon; returns the process handle (caller terminates)."""
    run_dir = os.path.join(root, "run")
    os.makedirs(run_dir, exist_ok=True)
    config = json.load(open(os.path.join(REPO, "web-config.json")))
    with open(os.path.join(run_dir, "web-config.json"), "w") as handle:
        json.dump(config, handle)
    env = dict(os.environ)
    env.update({
        "SPRY_DB_PATH": os.path.join(root, "deploy.sqlite"),
        "SPRY_SEED_ADMIN_USERNAME": "admin",
        "SPRY_SEED_ADMIN_PASSWORD": "adminpass123",
        # A registered saddle's bearer token is needed for the canon
        # routes; the deployed app shares nothing with the primary
        # shard, so point its ingest at the shard's fake daemon — its
        # /register handshake delivers a token we can read from the
        # primary app's /saddles slot.  The SADDLE_* knobs keep the
        # poller quiet while the shard tears down.
        "SADDLE_LLM_BASE_URL": "http://127.0.0.1:%d" % DAEMON_PORT,
        "SADDLE_HTTP_TIMEOUT_S": "2",
        "SADDLE_DISPATCH_TICK_S": "1",
        "LD_LIBRARY_PATH": "",
        "XDG_DATA_DIRS": "/usr/share",
    })
    env.pop("CLANKER_CANON_DIR", None)
    log = open(os.path.join(root, "deploy-app.log"), "w")
    process = subprocess.Popen(
        [os.path.join(root, "usr", "bin", "clanker-saddle"), str(port)],
        cwd=run_dir, env=env, stdout=log, stderr=subprocess.STDOUT)
    return process


def wait_boot(base, process, timeout=30):
    def ready():
        if process.poll() is not None:
            return "exited-%d" % process.poll()
        try:
            with urllib.request.urlopen(base + "/login", timeout=3) as response:
                return response.status == 200 and "ready" or False
        except (urllib.error.URLError, OSError):
            return False
    return wait_for(ready, timeout=timeout, what="deployed app ready")


def register_saddle(base, auth, name, host):
    """Registers a saddle THROUGH the deployed app (the canon routes
    need its own bearer token) and returns the token."""
    state = slot_public_on(base, auth, "/saddles", "saddles")
    act_on(base, auth, state["register"], {"name": name, "host": host})
    rows = [s for s in slot_public_on(base, auth, "/saddles", "saddles")["saddles"]
            if s["name"] == name]
    return rows[0]["token"]


def act_on(base, auth, action, fields):
    headers = {"X-Statum-Slot": "key=%s; as_at=%s" % (auth["key"], auth["as_at"])}
    if action.get("private"):
        headers["X-Statum-Private"] = action["private"]
    request = urllib.request.Request(
        base + action.get("uri", ""),
        data=urllib.parse.urlencode(fields).encode(), method="POST")
    request.add_header("Content-Type", "application/x-www-form-urlencoded")
    for key, value in headers.items():
        request.add_header(key, value)
    with urllib.request.urlopen(request, timeout=15) as response:
        return json.loads(response.read().decode() or "{}")


def slot_public_on(base, auth, uri, type_name):
    request = urllib.request.Request(
        base + "/_statum/entrypoint?uri=" + urllib.parse.quote(uri.split("?")[0]))
    request.add_header("X-Statum-Slot",
                       "key=%s; as_at=%s" % (auth["key"], auth["as_at"]))
    with urllib.request.urlopen(request, timeout=15) as response:
        frames = json.loads(response.read().decode())
    for directive in frames:
        if directive["type"] == "set":
            content = directive["snapshot_frame"].get("content")
            if not content:
                continue
            content = json.loads(content)
            if content["slot"]["type"] == type_name:
                return content["public"]
    raise SystemExit("no %s slot in deployed entrypoint state for %s"
                     % (type_name, uri))


def login_on(base, username, password, timeout=20):
    """Login against an arbitrary base URL. The deployed app seeds its
    admin from a GLib Idle callback, so /login answering 200 does NOT
    mean the seed landed — poll until the auth slot arrives."""
    deadline = time.monotonic() + timeout
    last_body = ""
    while time.monotonic() < deadline:
        request = urllib.request.Request(base + "/_statum/entrypoint?uri=%2Flogin")
        with urllib.request.urlopen(request, timeout=15) as response:
            frames = json.loads(response.read().decode())
        content = json.loads(frames[0]["snapshot_frame"]["content"])
        action = content["public"]["login"]
        request = urllib.request.Request(
            base + action.get("uri", ""),
            data=urllib.parse.urlencode(
                {"username": username, "password": password}).encode(),
            method="POST")
        request.add_header("Content-Type", "application/x-www-form-urlencoded")
        try:
            with urllib.request.urlopen(request, timeout=15) as response:
                body = response.read().decode()
        except urllib.error.HTTPError as error:
            last_body = error.read().decode()
            time.sleep(0.2)
            continue
        for directive in json.loads(body):
            if directive["type"] == "set":
                frame = json.loads(directive["snapshot_frame"]["content"])
                if frame["slot"]["type"] == "auth":
                    return {"key": frame["slot"]["key"], "as_at": frame["as_at"]}
    raise SystemExit("deployed login response carried no auth slot (last: %s)"
                     % last_body[:300])


def run(ctx):
    auth, fake, fake_token = ctx.auth, ctx.fake, ctx.fake_token

    print("== deploy: the manifest versions moved past 1.0 ==")
    manifest = json.load(open(os.path.join(REPO, "MANIFEST.usm")))
    check(manifest.get("version") != "1.0"
          and manifest.get("version") == "1.1",
          "MANIFEST.usm carries the bumped version 1.1 (%s)" % manifest.get("version"))

    print("== deploy: the packaged archive carries the canon ==")
    # `usm manifest package` writes ../<name>-<version>.usmc relative to
    # the repo. Building it here keeps the check honest (a stale cached
    # package is exactly the miss class this bundle pins).
    usm = shutil.which("usm")
    archive = None
    if usm is None:
        print("  (usm not on PATH — asserting the install rule instead)")
    else:
        done = subprocess.run([usm, "manifest", "package"],
                              cwd=REPO, capture_output=True, text=True, timeout=600)
        if done.returncode != 0:
            print(done.stdout[-1000:])
            print(done.stderr[-1000:])
        check(done.returncode == 0, "usm manifest package succeeds")
        archive = os.path.join(os.path.dirname(REPO),
                               "clanker-saddle-%s.usmc" % manifest["version"])
        check(os.path.exists(archive),
              "the .usmc lands beside the repo (%s)" % archive)
        if os.path.exists(archive):
            with tarfile.open(archive) as bundle:
                names = bundle.getnames()
                canon_files = [n for n in names
                               if n.startswith("./canon/") and n.endswith(".md")]
                check(len(canon_files) >= 15,
                      "the archive carries the 15 canon .md files (%d)"
                      % len(canon_files))
                check(any(n.endswith("gnu-manifesto.md") for n in canon_files)
                      and any(n.endswith("magnifica-humanitas-summary.md")
                              for n in canon_files),
                      "the archive's canon includes the anchor texts")
                # The version INSIDE the archive must be the bumped one
                # (the stale-artifact class: an old 1.0 archive would
                # satisfy a deploy with no canon at all).
                member = bundle.extractfile("./MANIFEST.usm")
                inside = json.loads(member.read().decode())
                check(inside.get("version") == manifest["version"],
                      "the archive's MANIFEST.usm carries version %s (%s)"
                      % (manifest["version"], inside.get("version")))

    print("== deploy: the deployed /usr layout serves the canon ==")
    root = tempfile.mkdtemp(prefix="canon-deploy-")
    deployed_port = free_port()
    deployed_base = "http://127.0.0.1:%d" % deployed_port
    ok, _ = deploy_tree(root)
    check(ok, "meson install --destdir <tmp>/usr with a /usr prefix")
    binary = os.path.join(root, "usr", "bin", "clanker-saddle")
    share = os.path.join(root, "usr", "share", "clanker-saddle", "canon")
    check(os.path.exists(binary), "the installed binary is at <tmp>/usr/bin")
    installed_md = [n for n in (os.listdir(share) if os.path.isdir(share) else [])
                    if n.endswith(".md")]
    check(len(installed_md) >= 15,
          "the installed share/clanker-saddle/canon carries 15 texts (%d)"
          % len(installed_md))
    if not ok or not os.path.exists(binary):
        shutil.rmtree(root, ignore_errors=True)
        return

    process = boot_deployed(root, deployed_port)
    try:
        outcome = wait_boot(deployed_base, process)
        check(outcome == "ready",
              "the deployed tree boots from a canon-less cwd (%s)" % outcome)
        if outcome != "ready":
            return
        deployed_auth = login_on(deployed_base, "admin", "adminpass123")
        token = register_saddle(deployed_base, deployed_auth,
                                "deploy-probe", "127.0.0.1:%d" % fake.port)
        check(bool(token), "a saddle registered through the deployed app")

        status, body = http_raw(deployed_base, "GET", "/api/saddle/canon", token)
        check(status == 200 and body.get("ok") is True,
              "GET /api/saddle/canon answers ok from the deployed tree")
        texts = body.get("texts") or []
        check(body.get("count") == 15 and len(texts) == 15,
              "the deployed canon lists all 15 texts (%s)" % body.get("count"))
        check(os.path.abspath(body.get("source_dir") or "") == share,
              "the deployed canon resolved the installed share dir (%s)"
              % body.get("source_dir"))
        status, doc = http_raw(deployed_base, "GET",
                               "/api/saddle/canon/free-software-definition", token)
        check(status == 200 and doc.get("ok") is True
              and doc.get("slug") == "free-software-definition"
              and "freedom" in (doc.get("body") or "").lower(),
              "the deployed canon serves one document by slug")

        print("== deploy: the startup health line ==")
        log_text = open(os.path.join(root, "deploy-app.log")).read()
        health = [line for line in log_text.splitlines()
                  if "Canon: 15 texts from" in line]
        logged = (health[0].split(" from ", 1)[1].strip()
                  if health else "")
        check(bool(health) and os.path.normpath(logged) == share,
              "the startup log names the resolved canon dir and count (%s)"
              % (health[:1] or ["absent"]))

        print("== deploy: the hero carries the canon count ==")
        dash = slot_public_on(deployed_base, deployed_auth, "/", "dashboard")
        check(int(dash.get("canon_texts", -1)) == 15,
              "the dashboard's public data carries canon_texts = 15 (%s)"
              % dash.get("canon_texts"))
    finally:
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()

    print("== deploy: the empty shelf is loud ==")
    # Boot the deployed binary in a cwd with no ./canon, no override,
    # and the installed share moved AWAY — the loud warning naming every
    # candidate is the deploying admin's only symptom, so it must be in
    # the log. (The absolute /usr/share and /usr/local/share candidates
    # do not exist on this machine — no install outside the tmpdir.)
    empty_root = tempfile.mkdtemp(prefix="canon-empty-")
    shutil.move(share, os.path.join(empty_root, "canon-parked"))
    empty_port = free_port()
    process = boot_deployed(root, empty_port)
    try:
        empty_base = "http://127.0.0.1:%d" % empty_port
        outcome = wait_boot(empty_base, process)
        check(outcome == "ready", "the canon-less deployed tree still boots (%s)"
              % outcome)
        log_text = open(os.path.join(root, "deploy-app.log")).read()
        check("no canon directory resolved" in log_text
              and "CLANKER_CANON_DIR" in log_text,
              "the empty shelf logs the loud warning naming the override")
        check("/usr/share/clanker-saddle/canon" in log_text
              and "/usr/local/share/clanker-saddle/canon" in log_text,
              "the warning names the absolute system share candidates")
        empty_auth = login_on(empty_base, "admin", "adminpass123")
        dash = slot_public_on(empty_base, empty_auth, "/", "dashboard")
        check(int(dash.get("canon_texts", -1)) == 0,
              "the canon-less dashboard reports canon_texts = 0 (%s)"
              % dash.get("canon_texts"))
        token = register_saddle(empty_base, empty_auth,
                                "empty-probe", "127.0.0.1:%d" % fake.port)
        status, body = http_raw(empty_base, "GET", "/api/saddle/canon", token)
        check(status == 200 and body.get("count") == 0,
              "the canon-less API answers an empty library, not an error")
    finally:
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
        shutil.move(os.path.join(empty_root, "canon-parked"), share)
        shutil.rmtree(empty_root, ignore_errors=True)
        shutil.rmtree(root, ignore_errors=True)