soleprint/soleprint/atlas2/docgen/selftest.py

"""
Prove the pipeline, offline, on a tree it builds itself.

    python3 selftest.py        # or: make check

No network, nothing installed. The render steps skip with a message when
graphviz is absent rather than failing — a machine without `dot` can still check
extraction, the IR and the style layer, and saying so is more useful than a red
mark about the host.

**A check that only ever proves things work is not worth running.** So the
negative cases carry equal weight, and the four checks that are the *design*
rather than a regression are marked below. If anything ever gets cut, those are
the ones to keep:

    the IR carries no visual field          extractors cannot decide appearance
    an emitter never reads a source file    the layering, from the other side
    style names slots, not colours          one colour language, not three
    ids are stable across runs              without it, diffing is noise

## Golden tests go on the IR, never on the SVG

Graphviz measures label text with the host's fonts to size nodes, so identical
input produces different geometry on a machine with different fontconfig. The IR
is deterministic; the SVG is not. Pinning the wrong one gives a suite that fails
on someone else's laptop for no reason anyone can act on.
"""

import ast
import json
import re
import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent))

PKG = HERE.name
ir_mod = __import__(f"{PKG}.ir", fromlist=["*"])
ir_validate = __import__(f"{PKG}.ir.validate", fromlist=["*"])
py_ex = __import__(f"{PKG}.extractors.python", fromlist=["*"])
style_mod = __import__(f"{PKG}.style", fromlist=["*"])
dot_mod = __import__(f"{PKG}.emitters.dot", fromlist=["*"])
index_mod = __import__(f"{PKG}.emitters.index", fromlist=["*"])
ops_mod = __import__(f"{PKG}.ops", fromlist=["*"])
erd_mod = __import__(f"{PKG}.emitters.erd", fromlist=["*"])
nb_mod = __import__(f"{PKG}.emitters.notebook", fromlist=["*"])
spec_mod = __import__(f"{PKG}.notebook", fromlist=["*"])
db_ex = __import__(f"{PKG}.extractors.db", fromlist=["*"])
usage_ex = __import__(f"{PKG}.extractors.usage", fromlist=["*"])

check_ir = ir_mod.check
Style, StyleError = style_mod.Style, style_mod.StyleError

PASS, FAIL, SKIP = [], [], []

# A docstring that could not plausibly be a style value or a module name, so
# finding it somewhere it should not be means something really read it.
MARKER = "Quarterly Revenue Ledger"

FIXTURE = {
    "app/__init__.py": '"""The app package."""\nfrom .db import Base\n',
    "app/db.py": '"""Database plumbing."""\n\n\nclass Base:\n    """Declarative base."""\n',
    "app/models.py": (
        '"""Domain models."""\n'
        "from .db import Base\n"
        "import json\n"
        "from third_party.orm import Mixin\n"
        "\n\n"
        "class User(Base):\n"
        f'    """{MARKER}."""\n'
        "\n"
        "    def save(self):\n"
        "        pass\n"
        "\n\n"
        "class Admin(User, Mixin):\n"
        "    pass\n"
    ),
    "app/sub/deep.py": "from ..db import Base\n\n\nclass Deep(Base):\n    class Inner:\n        pass\n",
    "app/broken.py": "def oops(:\n",          # must not cost the run
}


def check(name, condition, detail=""):
    (PASS if condition else FAIL).append(name)
    print(f"  {'ok  ' if condition else 'FAIL'} {name}"
          + (f"\n         {detail}" if detail and not condition else ""))
    return condition


def _err(fn):
    """The message from a call that is expected to fail."""
    try:
        fn()
    except Exception as e:  # noqa: BLE001 - the message is the assertion
        return str(e)
    return ""


def skip(name, why):
    SKIP.append(name)
    print(f"  --   {name}  ({why})")


def build_tree(root: Path):
    for rel, text in FIXTURE.items():
        p = root / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(text)


tmp = tempfile.TemporaryDirectory(prefix="docgen-selftest-")
ROOT = Path(tmp.name) / "fx"
build_tree(ROOT)

ir = py_ex.extract(ROOT).to_dict()
ids = {n["id"] for n in ir["nodes"]}
edges = {(e["source"], e["target"], e["kind"]) for e in ir["edges"]}


# --------------------------------------------------------------------------
print("\n1. extraction — is it analysis, or is it guessing")

check("the IR validates", check_ir(ir) == [], str(check_ir(ir)[:3]))

# The claim the whole thing rests on: `ast` hands over the string "Base", and
# two-pass resolution turns it into the id of the thing it actually names.
check(
    "a relative import resolves to the real id",
    ("app.models.User", "app.db.Base", "inherits") in edges,
    f"got {sorted(e for e in edges if e[2] == 'inherits')}",
)
check(
    "a relative import two levels up resolves",
    ("app.sub.deep.Deep", "app.db.Base", "inherits") in edges,
)
check(
    "a base defined in the same module resolves",
    ("app.models.Admin", "app.models.User", "inherits") in edges,
)
check(
    "a package's own __init__ resolves against itself",
    ("app", "app.db", "imports") in edges,
    "from .db inside app/__init__.py must mean app.db, not db",
)
check(
    "no raw source name leaks in as an id",
    "Base" not in ids and "User" not in ids,
    f"unresolved names present as bare ids: {sorted(ids & {'Base', 'User'})}",
)

