Coverage for diagrams / generate.py: 100.00%
201 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1#!/usr/bin/env python3
2"""Regenerate or structurally verify every committed diagram catalogue."""
4from __future__ import annotations
6import argparse
7import os
8import re
9import subprocess
10import sys
11from pathlib import Path
13ROOT = Path(__file__).resolve().parent.parent
14sys.path.insert(0, str(ROOT))
16from diagrams.code_diagrams._renderer import _output_stem_for # noqa: E402
17from diagrams.code_diagrams._source_marker import SENTINEL # noqa: E402
18from diagrams.code_diagrams._targets import TARGETS # noqa: E402
19from diagrams.code_diagrams.generate import ( # noqa: E402
20 REGENERATION_HINT,
21 marker_allowed_sources,
22 newest_provenance_stamp,
23 verify_targets_match_provenance_manifest,
24)
25from diagrams.infra_diagrams._catalog import INFRA_DIAGRAM_NAMES # noqa: E402
26from gco.lambda_shared_sources import LAMBDA_SHARED_SOURCE_TARGETS # noqa: E402
28_TIMESTAMP_RE = re.compile(r"Generated at \(UTC\):[^\n]*?(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z)")
29_SOURCE_COMMIT_RE = re.compile(r"Generated from Git commit:[^\n]*?([0-9a-f]{40})")
30_HTML_SOURCE_COMMIT_RE = re.compile(r'<meta name="gco-source-commit" content="([0-9a-f]{40})">')
31_FLOW_DIGEST_RE = re.compile(r'<meta name="gco-flow-digest" content="([0-9a-f]{16})">')
32_MARKER_BLOCK_RE = re.compile(
33 rf"(?s)# <{re.escape(SENTINEL)}> BEGIN[^\n]*\n(.*?)# <{re.escape(SENTINEL)}> END"
34)
35_MARKER_ENTRY_RE = re.compile(
36 r"# \* ``([^`]+)`` -> ``([^`]+\.html)``\n"
37 r"(?:# \(PNG: ``([^`]+\.png)``\)\n)?"
38)
39_MARKER_BODY_RE = re.compile(
40 r"# Generated at \(UTC\): \d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z\n"
41 r"# Generated from Git commit: [0-9a-f]{40}\n"
42 r"# Flowchart\(s\) generated from this file:\n"
43 r"(?:# \* ``[^`]+`` -> ``[^`]+\.html``\n"
44 r"# \(PNG: ``[^`]+\.png``\)\n)+"
45 r"# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> "
46 r"GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> "
47 r"python diagrams/generate\.py --code-only``\.\n?\Z"
48)
49MarkerPointer = tuple[str, str, str | None]
52def _shared_source_copy_issues(project_root: Path, target_sources: set[str]) -> list[str]:
53 """Require every checked-in shared Lambda copy to equal its target source."""
54 issues: list[str] = []
55 for source, copies in LAMBDA_SHARED_SOURCE_TARGETS.items():
56 if source not in target_sources:
57 continue
58 canonical_path = project_root / source
59 if not canonical_path.is_file():
60 issues.append(f"missing canonical shared source: {source}")
61 continue
62 canonical = canonical_path.read_bytes()
63 for copy in copies:
64 copy_path = project_root / copy
65 if not copy_path.is_file():
66 issues.append(f"missing shared source copy: {copy}")
67 elif copy_path.read_bytes() != canonical:
68 issues.append(f"shared source copy drifted: {copy} != {source}")
69 return issues
72def _marker_pointer_issues(
73 source: str,
74 expected: set[MarkerPointer],
75 source_name: str,
76) -> list[str]:
77 """Compare a source marker's exact generated grammar and pointers."""
78 begin = f"# <{SENTINEL}> BEGIN"
79 end = f"# <{SENTINEL}> END"
80 if source.count(SENTINEL) != 2 or source.count(begin) != 1 or source.count(end) != 1:
81 return [f"source marker delimiter count invalid: {source_name}"]
82 blocks = _MARKER_BLOCK_RE.findall(source)
83 if len(blocks) != 1:
84 return [f"source marker block count invalid: {source_name}"]
86 issues: list[str] = []
87 if _MARKER_BODY_RE.fullmatch(blocks[0]) is None:
88 issues.append(f"source marker grammar invalid: {source_name}")
89 matches = _MARKER_ENTRY_RE.findall(blocks[0])
90 actual: set[MarkerPointer] = {
91 (function, html_path, png_path or None) for function, html_path, png_path in matches
92 }
93 if len(matches) != len(actual):
94 issues.append(f"source marker contains duplicate pointers: {source_name}")
95 if actual != expected:
96 issues.append(
97 f"source marker pointers drifted: {source_name}: "
98 f"missing={sorted(expected - actual)!r}, stale={sorted(actual - expected)!r}"
99 )
100 return issues
103def _code_artifact_contract(project_root: Path) -> list[str]:
104 output_dir = project_root / "diagrams" / "code_diagrams"
105 expected_html: set[Path] = set()
106 expected_png: set[Path] = set()
107 for target in TARGETS:
108 stem = _output_stem_for(target, output_dir=output_dir)
109 expected_html.add(stem.parent / f"{stem.name}.html")
110 expected_png.add(stem.parent / f"{stem.name}.png")
112 actual_html = set(output_dir.rglob("*.html"))
113 actual_png = set(output_dir.rglob("*.png"))
114 issues = [
115 *(
116 f"missing code artifact: {path.relative_to(project_root)}"
117 for path in sorted(expected_html - actual_html)
118 ),
119 *(
120 f"missing code artifact: {path.relative_to(project_root)}"
121 for path in sorted(expected_png - actual_png)
122 ),
123 *(
124 f"orphan code artifact: {path.relative_to(project_root)}"
125 for path in sorted(actual_html - expected_html)
126 ),
127 *(
128 f"orphan code artifact: {path.relative_to(project_root)}"
129 for path in sorted(actual_png - expected_png)
130 ),
131 ]
133 # Freshness is verified against the committed digest manifest, not Git
134 # history: squash merges delete branch commits, so a recorded SHA is
135 # provenance metadata rather than a resolvable object. The manifest also
136 # carries each source's own stamp, which is what lets one PR restamp only
137 # the sources it changed.
138 manifest: dict[str, dict[str, str]] = {}
139 try:
140 manifest = verify_targets_match_provenance_manifest(
141 project_root=project_root,
142 targets=TARGETS,
143 )
144 except RuntimeError as exc:
145 issues.append(str(exc))
147 readme = (output_dir / "README.md").read_text(encoding="utf-8")
148 readme_timestamps = set(_TIMESTAMP_RE.findall(readme))
149 readme_source_commits = set(_SOURCE_COMMIT_RE.findall(readme))
150 if len(readme_timestamps) != 1:
151 issues.append(f"code index timestamp invalid: {sorted(readme_timestamps)}")
152 if len(readme_source_commits) != 1:
153 issues.append(f"code index source commit invalid: {sorted(readme_source_commits)}")
154 if manifest and len(readme_timestamps) == 1 and len(readme_source_commits) == 1:
155 newest_at, newest_commit = newest_provenance_stamp(manifest)
156 if (next(iter(readme_timestamps)), next(iter(readme_source_commits))) != (
157 newest_at,
158 newest_commit,
159 ):
160 issues.append(
161 "diagrams/code_diagrams/README.md: its header stamp must match the "
162 f"newest provenance entry ({newest_at} / {newest_commit}) — rerun the "
163 f"generator to rewrite the index ({REGENERATION_HINT})"
164 )
165 expected_index_links: set[str] = set()
166 checked_sources: set[str] = set()
167 html_source: dict[Path, str] = {}
168 for target in TARGETS:
169 stem = _output_stem_for(target, output_dir=output_dir)
170 for suffix in ("html", "png"):
171 path = stem.parent / f"{stem.name}.{suffix}"
172 if suffix == "html":
173 html_source[path] = target.source
174 relative = path.relative_to(output_dir).as_posix()
175 expected_index_links.add(relative)
176 if f"./{relative}" not in readme:
177 issues.append(f"code index omitted: {relative}")
178 source = (project_root / target.source).read_text(encoding="utf-8")
179 if target.source not in checked_sources:
180 checked_sources.add(target.source)
181 expected_pointers: set[MarkerPointer] = set()
182 for source_target in TARGETS:
183 if source_target.source != target.source:
184 continue
185 source_stem = _output_stem_for(source_target, output_dir=output_dir)
186 html_path = source_stem.parent / f"{source_stem.name}.html"
187 png_path = source_stem.parent / f"{source_stem.name}.png"
188 expected_pointers.add(
189 (
190 source_target.function,
191 html_path.relative_to(project_root).as_posix(),
192 png_path.relative_to(project_root).as_posix(),
193 )
194 )
195 issues.extend(_marker_pointer_issues(source, expected_pointers, target.source))
196 source_timestamps = set(_TIMESTAMP_RE.findall(source))
197 marker_source_commits = set(_SOURCE_COMMIT_RE.findall(source))
198 if len(source_timestamps) != 1:
199 issues.append(
200 f"source marker timestamp invalid: {target.source}: {sorted(source_timestamps)}"
201 )
202 if len(marker_source_commits) != 1:
203 issues.append(
204 f"source marker commit invalid: {target.source}: "
205 f"{sorted(marker_source_commits)}"
206 )
207 # Each source's marker must agree with that source's own recorded
208 # provenance — not with the rest of the catalogue.
209 if (
210 manifest
211 and len(source_timestamps) == 1
212 and len(marker_source_commits) == 1
213 and target.source in manifest
214 ):
215 entry = manifest[target.source]
216 if (next(iter(source_timestamps)), next(iter(marker_source_commits))) != (
217 entry["generated_at"],
218 entry["source_commit"],
219 ):
220 issues.append(
221 f"{target.source}: its marker block's stamp disagrees with the "
222 "provenance recorded for it — regenerate this source's diagrams "
223 f"({REGENERATION_HINT})"
224 )
225 if f"``{target.function}``" not in source:
226 issues.append(f"source marker omitted: {target.source}:{target.function}")
228 indexed_artifacts = set(re.findall(r"\]\(\./([^)]+\.(?:html|png))\)", readme))
229 for relative in sorted(indexed_artifacts - expected_index_links):
230 issues.append(f"orphan code index entry: {relative}")
232 for html in expected_html & actual_html:
233 html_text = html.read_text(encoding="utf-8")
234 html_timestamps = set(_TIMESTAMP_RE.findall(html_text))
235 if len(html_timestamps) != 1:
236 issues.append(
237 f"code artifact timestamp invalid: {html.relative_to(project_root)}: "
238 f"{sorted(html_timestamps)}"
239 )
240 html_source_commits = set(_HTML_SOURCE_COMMIT_RE.findall(html_text))
241 if len(html_source_commits) != 1:
242 issues.append(
243 f"code artifact source commit invalid: {html.relative_to(project_root)}: "
244 f"{sorted(html_source_commits)}"
245 )
246 elif f"<code>{next(iter(html_source_commits))}</code>" not in html_text:
247 issues.append(
248 f"code artifact visible source commit omitted: {html.relative_to(project_root)}"
249 )
250 # An artifact must carry its own source's stamp; sibling artifacts
251 # derived from other sources are free to be a different vintage.
252 owner = html_source.get(html)
253 if (
254 manifest
255 and owner in manifest
256 and len(html_timestamps) == 1
257 and len(html_source_commits) == 1
258 ):
259 entry = manifest[owner]
260 if (next(iter(html_timestamps)), next(iter(html_source_commits))) != (
261 entry["generated_at"],
262 entry["source_commit"],
263 ):
264 issues.append(
265 f"{html.relative_to(project_root)}: this artifact's stamp disagrees "
266 f"with the provenance recorded for {owner} — regenerate that "
267 f"source's diagrams ({REGENERATION_HINT})"
268 )
269 flow_digests = set(_FLOW_DIGEST_RE.findall(html_text))
270 if len(flow_digests) != 1:
271 issues.append(
272 f"code artifact flow digest invalid: {html.relative_to(project_root)}: "
273 f"{sorted(flow_digests)}"
274 )
275 elif f"<code>{next(iter(flow_digests))}</code>" not in html_text:
276 issues.append(
277 f"code artifact visible flow digest omitted: {html.relative_to(project_root)}"
278 )
280 issues.extend(_shared_source_copy_issues(project_root, {target.source for target in TARGETS}))
281 # Shared with the generator's marker pruning so the two can never disagree
282 # about which files may legitimately carry a marker.
283 allowed_marker_sources = marker_allowed_sources(TARGETS)
284 marker_roots = [
285 project_root / "app.py",
286 project_root / "cli",
287 project_root / "gco",
288 project_root / "gco_mcp",
289 project_root / "lambda",
290 ]
291 for marker_root in marker_roots:
292 paths = [marker_root] if marker_root.is_file() else marker_root.rglob("*.py")
293 for path in paths:
294 if not path.is_file() or "-build" in path.as_posix():
295 continue
296 if SENTINEL not in path.read_text(encoding="utf-8"):
297 continue
298 relative = path.relative_to(project_root).as_posix()
299 if relative not in allowed_marker_sources:
300 issues.append(f"retired source marker: {relative}")
301 return issues
304def _infra_artifact_contract(project_root: Path) -> list[str]:
305 output_dir = project_root / "diagrams" / "infra_diagrams"
306 expected = {output_dir / f"{name}.png" for name in INFRA_DIAGRAM_NAMES}
307 actual = set(output_dir.glob("*.png"))
308 issues = [
309 *(f"missing infrastructure artifact: {path.name}" for path in sorted(expected - actual)),
310 *(f"orphan infrastructure artifact: {path.name}" for path in sorted(actual - expected)),
311 ]
312 issues.extend(
313 f"transient Graphviz sidecar: {path.name}" for path in sorted(output_dir.glob("*.dot"))
314 )
315 return issues
318def check_diagram_contract(
319 project_root: Path = ROOT,
320 *,
321 code: bool = True,
322 infra: bool = True,
323) -> list[str]:
324 """Return structural catalogue violations without modifying the checkout."""
325 issues: list[str] = []
326 if code:
327 issues.extend(_code_artifact_contract(project_root))
328 if infra:
329 issues.extend(_infra_artifact_contract(project_root))
330 return issues
333def main() -> None:
334 parser = argparse.ArgumentParser(description=__doc__)
335 selection = parser.add_mutually_exclusive_group()
336 selection.add_argument("--code-only", action="store_true")
337 selection.add_argument("--infra-only", action="store_true")
338 parser.add_argument(
339 "--check",
340 action="store_true",
341 help="Verify artifact/index/marker structure without rendering",
342 )
343 args = parser.parse_args()
344 code = not args.infra_only
345 infra = not args.code_only
347 if args.check:
348 issues = check_diagram_contract(code=code, infra=infra)
349 if issues:
350 for issue in issues:
351 print(f"ERROR: {issue}", file=sys.stderr)
352 raise SystemExit(1)
353 print("Diagram artifact contract is current")
354 return
356 if code and "SOURCE_DATE_EPOCH" not in os.environ:
357 parser.error("canonical code generation requires integer SOURCE_DATE_EPOCH")
358 if code and "GCO_DIAGRAM_SOURCE_COMMIT" not in os.environ:
359 parser.error("canonical code generation requires 40-character GCO_DIAGRAM_SOURCE_COMMIT")
360 if code:
361 subprocess.run(
362 [sys.executable, "diagrams/code_diagrams/generate.py", "--require-png"],
363 cwd=ROOT,
364 check=True,
365 )
366 if infra:
367 subprocess.run(
368 [sys.executable, "diagrams/infra_diagrams/generate.py", "--stack", "all"],
369 cwd=ROOT,
370 check=True,
371 )
373 issues = check_diagram_contract(code=code, infra=infra)
374 if issues:
375 raise RuntimeError(
376 "diagram generation completed with structural drift: " + "; ".join(issues)
377 )
380if __name__ == "__main__":
381 main()