#!/usr/bin/env python3
"""Scan a synthetic school fixture with the local leita binary and check it.

The tree is made-up departments, not a customer. The scan runs on a 64 MiB
tmpfs inside a user namespace so the report's volume figures are that cap,
not the machine's disk. History is written under a temporary XDG_DATA_HOME.
"""

import hashlib
import html
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

SEED = "616-school-1"
DISCLOSURE = "Synthetic example data. Not a school and not a customer."
LABEL_NAME = "SYNTHETIC-EXAMPLE.txt"
ROOT_NAME = "synthetic-example-school"
TMPFS_BYTES = 64 * 1024 * 1024
HISTORY_MIN = 1024 * 1024
MOUNT = Path("/tmp/leita-school-example/mnt")

# Logical sizes. The checker compares Leita with the allocated size stat reports.
BEFORE = {
    "administration/payroll.bin": 2 * 1024 * 1024,
    "library/catalogue.bin": 4 * 1024 * 1024,
    "science/lab.bin": 1 * 1024 * 1024,
    "science/sensors.bin": 1 * 1024 * 1024,
    "arts/recital.bin": 1 * 1024 * 1024,
}
AFTER_REPLACE = {
    "science/sensors.bin": 3 * 1024 * 1024,
}
AFTER_CREATE = {
    "arts/posters.bin": 1 * 1024 * 1024,
}

DEPARTMENTS = ("administration", "library", "science", "arts")


def allocated(path):
    st = path.stat()
    return st.st_blocks * 512


def write_file(path, size):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(b"\0" * size)


def build_before(root):
    if root.exists():
        shutil.rmtree(root)
    root.mkdir(parents=True)
    (root / LABEL_NAME).write_text(DISCLOSURE + "\n", encoding="utf-8")
    for name, size in BEFORE.items():
        write_file(root / name, size)


def apply_mutation(root):
    for name, size in AFTER_REPLACE.items():
        write_file(root / name, size)
    for name, size in AFTER_CREATE.items():
        write_file(root / name, size)


def file_truth(root):
    """Allocated bytes of every regular file, keyed by path relative to root."""
    out = {}
    for path in sorted(root.rglob("*")):
        if path.is_file():
            out[str(path.relative_to(root))] = allocated(path)
    return out


def department_bytes(files):
    totals = {name: 0 for name in DEPARTMENTS}
    label = 0
    for name, size in files.items():
        head = name.split("/", 1)[0]
        if head in totals and "/" in name:
            totals[head] += size
        elif name == LABEL_NAME:
            label += size
        else:
            raise SystemExit(f"unexpected fixture path: {name}")
    return totals, label


def rows_by_path(report, key):
    return {row["path"]: row for row in report[key]}


def require(condition, message, errors):
    if not condition:
        errors.append(message)