# Dropping an edge you cannot resolve is the worse failure: the diagram looks
# complete and is quietly missing a dependency.
externals = {n["id"] for n in ir["nodes"] if n["kind"] == "external"}
check(
    "an unresolvable name survives as `external`",
    "third_party.orm.Mixin" in externals,
    f"got {sorted(externals)}",
)
check(
    "and it keeps its edge",
    ("app.models.Admin", "third_party.orm.Mixin", "inherits") in edges,
)

check(
    "nested definitions are reached",
    "app.sub.deep.Deep.Inner" in ids and "app.models.User.save" in ids,
    "a missing generic_visit() silently loses everything below a definition",
)
check(
    "containment is recorded",
    next(n for n in ir["nodes"] if n["id"] == "app.models.User.save")["parent"]
    == "app.models.User",
)
# Spans, not just a start line. "This class is 400 lines" is structure, and a
# density map sizes a block by it — `ast` carries end_lineno, so recording only
# where something begins throws away half the fact for nothing.
spans = [n for n in ir["nodes"] if n["kind"] in ("class", "function")]
check(
    "a construct records how many lines it spans",
    all((n.get("attrs") or {}).get("lines", 0) >= 1 for n in spans),
    f"missing on {[n['id'] for n in spans if not (n.get('attrs') or {}).get('lines')][:3]}",
)
check(
    "a multi-line class spans more than one",
    any((n.get("attrs") or {}).get("lines", 0) > 1 for n in spans),
)
check(
    "a module records its length",
    all((n.get("attrs") or {}).get("lines", 0) >= 1
        for n in ir["nodes"] if n["kind"] == "module"
        and not (n.get("attrs") or {}).get("error")),
)
check(
    "a span is not a visual field",
    check_ir(ir) == [],
    "`lines` is a fact about the code, not about how it is drawn",
)

check(
    "source anchors are carried",
    all(
        (n.get("attrs") or {}).get("file")
        for n in ir["nodes"]
        if n["kind"] in ("class", "function")
    ),
)

broken = [n for n in ir["nodes"] if (n.get("attrs") or {}).get("error")]
check(
    "an unparseable file is recorded, not fatal",
    len(broken) == 1 and "broken.py" in broken[0]["attrs"]["file"],
    f"got {[(n['id'], n['attrs'].get('error')) for n in broken]}",
)
check("...and the rest of the tree still extracted", len(ids) > 10)


# --------------------------------------------------------------------------
print("\n2. the design rules")

# (design) Without this the diff emitter reports noise and nobody trusts it.
again = py_ex.extract(ROOT).to_dict()
check(
    "ids are stable — the same tree twice is the same bytes",
    json.dumps(ir, sort_keys=True) == json.dumps(again, sort_keys=True),
)

# (design) An extractor that sets a colour decides how the graph looks forever.
visual = set()
for n in ir["nodes"]:
    visual |= set(n.get("attrs") or {}) & ir_validate.VISUAL_KEYS
for e in ir["edges"]:
    visual |= set(e.get("attrs") or {}) & ir_validate.VISUAL_KEYS
check("no visual field reaches the IR", not visual, f"found {sorted(visual)}")

check(
    "the IR is readable without this package",
    isinstance(json.loads(json.dumps(ir)), dict),
    "it must be plain JSON, not something that needs a library to open",
)

# (design) The layering, checked from the emitter side. Parsed, not grepped:
# a docstring mentioning `ast` is prose, an import is a dependency.
offenders = []
for src in sorted((HERE / "emitters").glob("*.py")):
    tree = ast.parse(src.read_text(), str(src))
    for node in ast.walk(tree):
        names = []
        if isinstance(node, ast.Import):
            names = [a.name for a in node.names]
        elif isinstance(node, ast.ImportFrom):
            names = [node.module or ""]
        for name in names:
            if name.split(".")[0] in {"ast", "sqlalchemy", "griffe"}:
                offenders.append(f"{src.name}:{node.lineno} imports {name}")
check(
    "no emitter reads source — it has never heard of Python or SQL",
    not offenders,
    "; ".join(offenders),
)

extractor_offenders = []
for src in sorted((HERE / "extractors").rglob("*.py")):
    tree = ast.parse(src.read_text(), str(src))
    for node in ast.walk(tree):
        names = []
        if isinstance(node, ast.Import):
            names = [a.name for a in node.names]
        elif isinstance(node, ast.ImportFrom):
            names = [node.module or ""]
        if any(n and ("emitters" in n or "style" in n) for n in names):
            extractor_offenders.append(f"{src.name}:{node.lineno}")
check(
    "no extractor knows about style or emitters",
    not extractor_offenders,
    "; ".join(extractor_offenders),
)


# --------------------------------------------------------------------------
print("\n3. style — one colour language")

styles = Style.available()
check("a style ships", "lucid" in styles, f"found {styles}")

lucid = Style.load("lucid")                      # the default: dark
print_theme = Style.load("lucid", theme="lucid")  # the print palette

check("it carries more than one theme", set(lucid.themes()) >= {"lucid", "dark"})
check(
    "an unknown kind falls back rather than crashing",
    lucid.node("no-such-kind-at-all") == lucid.node("default") != {},
)

