313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""
|
|
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 re
|
|
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__}")
|
|
|
|
problems.extend(_check_larder(data["meta"].get("larder"), meta_schema))
|
|
|
|
# -- 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
|
|
|
|
|
|
# A credential that survived redaction, detected independently of the code that
|
|
# does the redacting. `book/larder.py` has its own key list; this has a second
|
|
# one on purpose. If the two ever disagree the check fires, which is the point —
|
|
# a scrubber that is graded by its own word is not graded. Same reason
|
|
# VISUAL_KEYS lives here and not in schema.json.
|
|
_DSN_PASSWORD = re.compile(r"[a-zA-Z][a-zA-Z0-9+.\-]*://[^:/@\s]+:(?P<secret>[^@/\s]+)@")
|
|
_SECRET_PARAM = re.compile(
|
|
r"(?i)\b(?:password|passwd|pwd|secret|token|access_token|refresh_token"
|
|
r"|api_key|apikey|sig|signature|credentials)\s*=\s*(?P<secret>[^&;\s]+)"
|
|
)
|
|
_MASKED = {"***", "xxx", "redacted", "[redacted]", "masked"}
|
|
|
|
|
|
def _check_larder(larder, meta_schema: dict) -> list[str]:
|
|
"""The input measure, checked — shape, arithmetic, and no leaked secret.
|
|
|
|
Absent is fine and means "not measured". Present and wrong is not, because
|
|
the whole value of the measure is that the number can be trusted against the
|
|
result; a larder claiming 47 reads when it read 45 is worse than no larder.
|
|
"""
|
|
if larder is None:
|
|
return []
|
|
if not isinstance(larder, dict):
|
|
return [f"meta.larder must be an object, got {type(larder).__name__}"]
|
|
|
|
problems = []
|
|
spec = meta_schema["properties"]["larder"]
|
|
for key in spec["required"]:
|
|
if key not in larder:
|
|
problems.append(f"meta.larder is missing {key!r}")
|
|
for key in larder:
|
|
if key not in spec["properties"]:
|
|
problems.append(f"meta.larder has unknown key {key!r}")
|
|
|
|
failed = larder.get("failed")
|
|
if failed is not None and not isinstance(failed, list):
|
|
problems.append("meta.larder.failed must be an array")
|
|
failed = None
|
|
elif failed:
|
|
for i, f in enumerate(failed):
|
|
if not isinstance(f, dict) or "name" not in f or "error" not in f:
|
|
problems.append(f"meta.larder.failed[{i}] needs both 'name' and 'error'")
|
|
|
|
# read is derived. Stored, it can disagree with itself, and this is where
|
|
# that shows up rather than in a book measure nobody can reconcile.
|
|
seen, read = larder.get("seen"), larder.get("read")
|
|
if isinstance(seen, int) and isinstance(read, int) and isinstance(failed, list):
|
|
if read != max(0, seen - len(failed)):
|
|
problems.append(
|
|
f"meta.larder.read is {read} but seen={seen} with {len(failed)} failed "
|
|
f"implies {max(0, seen - len(failed))} — the measure disagrees with itself"
|
|
)
|
|
if isinstance(read, int) and isinstance(seen, int) and read > seen:
|
|
problems.append(f"meta.larder.read ({read}) exceeds seen ({seen})")
|
|
|
|
unit = larder.get("unit")
|
|
if unit is not None and not isinstance(unit, str):
|
|
problems.append(f"meta.larder.unit must be a string, got {type(unit).__name__}")
|
|
|
|
identity = larder.get("identity")
|
|
if isinstance(identity, str):
|
|
for pattern, what in ((_DSN_PASSWORD, "a DSN password"),
|
|
(_SECRET_PARAM, "a secret parameter")):
|
|
m = pattern.search(identity)
|
|
if m and m.group("secret").lower() not in _MASKED:
|
|
problems.append(
|
|
f"meta.larder.identity still contains {what} — "
|
|
"redact() in book/larder.py did not mask it, and this document "
|
|
"must not be written anywhere"
|
|
)
|
|
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
|