def check_scan(report, root, files, compared, errors):
    """Check one Leita JSON report against allocated fixture bytes."""
    prefix = str(root)
    totals, label = department_bytes(files)
    expect_bytes = sum(totals.values()) + label
    require(report.get("schema") == "leita.report/1", "schema is not leita.report/1", errors)
    require(report.get("generator", {}).get("licensee") is None, "a licensee was present", errors)
    target = report.get("target") or {}
    require(target.get("path") == prefix, f"target path {target.get('path')!r}", errors)
    require(target.get("volume") == "tmpfs", f"volume {target.get('volume')!r}", errors)
    require(target.get("filesystem") == "other", f"filesystem {target.get('filesystem')!r}", errors)
    require(target.get("engine") == "walk", f"engine {target.get('engine')!r}", errors)
    got = report.get("totals") or {}
    require(got.get("bytes") == expect_bytes, f"total bytes {got.get('bytes')} != {expect_bytes}", errors)
    require(got.get("files") == len(files), f"file count {got.get('files')} != {len(files)}", errors)
    require(got.get("directories") == len(DEPARTMENTS), f"directory count {got.get('directories')}", errors)
    space = report.get("space") or {}
    require(space.get("total") == TMPFS_BYTES, f"tmpfs cap {space.get('total')}", errors)
    require(space.get("free") == TMPFS_BYTES - expect_bytes, f"free {space.get('free')}", errors)
    dirs = rows_by_path(report, "largest_directories")
    listed = rows_by_path(report, "largest_files")
    for name, size in totals.items():
        row = dirs.get(f"{prefix}/{name}")
        require(row is not None and row.get("bytes") == size, f"{name} bytes {None if row is None else row.get('bytes')} != {size}", errors)
    for name, size in files.items():
        row = listed.get(f"{prefix}/{name}")
        require(row is not None and row.get("bytes") == size, f"{name} listed {None if row is None else row.get('bytes')} != {size}", errors)
    if compared is None:
        require(report.get("compared") is None, "no-history scan still compared", errors)
        return
    require(report.get("compared") is not None, "second scan did not compare", errors)
    for name, size in totals.items():
        row = dirs.get(f"{prefix}/{name}") or {}
        before = compared["departments"][name]
        require(row.get("bytes_before") == before, f"{name} bytes_before {row.get('bytes_before')} != {before}", errors)
        require(row.get("change") == size - before, f"{name} change {row.get('change')}", errors)
    for name, size in files.items():
        row = listed.get(f"{prefix}/{name}") or {}
        before = compared["files"].get(name)
        if before is None or before < HISTORY_MIN or size < HISTORY_MIN:
            require(row.get("bytes_before") is None and row.get("change") is None, f"{name} should have an empty change", errors)
        else:
            require(row.get("bytes_before") == before, f"{name} bytes_before {row.get('bytes_before')}", errors)
            require(row.get("change") == size - before, f"{name} change {row.get('change')}", errors)


def sanitised(text):
    home = Path.home().name
    return "/home/" not in text and "/Users/" not in text and home not in text


def published_command(argv, report):
    shown = ["leita", *argv[1:]]
    shown[shown.index(str(report))] = report.name
    return " ".join(shown)


def run_leita(binary, root, history, report, fmt, no_history):
    env = os.environ.copy()
    env["XDG_DATA_HOME"] = str(history)
    env["LEITA_COLOR"] = "never"
    argv = [str(binary), "--top", "20", "--report", str(report), "--format", fmt]
    if no_history:
        argv.append("--no-history")
    argv.append(str(root))
    proc = subprocess.run(argv, env=env, text=True, capture_output=True, check=False)
    if proc.returncode != 0:
        raise SystemExit(f"leita exited {proc.returncode}\n{proc.stderr}\n{proc.stdout}")
    if not sanitised(proc.stdout) or not sanitised(proc.stderr) or not sanitised(report.read_text(encoding="utf-8")):
        raise SystemExit("scan output contained a home path; refusing to publish it")
    return published_command(argv, report), proc.stdout