# (design) A rule naming a hex is a rule outside the shared vocabulary.
raw = json.loads((HERE / "style" / "lucid.json").read_text())
leaked = []
for section in ("nodes", "groups", "edges", "graph"):
    for kind, rule in (raw.get(section) or {}).items():
        if not isinstance(rule, dict):
            continue
        for key, value in rule.items():
            if isinstance(value, str) and re.fullmatch(r"#[0-9A-Fa-f]{3,8}", value):
                leaked.append(f"{section}.{kind}.{key}={value}")
check(
    "style rules name slots, never colours",
    not leaked,
    "; ".join(leaked) + "  <- the colour language has stopped being one language",
)

# A theme that defines only some of the slots renders half a diagram in the
# right colours and the rest in whatever DOT does with an empty string.
half_bound = []
for theme in lucid.themes():
    try:
        Style.load("lucid", theme=theme)
    except StyleError as e:
        half_bound.append(f"{theme}: {e}")
check("every slot resolves in every theme", not half_bound, "; ".join(half_bound))

# The point of binding to spr's models: a diagram and the page around it must
# not drift apart. docs/graphs/README.md states the rule.
SYSTEM_ACCENTS = {
    "artery": ("artery/index.html", "#b91c1c"),
    "atlas": ("atlas/index.html", "#15803d"),
    "station": ("station/index.html", "#1d4ed8"),
}
spr_root = HERE.parent.parent
mismatched = []
for slot, (page, expected) in SYSTEM_ACCENTS.items():
    got = lucid.slot(slot).lower()
    if got != expected:
        mismatched.append(f"{slot}: style has {got}, {page} sets {expected}")
    src = spr_root / page
    if src.exists() and expected not in src.read_text():
        mismatched.append(f"{slot}: {page} no longer sets {expected}")
check(
    "the dark theme's slots are spr's own --system-accent values",
    not mismatched,
    "; ".join(mismatched),
)

check(
    "domain -> slot is a mapping, not a colour choice",
    lucid.domain_slot("atlas") == "atlas" and lucid.domain_slot("station") == "station",
)
check(
    "an undomained group still gets a deterministic slot",
    [lucid.domain_slot(None, i) for i in range(4)]
    == [lucid.domain_slot(None, i) for i in range(4)],
)

check(
    "the style records where DOT runs out",
    {"header-bar", "dasharray", "sequence-badge"} <= set(lucid.limits()),
    f"got {sorted(lucid.limits())}",
)


# --------------------------------------------------------------------------
print("\n4. emitters")

dot_text = dot_mod.emit(ir, lucid)
check("DOT is produced", dot_text.startswith("digraph ir {"))
check(
    "every IR edge reaches the DOT",
    dot_text.count("->") == len(ir["edges"]),
    f"{dot_text.count('->')} of {len(ir['edges'])} — cluster endpoints need lhead/ltail",
)
check("the profile's colours are applied", lucid.node("class")["border"] in dot_text)
check(
    "ids and kinds reach the output for a front end to bind to",
    'id="app.models.User"' in dot_text and 'class="class"' in dot_text,
)
check("source anchors become links", "href=" in dot_text)

print_text = dot_mod.emit(ir, print_theme)
check(
    "one IR, two looks, no re-extraction",
    print_text != dot_text and print_theme.slot("surface-0") in print_text,
)

md = index_mod.to_markdown(ir)
check("the index names what exists", "app.models" in md or "models" in md)
check("the index carries the docstrings", MARKER in md)
check(
    "the index reports the dependency surface",
    "third_party.orm.Mixin" in md,
    "externals are the thing a reader most often wants and a diagram scatters",
)
check("the index reports what could not be parsed", "Not parsed" in md)

sidebar = index_mod.to_sidebar(ir)
check("the sidebar is nested, not flat", any(i.get("children") for i in sidebar["items"]))


# --------------------------------------------------------------------------
print("\n5. views")

no_std = ops_mod.drop_stdlib(ir)
check(
    "dropping the stdlib removes json and keeps third-party",
    "json" not in {n["id"] for n in no_std["nodes"]}
    and "third_party.orm.Mixin" in {n["id"] for n in no_std["nodes"]},
)
check("a view still validates", check_ir(no_std) == [], str(check_ir(no_std)[:3]))

classes = ops_mod.only_kinds(ir, {"class"})
check(
    "keeping one kind keeps its ancestors, so containment survives",
    check_ir(classes) == [] and "app.models" in {n["id"] for n in classes["nodes"]},
)

around = ops_mod.neighbourhood(ir, "app.db.Base", hops=1)
check(
    "a neighbourhood keeps what points at it",
    "app.models.User" in {n["id"] for n in around["nodes"]} and check_ir(around) == [],
)

shallow = ops_mod.collapse_to_depth(ir, 1)
check(
    "collapsing to a depth drops what is inside",
    "app.models.User.save" not in {n["id"] for n in shallow["nodes"]}
    and check_ir(shallow) == [],
)

