docgen iter 2
This commit is contained in:
236
soleprint/atlas2/docgen/ir/validate.py
Normal file
236
soleprint/atlas2/docgen/ir/validate.py
Normal file
@@ -0,0 +1,236 @@
|
||||
"""
|
||||
Check an IR document at the boundary.
|
||||
|
||||
Called wherever the IR crosses between layers: after an extractor writes one,
|
||||
before an emitter reads one. That is the whole point of having one format — the
|
||||
check is in one place instead of every consumer defending itself.
|
||||
|
||||
python3 -m docgen.ir path/to/ir.json
|
||||
|
||||
Stdlib only. This is not a general JSON Schema engine and does not want to be;
|
||||
it reads the field lists **out of `schema.json`** so the two cannot drift, then
|
||||
checks the handful of things that actually go wrong. `jsonschema` would validate
|
||||
the shape and still miss every item below the first section, which are the ones
|
||||
that produce a broken diagram.
|
||||
|
||||
## What it catches that a schema cannot
|
||||
|
||||
- an edge naming a node that does not exist
|
||||
- a `parent` naming a node that does not exist, or a containment cycle
|
||||
- duplicate ids
|
||||
- **a visual field smuggled into the IR** — the architectural rule, as a check
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
SCHEMA_PATH = HERE / "schema.json"
|
||||
|
||||
# Not "fields we forgot to allow" — fields that mean the layering has broken.
|
||||
# A colour in the IR means an extractor decided what something looks like, and
|
||||
# from then on the graph renders one way forever. Named here rather than in
|
||||
# schema.json because a schema can say a key is disallowed but not why.
|
||||
VISUAL_KEYS = {
|
||||
"color", "colour", "fill", "fillcolor", "bgcolor", "background",
|
||||
"shape", "style", "penwidth", "stroke", "width", "height",
|
||||
"font", "fontname", "fontsize", "fontcolor",
|
||||
"pos", "x", "y", "rank", "layout", "theme", "class", "cls",
|
||||
}
|
||||
|
||||
|
||||
class IRError(ValueError):
|
||||
"""An IR document that a consumer cannot safely read."""
|
||||
|
||||
|
||||
def _schema() -> dict:
|
||||
return json.loads(SCHEMA_PATH.read_text())
|
||||
|
||||
|
||||
def _props(schema: dict, *path: str) -> dict:
|
||||
node = schema
|
||||
for step in path:
|
||||
node = node["properties"][step] if "properties" in node else node[step]
|
||||
return node
|
||||
|
||||
|
||||
def _fields(schema: dict, section: str) -> tuple[set, set]:
|
||||
"""(required, allowed) for `nodes`/`edges` items, read from the schema."""
|
||||
item = schema["properties"][section]["items"]
|
||||
return set(item.get("required", [])), set(item.get("properties", {}))
|
||||
|
||||
|
||||
def check(data: dict, *, strict_visual: bool = True) -> list[str]:
|
||||
"""Every problem, as a list. Empty means the document is sound.
|
||||
|
||||
A list rather than raising on the first: an extractor with four dangling
|
||||
edges should report four, not make you run it four times.
|
||||
"""
|
||||
problems: list[str] = []
|
||||
schema = _schema()
|
||||
|
||||
# -- top level --------------------------------------------------------
|
||||
for key in schema["required"]:
|
||||
if key not in data:
|
||||
problems.append(f"missing top-level {key!r}")
|
||||
if problems:
|
||||
return problems # nothing below is meaningful without these
|
||||
|
||||
meta_schema = schema["properties"]["meta"]
|
||||
for key in meta_schema["required"]:
|
||||
if key not in data["meta"]:
|
||||
problems.append(f"meta is missing {key!r}")
|
||||
for key in data["meta"]:
|
||||
if key not in meta_schema["properties"]:
|
||||
problems.append(f"meta has unknown key {key!r}")
|
||||
|
||||
version = data["meta"].get("schema_version")
|
||||
expected = _props(schema, "meta")["properties"]["schema_version"]
|
||||
if version is not None and not isinstance(version, str):
|
||||
problems.append(f"meta.schema_version must be a string, got {type(version).__name__}")
|
||||
|
||||
# -- nodes ------------------------------------------------------------
|
||||
node_required, node_allowed = _fields(schema, "nodes")
|
||||
seen: set[str] = set()
|
||||
for i, n in enumerate(data["nodes"]):
|
||||
where = f"nodes[{i}]"
|
||||
if not isinstance(n, dict):
|
||||
problems.append(f"{where} is not an object")
|
||||
continue
|
||||
for key in node_required - set(n):
|
||||
problems.append(f"{where} is missing {key!r}")
|
||||
for key in set(n) - node_allowed:
|
||||
problems.append(f"{where} has unknown key {key!r}")
|
||||
nid = n.get("id")
|
||||
if isinstance(nid, str):
|
||||
if nid in seen:
|
||||
problems.append(f"{where} id {nid!r} is declared more than once")
|
||||
seen.add(nid)
|
||||
|
||||
# -- edges ------------------------------------------------------------
|
||||
edge_required, edge_allowed = _fields(schema, "edges")
|
||||
for i, e in enumerate(data["edges"]):
|
||||
where = f"edges[{i}]"
|
||||
if not isinstance(e, dict):
|
||||
problems.append(f"{where} is not an object")
|
||||
continue
|
||||
for key in edge_required - set(e):
|
||||
problems.append(f"{where} is missing {key!r}")
|
||||
for key in set(e) - edge_allowed:
|
||||
problems.append(f"{where} has unknown key {key!r}")
|
||||
for end in ("source", "target"):
|
||||
ref = e.get(end)
|
||||
if isinstance(ref, str) and ref not in seen:
|
||||
problems.append(f"{where} {end} names unknown node {ref!r}")
|
||||
|
||||
# -- containment ------------------------------------------------------
|
||||
parents = {
|
||||
n["id"]: n.get("parent")
|
||||
for n in data["nodes"]
|
||||
if isinstance(n, dict) and isinstance(n.get("id"), str)
|
||||
}
|
||||
for nid, parent in parents.items():
|
||||
if parent is None:
|
||||
continue
|
||||
if parent not in seen:
|
||||
problems.append(f"node {nid!r} has parent {parent!r}, which is not a node")
|
||||
continue
|
||||
# A containment cycle makes any tree walk non-terminating, and the index
|
||||
# emitter is a tree walk.
|
||||
slow, fast = nid, parent
|
||||
while fast is not None and fast in parents:
|
||||
if slow == fast:
|
||||
problems.append(f"node {nid!r} is in a containment cycle")
|
||||
break
|
||||
fast = parents[fast]
|
||||
if fast is None or fast not in parents:
|
||||
break
|
||||
fast = parents[fast]
|
||||
slow = parents[slow]
|
||||
|
||||
# -- the architectural rule -------------------------------------------
|
||||
if strict_visual:
|
||||
for i, n in enumerate(data["nodes"]):
|
||||
if isinstance(n, dict):
|
||||
for key in set(n.get("attrs") or {}) & VISUAL_KEYS:
|
||||
problems.append(
|
||||
f"nodes[{i}].attrs.{key!r} is a visual field — "
|
||||
"style belongs in style/*.json keyed on `kind`, not in the IR"
|
||||
)
|
||||
for i, e in enumerate(data["edges"]):
|
||||
if isinstance(e, dict):
|
||||
for key in set(e.get("attrs") or {}) & VISUAL_KEYS:
|
||||
problems.append(
|
||||
f"edges[{i}].attrs.{key!r} is a visual field — "
|
||||
"style belongs in style/*.json keyed on `kind`, not in the IR"
|
||||
)
|
||||
|
||||
return problems
|
||||
|
||||
|
||||
def validate(data: dict, **kw) -> dict:
|
||||
"""check(), but raises. For use at a boundary where carrying on is wrong."""
|
||||
problems = check(data, **kw)
|
||||
if problems:
|
||||
raise IRError(f"{len(problems)} problem(s):\n " + "\n ".join(problems))
|
||||
return data
|
||||
|
||||
|
||||
def check_model_matches_schema() -> list[str]:
|
||||
"""The sanctioned duplication, asserted.
|
||||
|
||||
`model.py` mirrors `schema.json` by hand. This is what stops the two from
|
||||
drifting: every field the schema declares must exist on the dataclass, and
|
||||
every dataclass field must be declared in the schema.
|
||||
"""
|
||||
from .model import Edge, Meta, Node
|
||||
|
||||
schema = _schema()
|
||||
problems = []
|
||||
pairs = [
|
||||
("meta", Meta, set(schema["properties"]["meta"]["properties"])),
|
||||
("nodes", Node, _fields(schema, "nodes")[1]),
|
||||
("edges", Edge, _fields(schema, "edges")[1]),
|
||||
]
|
||||
for name, cls, declared in pairs:
|
||||
actual = set(cls.__dataclass_fields__)
|
||||
for missing in declared - actual:
|
||||
problems.append(f"schema declares {name}.{missing!r}; {cls.__name__} has no such field")
|
||||
for extra in actual - declared:
|
||||
problems.append(f"{cls.__name__} has {extra!r}; schema does not declare it")
|
||||
return problems
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
argv = sys.argv[1:] if argv is None else argv
|
||||
if not argv:
|
||||
print("usage: python3 -m docgen.ir <ir.json>", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
drift = check_model_matches_schema()
|
||||
if drift:
|
||||
print("model.py and schema.json disagree:", file=sys.stderr)
|
||||
for d in drift:
|
||||
print(f" {d}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
path = Path(argv[0])
|
||||
try:
|
||||
data = json.loads(path.read_text())
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
print(f"Error: could not read {path}: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
problems = check(data)
|
||||
if problems:
|
||||
print(f"{path}: {len(problems)} problem(s)", file=sys.stderr)
|
||||
for p in problems:
|
||||
print(f" {p}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print(
|
||||
f"{path}: ok — {len(data['nodes'])} nodes, {len(data['edges'])} edges, "
|
||||
f"schema v{data['meta'].get('schema_version')}"
|
||||
)
|
||||
return 0
|
||||
Reference in New Issue
Block a user