#!/usr/bin/env python3
"""Structural verification for the draw.io import + output-spec work.
Drives the real shipped artifacts (drawio_extract.py, import-drawio.md,
output-spec.md, SKILL.md, the slash command) against the checked-in fixture.
Exit 0 only when all gates pass. Intended for local/PR checks alongside:
python scripts/lint-skin.py --all --baseline
"""
from __future__ import annotations
import base64
import importlib.util
import json
import re
import struct
import subprocess
import sys
import tempfile
import zlib
from pathlib import Path
from urllib.parse import quote
sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parent.parent
SKILL = ROOT / "skills/diagram-design/SKILL.md"
EXTRACT = ROOT / "skills/diagram-design/scripts/drawio_extract.py"
IMPORT_REF = ROOT / "skills/diagram-design/references/import-drawio.md"
OUTPUT_REF = ROOT / "skills/diagram-design/references/output-spec.md"
EXPORT_REF = ROOT / "skills/diagram-design/references/export.md"
COMMAND = ROOT / "commands/import-drawio.md"
FIXTURE = ROOT / "scripts/fixtures/sample-architecture.drawio"
EXAMPLE = ROOT / "skills/diagram-design/assets/example-import-drawio.html"
def fail(msg: str) -> None:
print(f"FAIL: {msg}", file=sys.stderr)
raise SystemExit(1)
def ok(msg: str) -> None:
print(f"OK: {msg}")
def run_extract(args: list[str]) -> str:
proc = subprocess.run(
[sys.executable, str(EXTRACT), *args],
capture_output=True,
text=True,
)
if proc.returncode != 0:
fail(f"extractor exited {proc.returncode} for {args}: {proc.stderr.strip()}")
return proc.stdout
def load_extractor_module():
spec = importlib.util.spec_from_file_location("diagram_design_drawio_extract", EXTRACT)
if spec is None or spec.loader is None:
fail("could not load drawio extractor module")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
def compress_model(model: str) -> str:
compressor = zlib.compressobj(9, zlib.DEFLATED, -15)
payload = quote(model, safe="")
raw = compressor.compress(payload.encode()) + compressor.flush()
return base64.b64encode(raw).decode()
def png_chunk(ctype: bytes, data: bytes) -> bytes:
return (
struct.pack(">I", len(data))
+ ctype
+ data
+ struct.pack(">I", zlib.crc32(ctype + data) & 0xFFFFFFFF)
)
def check_files() -> None:
for path in (SKILL, EXTRACT, IMPORT_REF, OUTPUT_REF, EXPORT_REF, COMMAND, FIXTURE, EXAMPLE):
if not path.is_file():
fail(f"missing {path.relative_to(ROOT)}")
ok("all import artifacts present")
def check_parse_raw() -> dict:
payload = json.loads(run_extract([str(FIXTURE), "--json"]))
page = payload["pages"][0]
analysis = page["analysis"]
nodes = {n["id"]: n for n in page["nodes"]}
if analysis["nodes_total"] != 12 or analysis["edges_total"] != 8:
fail(f"fixture graph mis-parsed: {analysis['nodes_total']}n/{analysis['edges_total']}e")
if analysis["containers"] != 2:
fail(f"expected 2 containers, got {analysis['containers']}")
# Child geometry must be resolved to absolute coordinates (parent 360 + 20).
if (nodes["gw"]["x"], nodes["gw"]["y"]) != (380.0, 80.0):
fail(f"absolute geometry wrong for gw: {nodes['gw']['x']},{nodes['gw']['y']}")
if nodes["gw"]["depth"] != 1:
fail("container depth not tracked")
# HTML labels flattened, entities decoded, preserved as a line break.
if nodes["web"]["label"] != "Web App\nnext.js":
fail(f"label not cleaned: {nodes['web']['label']!r}")
# Shape classification across style keys, shape= values, and stencil families.
expected_shapes = {
"zone-edge": "swimlane",
"check": "rhombus",
"pg": "cylinder",
"s3": "icon:aws",
"note1": "note",
}
for nid, shape in expected_shapes.items():
if nodes[nid]["shape"] != shape:
fail(f"{nid}: expected shape {shape}, got {nodes[nid]['shape']}")
# Degree counts feed focal-node selection.
if (nodes["gw"]["in_degree"], nodes["gw"]["out_degree"]) != (2, 2):
fail("degree counting wrong for the hub node")
if not analysis["hubs"] or analysis["hubs"][0]["id"] != "gw":
fail("hub ranking did not surface the gateway")
# An edge label carried by a child edgeLabel cell must fold into the edge.
e1 = next(e for e in page["edges"] if e["id"] == "e1")
if e1["label"] != "HTTPS":
fail(f"edgeLabel child not folded into edge: {e1['label']!r}")
if not next(e for e in page["edges"] if e["id"] == "e7")["dashed"]:
fail("dashed edge style not detected")
if any(e["source"] is None or e["target"] is None for e in page["edges"]):
fail("edge endpoints lost")
# Container zones are not lanes: the swimlane candidate must not fire here.
if analysis["type_candidates"][0] == "swimlane":
fail("unaligned container zones mis-detected as a swimlane")
if not analysis["over_node_budget"]:
fail("12-node source should flag the 9-node budget")
if not analysis["collapsible_groups"]:
fail("collapsible groups not reported")
ok("raw XML fixture parses: geometry, labels, shapes, degrees, edges")
return payload
def check_containers(tmp: Path) -> None:
model = re.search(
r"", FIXTURE.read_text(encoding="utf-8"), re.S
)
if not model:
fail("fixture has no mxGraphModel")
encoded = compress_model(model.group(0))
mxfile = (
''
f'{encoded}'
f'{encoded}'
""
)
compressed = tmp / "compressed.drawio"
compressed.write_text(mxfile, encoding="utf-8")
png = tmp / "embedded.drawio.png"
png.write_bytes(
b"\x89PNG\r\n\x1a\n"
+ png_chunk(b"IHDR", struct.pack(">IIBBBBB", 1, 1, 8, 6, 0, 0, 0))
+ png_chunk(b"tEXt", b"mxfile\x00" + quote(mxfile, safe="").encode())
+ png_chunk(b"IDAT", zlib.compress(b"\x00\x00\x00\x00\x00"))
+ png_chunk(b"IEND", b"")
)
svg = tmp / "embedded.drawio.svg"
svg.write_text(
'',
encoding="utf-8",
)
decoy_svg = tmp / "embedded-with-decoy.drawio.svg"
escaped_mxfile = (
mxfile.replace("&", "&").replace("<", "<").replace('"', """)
)
decoy_svg.write_text(
'",
encoding="utf-8",
)
for path in (compressed, png, svg, decoy_svg):
payload = json.loads(run_extract([str(path), "--json"]))
analysis = payload["pages"][0]["analysis"]
if analysis["nodes_total"] != 12 or analysis["edges_total"] != 8:
fail(f"{path.name}: decoded graph differs from the raw fixture")
ok("deflate+base64, PNG-embedded, and SVG-embedded containers all decode")
# Page selection.
digest = run_extract([str(compressed)])
if "2 page(s)" not in digest:
fail("multi-page header not reported")
if "## Page 1" in digest:
fail("default run must emit only the first page")
if "## Page 1" not in run_extract([str(compressed), "--page", "all"]):
fail("--page all did not emit both pages")
if "## Page 1" not in run_extract([str(compressed), "--page", "Second"]):
fail("--page by name failed")
ok("page selection: default, --page all, --page by name")
bad = tmp / "notadiagram.txt"
bad.write_text("hello", encoding="utf-8")
proc = subprocess.run(
[sys.executable, str(EXTRACT), str(bad)], capture_output=True, text=True
)
if proc.returncode != 2 or "not a draw.io file" not in proc.stderr:
fail("non-draw.io input must exit 2 with a clear message")
ok("unsupported input exits 2 with a diagnostic")
def check_security_and_limits(tmp: Path) -> None:
extractor = load_extractor_module()
compressor = zlib.compressobj(9, zlib.DEFLATED, -15)
compressed = compressor.compress(b"A" * 4096) + compressor.flush()
try:
extractor._decompress_limited(compressed, -15, limit=1024)
except extractor.PayloadTooLarge:
pass
else:
fail("bounded decompression accepted a payload above its output limit")
dtd = tmp / "entity.drawio"
dtd.write_text(
']>'
'',
encoding="utf-8",
)
proc = subprocess.run(
[sys.executable, str(EXTRACT), str(dtd)], capture_output=True, text=True
)
if proc.returncode != 2 or "DTD and entity declarations" not in proc.stderr:
fail("DTD/entity input must be rejected with a clear diagnostic")
compressed_dtd = tmp / "compressed-entity.drawio"
compressed_dtd.write_text(
''
+ compress_model(
']>'
''
)
+ "",
encoding="utf-8",
)
proc = subprocess.run(
[sys.executable, str(EXTRACT), str(compressed_dtd)],
capture_output=True,
text=True,
)
if proc.returncode != 2 or "DTD and entity declarations" not in proc.stderr:
fail("DTD/entity declarations in compressed pages must be rejected")
truncated_png = tmp / "truncated.drawio.png"
truncated_png.write_bytes(b"\x89PNG\r\n\x1a\n" + struct.pack(">I", 100) + b"tEXt")
proc = subprocess.run(
[sys.executable, str(EXTRACT), str(truncated_png)],
capture_output=True,
text=True,
)
if proc.returncode != 2 or "truncated metadata chunk" not in proc.stderr:
fail("truncated PNG metadata must be rejected with a clear diagnostic")
proc = subprocess.run(
[sys.executable, str(EXTRACT), str(FIXTURE), "--max-rows", "0"],
capture_output=True,
text=True,
)
if proc.returncode != 2 or "--max-rows must be at least 1" not in proc.stderr:
fail("non-positive --max-rows must be rejected")
ok("resource limits, DTD rejection, PNG bounds, and CLI validation")
def check_docs() -> None:
import_text = IMPORT_REF.read_text(encoding="utf-8")
for needle in (
"drawio_extract.py",
"output-spec.md",
"## Step 1 — Extract the IR",
"## Step 3 — Pick the target type",
"Multi-page files",
"## Anti-patterns",
"fidelity ledger",
"untrusted data",
"--page all",
):
if needle not in import_text:
fail(f"import-drawio.md missing {needle!r}")
for container in (".drawio.png", ".drawio.svg"):
if container not in import_text:
fail(f"import-drawio.md does not cover {container}")
output_text = OUTPUT_REF.read_text(encoding="utf-8")
for preset in (
"doc-inline",
"doc-wide",
"slide-16x9",
"slide-4x3",
"social-og",
"social-square",
"print-a4-landscape",
"print-letter-landscape",
"`fit`",
):
if preset not in output_text:
fail(f"output-spec.md missing size preset {preset}")
for level in ("faithful", "balanced", "simplified"):
if level not in output_text:
fail(f"output-spec.md missing detail level {level}")
for audience in ("engineer", "mixed", "executive"):
if audience not in output_text:
fail(f"output-spec.md missing audience level {audience}")
for needle in ("Degrade ladder", "Fidelity ledger", "Type ramp", "Non-Latin labels"):
if needle not in output_text:
fail(f"output-spec.md missing section {needle!r}")
# Every viewBox preset must respect the 4px grid rule (SKILL.md §7).
for w, h in re.findall(r"`0 0 (\d+) (\d+)`", output_text):
if int(w) % 4 or int(h) % 4:
fail(f"viewBox preset {w}×{h} is off the 4px grid")
if "Sizing the export" not in EXPORT_REF.read_text(encoding="utf-8"):
fail("export.md missing the sizing section")
skill_text = SKILL.read_text(encoding="utf-8")
for needle in (
"## 11. Importing an Existing Diagram (draw.io)",
"## 12. Output",
"references/import-drawio.md",
"references/output-spec.md",
"drawio_extract.py",
):
if needle not in skill_text:
fail(f"SKILL.md missing {needle!r}")
if ".drawio" not in skill_text.split("---")[1]:
fail("SKILL.md frontmatter description does not mention draw.io import")
command_text = COMMAND.read_text(encoding="utf-8")
for flag in ("--format", "--size", "--detail", "--audience", "--page", "--variant"):
if flag not in command_text:
fail(f"import command missing {flag}")
if "example-import-drawio.html" not in import_text:
fail("import-drawio.md does not point at the worked example")
example = EXAMPLE.read_text(encoding="utf-8")
if 'viewBox="0 0 960 600"' not in example:
fail("worked example does not use the doc-inline viewBox")
if example.count("#eb6c36") > 4:
fail("worked example uses the accent on more than the focal node + legend")
proc = subprocess.run(
[sys.executable, str(ROOT / "scripts/lint-skin.py"), str(EXAMPLE)],
capture_output=True,
text=True,
)
if proc.returncode != 0:
fail(f"worked example fails lint-skin: {proc.stdout.strip()}")
ok("worked example present, doc-inline sized, and skin-clean")
ok("references, SKILL.md wiring, and slash command are consistent")
def main() -> int:
with tempfile.TemporaryDirectory(prefix="diagram-design-drawio-") as tmp_dir:
tmp = Path(tmp_dir)
check_files()
check_parse_raw()
check_containers(tmp)
check_security_and_limits(tmp)
check_docs()
print("\nAll draw.io import gates passed.")
return 0
if __name__ == "__main__":
raise SystemExit(main())