# The difference between a simpler picture and a wrong one. `app.models.User`
# inherits from `app.db.Base`; look at the module level and that is a dependency
# of app.models on app.db. Dropping the edge says they are unrelated.
mods = ops_mod.only_kinds(ir, {"module"})
mod_edges = {(e["source"], e["target"], e["kind"]) for e in mods["edges"]}
check(
    "an edge is lifted to the surviving node, not dropped",
    ("app.models", "app.db", "inherits") in mod_edges,
    f"got {sorted(mod_edges)}",
)
check(
    "a lifted edge says how many it stands for",
    any((e.get("attrs") or {}).get("weight") for e in mods["edges"]),
)
check(
    "lifting never leaves a self-loop",
    not [e for e in mods["edges"] if e["source"] == e["target"]],
    "both ends inside one node is a fact about that node, not a line",
)
check(
    "a label that no longer applies is dropped with the lift",
    not any(
        (e.get("attrs") or {}).get("label")
        for e in mods["edges"]
        if (e.get("attrs") or {}).get("weight")
    ),
)

# The default view: what you get without knowing which filters to ask for.
ov = ops_mod.overview(ir)
ov_kinds = {n["kind"] for n in ov["nodes"]}
check(
    "the python overview is modules and what is outside",
    ov_kinds <= {"module", "external"} and check_ir(ov) == [],
    f"got {sorted(ov_kinds)}",
)
check(
    "it drops the stdlib but keeps third-party",
    "json" not in {n["id"] for n in ov["nodes"]}
    and "third_party.orm.Mixin" in {n["id"] for n in ov["nodes"]},
)
# Depth is the tempting knob and the wrong one: a directory without
# `__init__.py` is not a package, so its modules have no parent and sit at
# depth 0. soleprint has 173 such roots.
fragmented = ops_mod.overview(
    {"meta": {"source": "python", "root": "x", "schema_version": "1"},
     "nodes": [{"id": "loose", "kind": "module", "label": "loose", "parent": None, "attrs": {}},
               {"id": "loose.C", "kind": "class", "label": "C", "parent": "loose", "attrs": {}}],
     "edges": []}
)
check(
    "a parentless module still reduces to its module",
    [n["id"] for n in fragmented["nodes"]] == ["loose"],
    "depth-based collapsing would have kept the class; kind-based does not care",
)


# --------------------------------------------------------------------------
print("\n5b. shape — knowing the picture before rendering it")

# Aspect ratio is a property of the graph, not of the renderer. A layered
# engine puts one dependency level in one row, so the widest level is the
# width. Measured on soleprint: 7 levels x 109 nodes rendered 14:1, and no
# layout flag helped — LR merely rotated it (1818x11183), ratio=compress
# squashed it to an unreadable 1008x75.
wide = {
    "meta": {"source": "python", "root": "x", "schema_version": "1"},
    "nodes": [{"id": f"n{i}", "kind": "module", "label": f"n{i}", "parent": None, "attrs": {}}
              for i in range(40)] +
             [{"id": "hub", "kind": "module", "label": "hub", "parent": None, "attrs": {}}],
    "edges": [{"source": f"n{i}", "target": "hub", "kind": "imports", "attrs": {}}
              for i in range(40)],
}
sh = ops_mod.shape(wide)
check("shape counts the widest level", sh["widest_level"] == 40, str(sh))
check("shape counts levels", sh["levels"] == 2, str(sh))
check("shape finds isolated nodes", ops_mod.shape(ir)["isolated"] >= 0)
check(
    "shape needs no renderer",
    isinstance(sh["nodes"], int) and "width" not in sh,
    "it is computed from the IR, so it can warn before writing a 375 KB image",
)

parts = ops_mod.split(ir)
check("split produces one document per subsystem", len(parts) >= 1, str(list(parts)))
check("each part still validates", all(check_ir(p) == [] for p in parts.values()))
check(
    "a crossing edge is kept on the side that reaches out",
    any(p["edges"] for p in parts.values()),
    "dropping it would understate the coupling",
)


# --------------------------------------------------------------------------
print("\n6. render")

if not dot_mod.have_graphviz():
    skip("render", "graphviz not installed — sudo apt install graphviz")
else:
    svg = dot_mod.render(dot_text)
    check("SVG comes out", svg.startswith(b"<?xml") and b"</svg>" in svg)
    check("the SVG is addressable", b'id="app.models.User"' in svg)
    check(
        "kinds survive as classes for CSS to reach",
        b'class="node class"' in svg and b'class="edge inherits"' in svg,
    )
    print_svg = dot_mod.render(print_text)
    check("the two themes really differ", print_svg != svg)

    # `fontname="Arial Bold"` passes through as a family no browser has, so it
    # renders neither Arial nor bold. Only the PostScript spelling produces a
    # real weight — and a correctly measured box to hold it.
    check(
        "labels are actually bold, not nominally bold",
        b'font-weight="bold"' in svg,
        "no font-weight in the SVG — the face name did not map to a weight",
    )
    check(
        "the font resolves to a real stack",
        b'font-family="Helvetica,sans-Serif"' in svg,
    )


# --------------------------------------------------------------------------
print("\n7. a second domain — the schema checkpoint")