def sha16(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()[:16]


def write_page(out, truth, hashes):
    def row(name, item):
        change = "unchanged" if item["change"] == 0 else f"+{item['change']:,} bytes"
        return (
            "<tr><td>" + html.escape(name) + "</td>"
            f"<td class=\"num\">{item['before']:,}</td>"
            f"<td class=\"num\">{item['after']:,}</td>"
            f"<td class=\"num\">{html.escape(change)}</td></tr>"
        )

    body = "\n".join(row(name.title(), truth["departments"][name]) for name in DEPARTMENTS)
    commands = "\n".join(html.escape(line) for line in truth["commands"])
    page = f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Synthetic school example · Leita</title>
<meta name="description" content="A made-up school scanned by Leita, with department totals checked against the fixture. Not a customer.">
<link rel="canonical" href="https://mordaunt.dev/leita/school/">
<meta property="og:type" content="website">
<meta property="og:title" content="Synthetic school example · Leita">
<meta property="og:description" content="A made-up school scanned by Leita, with department totals checked against the fixture. Not a customer.">
<meta property="og:url" content="https://mordaunt.dev/leita/school/">
<meta name="theme-color" content="#FAF9F6" media="(prefers-color-scheme: light)">
<meta name="theme-color" content="#131312" media="(prefers-color-scheme: dark)">
<link rel="icon" href="/leita/favicon.svg?v=1ceb6ee6ae135622" type="image/svg+xml">
<script src="/leita/theme.js?v=4b8521f8216f9583"></script>
<link rel="stylesheet" href="/leita/leita.css?v=e1d12a3b3d494f4d">
<style>.school-proof pre{{max-width:100%;overflow-x:auto}}.school-table{{overflow-x:auto}}.school-table table{{width:100%;border-collapse:collapse}}.school-table th,.school-table td{{padding:8px;text-align:left;white-space:nowrap}}</style>
</head>
<body>
<header class="bar">
<div class="wrap bar-row">
<a class="brand" href="/leita/">Leita</a>
<nav aria-label="Main">
<a class="opt" href="/leita/#features">Features</a>
<a class="opt" href="/leita/#pricing">Pricing</a>
<a class="opt" href="/leita/#faq">FAQ</a>
<a class="btn small ghost" href="/leita/#download">Download</a>
</nav>
<button class="theme-control" type="button" hidden data-theme-choice>
<svg class="icon-system" viewBox="0 0 24 24" aria-hidden="true"><circle cx="12" cy="12" r="8.5"/><path d="M12 3.5a8.5 8.5 0 0 1 0 17z" fill="currentColor" stroke="none"/></svg>
<svg class="icon-light" viewBox="0 0 24 24" aria-hidden="true"><circle cx="12" cy="12" r="4"/><path d="M12 3v2M12 19v2M3 12h2M19 12h2M5.6 5.6l1.4 1.4M17 17l1.4 1.4M5.6 18.4 7 17M17 7l1.4-1.4"/></svg>
<svg class="icon-dark" viewBox="0 0 24 24" aria-hidden="true"><path d="M20.5 14.2A8.5 8.5 0 0 1 9.8 3.5a8.5 8.5 0 1 0 10.7 10.7z"/></svg>
</button>
</div>
</header>
<main class="narrow page school-proof">
<p class="kicker">Example data</p>
<h1>A synthetic school, counted by Leita.</h1>
<div class="prose">
<p><b>{html.escape(DISCLOSURE)}</b> The departments below are not a customer, a deployment, or a testimonial. The figures are this run's totals, not a saving and not a performance claim.</p>
<p>Leita {html.escape(truth["binary"]["version"])} scanned seed <code>{html.escape(SEED)}</code> on a {TMPFS_BYTES // (1024 * 1024)} MiB tmpfs. The report calls the volume tmpfs and the filesystem "other", and names the walk engine. The kernel was {html.escape(truth["host"]["uname"])}. This run does not measure ext4, btrfs, XFS, NTFS, APFS, a network share, Windows, macOS, or a VPS.</p>
<h2>What changed</h2>
<p>Four directories stand in for departments. Every counted file is NUL bytes of a fixed length. Between the scans, <code>science/sensors.bin</code> grows from 1 MiB to 3 MiB and <code>arts/posters.bin</code> appears at 1 MiB. Administration and Library are untouched. A short text file at the root carries the disclosure.</p>
<div class="school-table" tabindex="0" role="region" aria-label="Department totals">
<table>
<thead><tr><th>Department</th><th class="num">Before (bytes)</th><th class="num">After (bytes)</th><th class="num">Change</th></tr></thead>
<tbody>
{body}
</tbody>
</table>
</div>
<p>The after scan totals {truth["after"]["bytes"]:,} bytes in {truth["after"]["files"]} files and {truth["after"]["directories"]} directories. That is the allocated size of the department files plus the label file ({truth["label_bytes"]:,} bytes). Science's directory grew by {truth["departments"]["science"]["change"]:,} bytes and Arts by {truth["departments"]["arts"]["change"]:,}. The new poster file has no earlier entry, so its change is empty. The label file is under 1 MiB, and this build leaves the change empty for an entry that small; the current scan still counts it.</p>
<p>The comparison scan's duration field is {html.escape(truth["after"]["duration"])}. That is a tmpfs reading of a few megabytes on this machine. It is not a disk benchmark. Leita's summary on this machine began “{html.escape(truth["scan_notice"])}”.</p>
<h2>What this binary wrote</h2>
<p>No licence was loaded. The landing page says a current scan's CSV and JSON are free, and that historical comparisons, before-and-after reports, and HTML and PDF reports are Pro.</p>
<p>This build wrote the comparison JSON and the HTML report without a licence. The JSON is the second scan in one data directory. The HTML is the second scan of the same fixture in another data directory. Both end with: {html.escape(truth["footer"])} The current-scan CSV and JSON were separate <code>--no-history</code> invocations, and their comparison field is empty. PDF was not published from this run. Scheduled delivery, reusable report profiles, and a total across machines were not run.</p>
<h2>Downloads</h2>
<p><a href="after.html?v={hashes["after.html"]}">HTML report of the after scan</a>. <a href="after.json?v={hashes["after.json"]}">Comparison JSON</a>. <a href="before.json?v={hashes["before.json"]}">JSON of the earlier scan</a>. <a href="current.csv?v={hashes["current.csv"]}">Current-scan CSV</a>. <a href="current.json?v={hashes["current.json"]}">Current-scan JSON</a>. <a href="truth.json?v={hashes["truth.json"]}">Fixture truth</a>. <a href="reproduce.py?v={hashes["reproduce.py"]}">The runner</a>.</p>
<h2>Reproduce</h2>
<p>On Linux, with <code>unshare</code> and <code>leita</code> on <code>PATH</code>:</p>
<pre><code>python3 reproduce.py</code></pre>
<p>The script mounts the tmpfs, writes the fixture, and runs:</p>
<pre><code>{commands}</code></pre>
<p>It exits non-zero if a department total, the aggregate, or the scripted change disagrees with the allocated sizes.</p>
</div>
</main>
<footer class="wrap">
<p class="motto" lang="la" title="Not content with quiet rest"><svg class="estoile" viewBox="0 0 24 24" aria-hidden="true"><path d="M12.0 1.0Q12 12 21.5 6.5Q12 12 21.5 17.5Q12 12 12.0 23.0Q12 12 2.5 17.5Q12 12 2.5 6.5Q12 12 12.0 1.0Z"/></svg>Nec placida contenta quiete est</p>
<span>© 2026 Librium LLC</span>
<span class="links"><a href="/leita/release-notes/">release notes</a><a href="/leita/buy/">pricing</a><a href="/leita/terms/">terms</a><a href="/leita/privacy/">privacy</a><a href="/leita/refunds/">refunds</a><a href="/code/leita/LICENSE.md">licence</a><a href="/code/leita/">source</a><a href="/leita/support/">support</a><a href="/site/">mordaunt.dev</a></span>
</footer>
</body>
</html>
"""
    (out / "index.html").write_text(page, encoding="utf-8")


def binary_identity(binary):
    proc = subprocess.run([str(binary), "--version"], text=True, capture_output=True, check=True)
    line = proc.stdout.strip()
    # leita gc9f3469 (linux/amd64, built 2026-10-14)
    version = line.split()[1]
    built = line.rsplit("built", 1)[-1].strip(" )")
    return {
        "argv0": "leita",
        "version": version,
        "built": built,
        "sha256": hashlib.sha256(binary.read_bytes()).hexdigest(),
        "licensee": None,
        "line": line,
    }


def publish(binary, out):
    root = MOUNT / ROOT_NAME
    errors = []
    commands = []
    with tempfile.TemporaryDirectory(prefix="leita-school-xdg-") as xdg_root:
        xdg_root = Path(xdg_root)
        build_before(root)
        before_files = file_truth(root)
        before_path = out / "before.json"
        argv, _ = run_leita(binary, root, xdg_root / "compare-json", before_path, "json", False)
        commands.append(argv)
        before = json.loads(before_path.read_text(encoding="utf-8"))
        departments, label = department_bytes(before_files)
        check_scan(before, root, before_files, None, errors)

        apply_mutation(root)
        after_files = file_truth(root)
        after_path = out / "after.json"
        argv, stdout = run_leita(binary, root, xdg_root / "compare-json", after_path, "json", False)
        commands.append(argv)
        after = json.loads(after_path.read_text(encoding="utf-8"))
        compared = {"departments": departments, "files": before_files}
        check_scan(after, root, after_files, compared, errors)

        build_before(root)
        html_path = out / "after.html"
        run_leita(binary, root, xdg_root / "compare-html", out / "before-html-scratch.json", "json", False)
        apply_mutation(root)
        argv, _ = run_leita(binary, root, xdg_root / "compare-html", html_path, "html", False)
        commands.append(argv)
        html_text = html_path.read_text(encoding="utf-8")
        for name, size in department_bytes(after_files)[0].items():
            require(f"{root}/{name}" in html_text and f'title="{size} bytes"' in html_text, f"html missing {name} {size}", errors)

        current_json = out / "current.json"
        current_csv = out / "current.csv"
        argv, _ = run_leita(binary, root, xdg_root / "current-json", current_json, "json", True)
        commands.append(argv)
        argv, _ = run_leita(binary, root, xdg_root / "current-csv", current_csv, "csv", True)
        commands.append(argv)
        current = json.loads(current_json.read_text(encoding="utf-8"))
        check_scan(current, root, after_files, None, errors)
        csv_text = current_csv.read_text(encoding="utf-8")
        require(str(sum(after_files.values())) in csv_text, "csv missing the total", errors)

    (out / "before-html-scratch.json").unlink(missing_ok=True)
    if errors:
        raise SystemExit("fixture check failed:\n- " + "\n- ".join(errors))

    footer = after["generated_by"].split(". ", 1)[-1]
    after_departments, label = department_bytes(after_files)
    truth = {
        "seed": SEED,
        "disclosure": DISCLOSURE,
        "binary": binary_identity(binary),
        "host": {"uname": os.uname().sysname + " " + os.uname().release + " " + os.uname().machine, "role": "dev machine, not the mordaunt.dev VPS"},
        "volume": {"kind": "tmpfs", "size_bytes": TMPFS_BYTES, "filesystem_field": after["target"]["filesystem"], "engine": after["target"]["engine"]},
        "root": str(root),
        "label_bytes": label,
        "history_min_bytes": HISTORY_MIN,
        "footer": footer,
        "before": {"bytes": before["totals"]["bytes"], "files": before["totals"]["files"], "directories": before["totals"]["directories"], "generated": before["generated"]},
        "after": {
            "bytes": after["totals"]["bytes"],
            "files": after["totals"]["files"],
            "directories": after["totals"]["directories"],
            "generated": after["generated"],
            "duration": after["scan"]["duration"],
            "compared_taken": after["compared"]["taken"],
            "growth_bytes": after["totals"]["bytes"] - before["totals"]["bytes"],
        },
        "departments": {
            name: {"before": departments[name], "after": after_departments[name], "change": after_departments[name] - departments[name]}
            for name in DEPARTMENTS
        },
        "commands": commands,
        "scan_notice": next((line for line in stdout.splitlines() if line.startswith("read ") or line.startswith("walk")), ""),
    }
    if not sanitised(json.dumps(truth)):
        raise SystemExit("truth contains a home path")
    (out / "truth.json").write_text(json.dumps(truth, indent=2) + "\n", encoding="utf-8")
    names = ["before.json", "after.json", "after.html", "current.json", "current.csv", "truth.json", "reproduce.py"]
    hashes = {name: sha16(out / name) for name in names}
    write_page(out, truth, hashes)
    print(json.dumps({"ok": True, "after_bytes": truth["after"]["bytes"], "version": truth["binary"]["version"]}, indent=2))


def enter():
    if os.environ.get("LEITA_SCHOOL_INNER") == "1":
        return
    os.execvpe(
        "unshare",
        ["unshare", "--user", "--map-root-user", "--mount", sys.executable, str(Path(__file__).resolve()), "--inner", *sys.argv[1:]],
        {**os.environ, "LEITA_SCHOOL_INNER": "1"},
    )


def main():
    args = sys.argv[1:]
    inner = "--inner" in args
    if not inner:
        enter()
    out = Path(__file__).resolve().parent
    binary = Path(shutil.which("leita") or "")
    if not binary.is_file():
        raise SystemExit("leita is not on PATH")
    MOUNT.mkdir(parents=True, exist_ok=True)
    subprocess.run(["mount", "-t", "tmpfs", "-o", f"size={TMPFS_BYTES},mode=0755", "tmpfs", str(MOUNT)], check=True)
    try:
        publish(binary, out)
    finally:
        subprocess.run(["umount", str(MOUNT)], check=False)


if __name__ == "__main__":
    main()