# The brief makes this a checkpoint: if tables and foreign keys need the schema
# contorted, the schema is wrong and must be fixed before more extractors.
SCHEMA = {
    "models": {
        "Customer": {"doc": "A buyer.", "fields": {"id": {"type": "int", "pk": True},
                                                   "email": {"type": "str"}}},
        "Invoice": {"fields": {"id": {"type": "int", "pk": True},
                               "customer_id": {"type": "FK:Customer"},
                               "due_at": {"type": "datetime", "nullable": True}}},
        "Tag": {"fields": {"id": {"type": "int", "pk": True},
                           "invoices": {"type": "M2M:Invoice"}}},
        "Orphan": {"fields": {"ref": {"type": "FK:NotDefinedHere"}}},
    }
}
db_ir = db_ex.from_schema_dict(SCHEMA, root="fixture").to_dict()
db_ids = {n["id"] for n in db_ir["nodes"]}
db_edges = {(e["source"], e["target"], e["kind"]) for e in db_ir["edges"]}
db_kinds = {n["kind"] for n in db_ir["nodes"]}

check("the db IR validates", check_ir(db_ir) == [], str(check_ir(db_ir)[:3]))
check(
    "it needed no new top-level field",
    set(db_ir) == set(ir),
    f"{sorted(set(db_ir) ^ set(ir))} — the schema did not survive the second domain",
)
check("tables and columns are kinds, not fields", db_kinds >= {"table", "column"})
check(
    "a column is contained by its table",
    next(n for n in db_ir["nodes"] if n["id"] == "Invoice.customer_id")["parent"] == "Invoice",
)
check("a foreign key is an edge", ("Invoice", "Customer", "foreign_key") in db_edges)
check("a many-to-many is a different edge kind", ("Tag", "Invoice", "references") in db_edges)
check(
    "domain detail lives in attrs, not in new columns",
    next(n for n in db_ir["nodes"] if n["id"] == "Customer.id")["attrs"].get("pk") is True,
)
check(
    "a key naming a table the schema does not define survives as external",
    "NotDefinedHere" in db_ids
    and next(n for n in db_ir["nodes"] if n["id"] == "NotDefinedHere")["kind"] == "external",
)

# "Done means: adding a source type costs one adapter and every existing emitter
# works on it unchanged." This is that claim, executed.
db_md = index_mod.to_markdown(db_ir)
check("the index emitter needed no change for a new domain", "Customer" in db_md)
db_dot = dot_mod.emit(db_ir, lucid)
check("the dot emitter needed no change either", db_dot.startswith("digraph ir {"))
check(
    "an unstyled kind still renders, via `default`",
    lucid.node("table") == lucid.node("default"),
    "no `table` rule exists yet, and the diagram is still legible",
)
check(
    "the same view filters both domains",
    check_ir(ops_mod.only_kinds(db_ir, {"table"})) == [],
)


# --------------------------------------------------------------------------
print("\n8. the structure picks the drawing")

# The lesson this encodes: a diagram that fights its layout engine is usually
# the wrong kind of diagram. The same 24-table schema rendered 32034x136 through
# `dot` (235:1) and 1740x1860 through `erd` (0.9:1) — not because one engine is
# better, but because a schema is peer entities with references, and drawing it
# in dependency ranks was never its shape.
verdict_db = ops_mod.classify(db_ir)
check("a schema is classified as entities", verdict_db["kind"] == "erd", str(verdict_db))
check("...and routed to the card emitter", verdict_db["emitter"] == "erd")

verdict_code = ops_mod.classify(ops_mod.overview(ir))
check(
    "a small module graph still goes to ranks",
    verdict_code["emitter"] == "dot",
    str(verdict_code),
)

sheet = {
    "meta": {"source": "python", "root": "x", "schema_version": "1"},
    "nodes": [{"id": f"n{i}", "kind": "module", "label": f"n{i}", "parent": None, "attrs": {}}
              for i in range(40)] +
             [{"id": "hub", "kind": "module", "label": "hub", "parent": None, "attrs": {}}],
    "edges": [{"source": f"n{i}", "target": "hub", "kind": "imports", "attrs": {}}
              for i in range(40)],
}
verdict_sheet = ops_mod.classify(sheet)
check(
    "a 40-wide level is called a sheet, not a diagram",
    verdict_sheet["kind"] == "sheet" and verdict_sheet["emitter"] == "index",
    str(verdict_sheet),
)
check("every verdict explains itself", all(
    ops_mod.classify(g)["why"] for g in (db_ir, sheet, ir)
), "advice without a reason gets overridden")

# The property that makes the card layout work, and the one dot cannot offer.
svg = erd_mod.emit(db_ir, lucid)
import re as _re
m = _re.search(r'width="(\d+)pt" height="(\d+)pt"', svg)
w, h = int(m.group(1)), int(m.group(2))
check(
    f"the ER layout stays near-square ({w}x{h})",
    0.25 < w / h < 4,
    "sqrt(n) columns should bound the aspect ratio whatever the table count",
)

wide_schema = {
    "meta": {"source": "db", "root": "x", "schema_version": "1"},
    "nodes": [{"id": f"t{i}", "kind": "table", "label": f"t{i}", "parent": None, "attrs": {}}
              for i in range(60)] +
             [{"id": f"t{i}.c", "kind": "column", "label": "c", "parent": f"t{i}",
               "attrs": {"type": "int"}} for i in range(60)],
    "edges": [],
}
svg60 = erd_mod.emit(wide_schema, lucid)
m = _re.search(r'width="(\d+)pt" height="(\d+)pt"', svg60)
w60, h60 = int(m.group(1)), int(m.group(2))
check(
    f"...and still does at 60 tables ({w60}x{h60})",
    0.25 < w60 / h60 < 4,
    "this is the whole reason it is not a rank-based layout",
)
check(
    "the ER geometry is deterministic",
    erd_mod.emit(db_ir, lucid) == svg,
    "nothing here measures a font, so it renders identically on any machine",
)
check(
    "it draws no tables only when there are none",
    "no tables" in (
        _err(lambda: erd_mod.emit(ops_mod.overview(ir), lucid))
    ),
    "pointing it at a code graph should say so, not emit an empty canvas",
)


# --------------------------------------------------------------------------
print("\n9. the notebook is a build artifact, not a source file")

# The whole argument: a hand-confected notebook drifts from whatever it
# documents and there is no way to tell by looking. Generated from the spec the
# server is built from, "is this current" becomes a question about the build.
SPEC_IR = {
    "meta": {"source": "openapi", "root": "petstore.yaml", "schema_version": "1"},
    "nodes": [
        {"id": "Pet", "kind": "table", "label": "Pet", "parent": None, "attrs": {}},
        {"id": "Pet.id", "kind": "column", "label": "id", "parent": "Pet",
         "attrs": {"type": "int", "pk": True}},
        {"id": "Pet.name", "kind": "column", "label": "name", "parent": "Pet",
         "attrs": {"type": "str"}},
        {"id": "Pet.neutered", "kind": "column", "label": "neutered", "parent": "Pet",
         "attrs": {"type": "bool"}},
        {"id": "GET /pets", "kind": "endpoint", "label": "GET /pets", "parent": None,
         "attrs": {"method": "GET", "path": "/pets", "summary": "List pets",
                   "status": 200, "response_model": "Pet", "returns_list": True}},
        {"id": "POST /pets", "kind": "endpoint", "label": "POST /pets", "parent": None,
         "attrs": {"method": "POST", "path": "/pets", "summary": "Create a pet",
                   "status": 201, "request_model": "Pet", "response_model": "Pet"}},
    ],
    "edges": [{"source": "POST /pets", "target": "Pet", "kind": "accepts", "attrs": {}}],
}
check("the spec IR validates", check_ir(SPEC_IR) == [], str(check_ir(SPEC_IR)[:2]))

base = spec_mod.from_ir(SPEC_IR)
book_spec, _ = spec_mod.merge(base, None)
book = nb_mod.build(book_spec)
text = nb_mod.emit(book_spec)
check("it is nbformat 4", book["nbformat"] == 4 and book["nbformat_minor"] == 5)
check(
    "regenerating gives byte-identical bytes",
    nb_mod.emit(book_spec) == text,
    "a notebook that changes every build cannot be reviewed",
)
check("nothing has run in it", all(
    c["outputs"] == [] and c["execution_count"] is None
    for c in book["cells"] if c["cell_type"] == "code"
))
intro = "".join(book["cells"][0]["source"]).lower()
check(
    "it says not to edit it, and where to put changes instead",
    "not be edited" in intro and "overlay" in intro,
    f"the frame only holds if the artifact states it — got: {intro[:120]!r}",
)
body = "".join("".join(c["source"]) for c in book["cells"])
check("every endpoint in the spec has a section", "GET `/pets`" in body and "POST `/pets`" in body)
check("the facts come from the spec", "returns **Pet**" in body and "`201`" in body)
check(
    "no credential is written into it",
    "Bearer sk-" not in body and "password" not in body.lower(),
)
check(
    "it needs nothing installed",
    "pip install" not in body and "import requests" not in body,
)

# Compiling is not enough. `json.dumps` writes `false`/`true`/`null`, which are
# valid *identifiers* in Python — a generated body full of them compiles and
# then raises NameError on the first run. This is the check that caught it.
import io as _io
import contextlib as _ctx

ns, ran, failure = {}, 0, None
with _ctx.redirect_stdout(_io.StringIO()):
    for c in book["cells"]:
        if c["cell_type"] != "code":
            continue
        src = "".join(c["source"])
        if "urlopen" in src or "call(" in src and "def call" not in src:
            # Anything that would reach the network is compiled, not run.
            try:
                compile(src, c["id"], "exec")
                ran += 1
            except SyntaxError as e:
                failure = f"{c['id']}: {e}"
                break
            continue
        try:
            exec(compile(src, c["id"], "exec"), ns)
            ran += 1
        except Exception as e:  # noqa: BLE001 - any failure is the finding
            failure = f"{c['id']}: {type(e).__name__}: {e}"
            break
check(f"its cells run, not merely compile ({ran})", failure is None, failure or "")

literal = next(
    ("".join(c["source"]) for c in book["cells"]
     if c["cell_type"] == "code" and "BODY" in "".join(c["source"])), ""
)
check(
    "a generated body is a Python literal, not JSON",
    "False" in literal and "false" not in literal,
    f"got: {literal[:120]}",
)
check(
    "the primary key is left out of a create body",
    "'id'" not in literal,
    "the server assigns it",
)

# The chain, if modelgen is next door: spec -> IR -> notebook, nothing by hand.
try:
    oa = __import__(f"{PKG}.extractors.openapi", fromlist=["*"])
    spec = HERE.parent.parent / "station/tools/shuntgen/fixtures/petstore.yaml"
    if not spec.exists():
        raise ImportError("no petstore fixture")
    real = oa.extract(spec).to_dict()
except (ImportError, Exception) as e:  # noqa: BLE001
    skip("openapi -> IR", str(e).splitlines()[0][:60])
else:
    check("a real spec extracts", check_ir(real) == [], str(check_ir(real)[:2]))
    eps = [n for n in real["nodes"] if n["kind"] == "endpoint"]
    check(f"its endpoints become nodes ({len(eps)})", len(eps) >= 3)
    check(
        "its schemas reuse the db vocabulary",
        {"table", "column"} <= {n["kind"] for n in real["nodes"]},
        "so the ER emitter draws an API's data model without knowing it is one",
    )
    check(
        "and the ER emitter does draw it",
        erd_mod.emit(ops_mod.only_kinds(real, {"table"}), lucid).startswith("<?xml"),
    )


# --------------------------------------------------------------------------
print("\n8b. usage — what a spec cannot say")

# A spec is a set; usage is a sequence. Everything asserted here is something no
# amount of reading the OpenAPI document harder would recover.
def _entry(t, method, url, status, body=None):
    req = {"method": method, "url": url,
           "headers": [{"name": "Authorization", "value": "Bearer LEAK-ME-IF-BROKEN"}]}
    if body is not None:
        req["postData"] = {"mimeType": "application/json", "text": json.dumps(body)}
    return {"startedDateTime": f"2026-01-01T00:00:{t:02d}Z",
            "request": req, "response": {"status": status, "content": {}}}


HAR = {"log": {"version": "1.2", "entries": [
    _entry(1, "POST", "https://x.test/auth", 200, {"user": "a", "secret": "b"}),
    _entry(2, "GET", "https://x.test/pets?status=available&limit=20", 200),
    _entry(3, "GET", "https://x.test/pets/1042", 200),
    _entry(4, "GET", "https://x.test/pets/7f3e9b2a-1c4d-4a5b-9e8f-0a1b2c3d4e5f", 200),
    _entry(5, "POST", "https://x.test/graphql", 200,
           {"operationName": "PetWithTags", "query": "query PetWithTags($id: ID!) { pet }",
            "variables": {"id": "1042"}}),
    _entry(6, "POST", "https://x.test/pets", 422, {"name": "Rex"}),
    _entry(7, "POST", "https://x.test/pets", 201, {"name": "Rex", "status": "available"}),
    _entry(8, "GET", "https://x.test/pets?status=available&limit=20", 200),
]}}
har_path = ROOT.parent / "session.har"
har_path.write_text(json.dumps(HAR))
use = usage_ex.extract(har_path).to_dict()
by = {n["id"]: n for n in use["nodes"]}
raw_json = json.dumps(use)

check("the usage IR validates", check_ir(use) == [], str(check_ir(use)[:2]))
check(
    "**no credential reaches the IR**",
    "LEAK-ME-IF-BROKEN" not in raw_json and "Authorization" not in raw_json,
    "a HAR is full of live tokens and a document gets committed",
)
check(
    "paths are templated into one route",
    "GET /pets/{id}" in by and "GET /pets/1042" not in by,
)
check(
    "...and both id formats stay on that one route",
    by["GET /pets/{id}"]["attrs"]["id_formats"] == ["numeric", "uuid"]
    and by["GET /pets/{id}"]["attrs"]["calls"] == 2,
    "splitting by format invents an endpoint that does not exist",
)
check(
    "what is always sent is distinguished from what is sometimes",
    by["GET /pets"]["attrs"]["params_always"] == ["limit", "status"],
    "a spec calls these optional; traffic says otherwise",
)
check(
    "the statuses that really happen are recorded",
    by["POST /pets"]["attrs"]["statuses"] == [201, 422],
    "the 422 everyone hits is not in the spec",
)
check(
    "a body field seen once out of twice is not marked always",
    [f["name"] for f in by["POST /pets"]["attrs"]["body_fields"] if f["always"]] == ["name"],
)
check(
    "a GraphQL operation is found and named",
    "query PetWithTags" in by and by["query PetWithTags"]["kind"] == "operation",
    "one endpoint carries many operations; the path alone says nothing",
)
check(
    "only the shape of a payload is kept, never a value",
    '"a"' not in raw_json and '"b"' not in raw_json,
)
seq = {(e["source"], e["target"]) for e in use["edges"]}
check(
    "the sequence is recorded as edges",
    ("POST /auth", "GET /pets") in seq and all(e["kind"] == "follows" for e in use["edges"]),
    "this is the half a spec structurally cannot contain",
)
check(
    "a repeated call is not a step in the sequence",
    not [e for e in use["edges"] if e["source"] == e["target"]],
)
check(
    "usage is a pipeline, and drawn as one",
    ops_mod.classify(use)["kind"] == "pipeline",
)

# The payoff: a walkthrough in the order people actually call things.
u_spec, _ = spec_mod.merge(spec_mod.from_ir(use), None)
ids = [st["id"] for st in u_spec["steps"] if st["kind"] == "call"]
check(
    "the notebook keeps the observed order",
    ids[0] == "POST /auth" and ids.index("GET /pets") < ids.index("POST /pets"),
    f"got {ids}",
)
u_body = "".join("".join(c["source"]) for c in nb_mod.build(u_spec)["cells"])
check(
    "it sends the parameters that are always sent",
    "PARAMS = " in u_body and "'status'" in u_body,
)
check(
    "a templated path becomes a variable, not a literal",
    'ID = "<id>"' in u_body and '"/pets/{id}"' not in u_body,
    "a literal {id} in the URL would 404",
)
check("the GraphQL operation gets a query cell", "QUERY = " in u_body)
har_path.unlink()


# --------------------------------------------------------------------------
print("\n9b. generated base + hand-written overlay")

# The arrangement that gets both halves: generation alone says nothing a parser
# could not work out; hand-authoring alone rots. The base is never edited, the
# overlay is the only file anyone touches, and it is re-applied every build.
OVERLAY = {
    "steps": {
        "GET /pets": {
            "note": "always called with status=available first",
            "code": 'show(call("GET", "/pets", params={"status": "available"}))',
        },
        "setup": {"_kind": "params", "_title": ""},     # scaffold hints, not edits
    },
    "insert": [
        {"id": "graphql", "after": "client", "kind": "md",
         "title": "The GraphQL endpoint",
         "text": "Not in the OpenAPI document at all."},
    ],
    "drop": ["shapes"],
}

merged, drift = spec_mod.merge(spec_mod.from_ir(SPEC_IR), OVERLAY)
ids = [st["id"] for st in merged["steps"]]
body_o = "".join("".join(c["source"]) for c in nb_mod.build(merged)["cells"])

check("no drift against a matching base", drift == [], str(drift))
check("an inserted step lands where it was asked to", ids.index("graphql") == ids.index("client") + 1)
check("a dropped step is gone", "shapes" not in ids)
check(
    "replaced code wins over the generated call",
    'params={"status": "available"}' in body_o,
    "this is how real usage gets in, and a spec cannot supply it",
)
check("an annotation reaches the prose", "always called with status=available" in body_o)
check(
    "scaffold hints are not mistaken for edits",
    "_kind" not in body_o and "_title" not in body_o,
)

# The two properties the whole arrangement rests on.
again, _ = spec_mod.merge(spec_mod.from_ir(SPEC_IR), OVERLAY)
check(
    "regenerating re-applies the overlay, byte for byte",
    nb_mod.emit(again) == nb_mod.emit(merged),
    "a regeneration that loses someone's work will not be run twice",
)

moved = dict(SPEC_IR, nodes=[n for n in SPEC_IR["nodes"] if n["id"] != "GET /pets"])
shrunk, drift2 = spec_mod.merge(spec_mod.from_ir(moved), OVERLAY)
check(
    "when the base moves, the overlay says so",
    any("GET /pets" in d for d in drift2),
    "silently dropping it is how an overlay goes stale without anyone noticing",
)
check("...and a document is still produced", len(shrunk["steps"]) > 1)

bare, drift3 = spec_mod.merge(spec_mod.from_ir(SPEC_IR), None)
check(
    "extraction works with the overlay absent",
    drift3 == [] and len(bare["steps"]) > 1,
    "the overlay is an addition, never a dependency",
)

blank = spec_mod.scaffold(spec_mod.from_ir(SPEC_IR))
check(
    "a scaffold hands over every step id",
    set(blank["steps"]) == {st["id"] for st in spec_mod.from_ir(SPEC_IR)["steps"]},
    "listing the ids is the difference between an overlay being written and meant to be",
)
check(
    "a blank scaffold changes nothing",
    nb_mod.emit(spec_mod.merge(spec_mod.from_ir(SPEC_IR), blank)[0]) == nb_mod.emit(bare),
)


# --------------------------------------------------------------------------
print("\n10. defaults")

check(
    "dark is the default theme",
    Style.load("lucid").theme == "dark",
    "a diagram lands in a dark docs page far more often than in print",
)
check("print is one flag away", Style.load("lucid", theme="lucid").theme == "lucid")

pipeline = {
    "meta": {"source": "airflow", "root": "etl", "schema_version": "1"},
    "nodes": [{"id": x, "kind": "task", "label": x, "parent": None, "attrs": {}}
              for x in ("extract", "transform", "load", "notify")],
    "edges": [{"source": a, "target": b, "kind": "depends", "attrs": {}}
              for a, b in (("extract", "transform"), ("transform", "load"),
                           ("load", "notify"))],
}
verdict = ops_mod.classify(pipeline)
check(
    "a DAG is recognised as a pipeline",
    verdict["kind"] == "pipeline",
    str(verdict),
)
check(
    "...and is drawn left to right",
    verdict.get("options", {}).get("rankdir") == "LR",
    "a sequence reads across, which is how every scheduler's own UI draws it",
)
check(
    "a task has its own style, not a module's",
    lucid.node("task") != lucid.node("module"),
    "a schedule should not render as if it were code structure",
)
check(
    "the emitter honours the rankdir it is given",
    "rankdir=LR" in dot_mod.emit(pipeline, lucid, rankdir="LR"),
)
check("every verdict carries options", "options" in ops_mod.classify(ir))


# --------------------------------------------------------------------------
tmp.cleanup()
print()
print(f"{len(PASS)} passed, {len(FAIL)} failed, {len(SKIP)} skipped")
if FAIL:
    print("\nfailed:")
    for name in FAIL:
        print(f"  {name}")
sys.exit(1 if FAIL else 0)