-
■ added
-
■ modified
-
■ deleted
+
+ ■ added
+ ■ modified
+ ■ deleted
-
-{graph}
-
-""")
print(f" wrote {dst}")
PY
echo
-echo "diagram : $OUT/diagram.md"
+
+echo "diagram : $DIAGRAM_OUT"
echo "preview : $OUT/preview.html"
if [ "$OPEN" != "no" ]; then
- if command -v open >/dev/null 2>&1; then open "$OUT/preview.html";
- elif command -v xdg-open >/dev/null 2>&1; then xdg-open "$OUT/preview.html";
- else echo "(open $OUT/preview.html in your browser)"; fi
+ if command -v open >/dev/null 2>&1; then
+ open "$OUT/preview.html"
+ elif command -v xdg-open >/dev/null 2>&1; then
+ xdg-open "$OUT/preview.html"
+ else
+ echo "(open $OUT/preview.html in your browser)"
+ fi
fi
diff --git a/scripts/submit_feedback.py b/scripts/submit_feedback.py
deleted file mode 100644
index b0d9edc..0000000
--- a/scripts/submit_feedback.py
+++ /dev/null
@@ -1,176 +0,0 @@
-"""Submit explicit user feedback (/codeboarding-feedback) to PostHog.
-
-Standard-library only, on purpose: this runs in the action's guard phase, before
-the engine checkout and any dependency install, so it must not import third-party
-packages. Unlike Core's anonymous telemetry, this event intentionally carries the
-user-written feedback text and PR context — that difference is documented in the
-README. All sending failures are swallowed; feedback must never break a PR.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import sys
-import urllib.error
-import urllib.request
-
-# Public PostHog ingest key — the same write-only project key Core ships.
-DEFAULT_POSTHOG_KEY = "phc_BQWpoXuPYQhW7mPWQcRv4yzSfuoAmh48EmXuUpeXPUB2"
-DEFAULT_POSTHOG_HOST = "https://us.i.posthog.com"
-DEFAULT_COMMAND = "/codeboarding-feedback"
-DEFAULT_MAX_CHARS = 4000
-EVENT_NAME = "codeboarding_feedback_submitted"
-SOURCE = "github_action_feedback"
-
-
-def telemetry_disabled(env: dict) -> bool:
- """Mirror Core's opt-out: DO_NOT_TRACK or CODEBOARDING_TELEMETRY=false."""
- if env.get("DO_NOT_TRACK", "").strip().lower() in ("1", "true", "yes"):
- return True
- return env.get("CODEBOARDING_TELEMETRY", "true").strip().lower() == "false"
-
-
-def resolve_key(env: dict) -> str:
- return (env.get("CODEBOARDING_POSTHOG_KEY") or env.get("POSTHOG_KEY") or DEFAULT_POSTHOG_KEY).strip()
-
-
-def resolve_host(env: dict) -> str:
- host = (env.get("CODEBOARDING_POSTHOG_HOST") or env.get("POSTHOG_HOST") or DEFAULT_POSTHOG_HOST).strip()
- return host.rstrip("/") or DEFAULT_POSTHOG_HOST
-
-
-def resolve_command(env: dict) -> str:
- return (env.get("FEEDBACK_COMMAND") or "").strip() or DEFAULT_COMMAND
-
-
-def resolve_max_chars(env: dict) -> int:
- try:
- n = int((env.get("FEEDBACK_MAX_CHARS") or "").strip())
- except ValueError:
- return DEFAULT_MAX_CHARS
- return n if n > 0 else DEFAULT_MAX_CHARS
-
-
-def extract_feedback(comment_body: str, command: str) -> str:
- """Return everything after the leading command token, newlines preserved.
-
- The command is the first whitespace-delimited token of the comment. Only that
- one token is removed; the remainder (including any later lines) is kept
- verbatim, then outer whitespace is trimmed. Returns "" when the comment does
- not actually start with the command, or carries no text after it.
- """
- body = (comment_body or "").replace("\r\n", "\n").replace("\r", "\n").lstrip()
- if not body:
- return ""
- parts = body.split(None, 1) # split once on the first run of whitespace
- if parts[0] != command:
- return ""
- return parts[1].strip() if len(parts) > 1 else ""
-
-
-def cap_feedback(text: str, max_chars: int) -> tuple[str, int, bool]:
- """Return (capped_text, original_length, truncated)."""
- original_length = len(text)
- truncated = original_length > max_chars
- return (text[:max_chars] if truncated else text), original_length, truncated
-
-
-def _first(env: dict, *names: str) -> str:
- for name in names:
- value = (env.get(name) or "").strip()
- if value:
- return value
- return ""
-
-
-def distinct_id(env: dict) -> str:
- sender_id = _first(env, "SENDER_ID")
- if sender_id:
- return f"github-user:{sender_id}"
- return f"github-run:{_first(env, 'RUN_ID', 'GITHUB_RUN_ID')}"
-
-
-def build_properties(env: dict, command: str, feedback_text: str, feedback_length: int, truncated: bool) -> dict:
- props: dict = {
- "source": SOURCE,
- "command": command,
- "feedback_text": feedback_text,
- "feedback_length": feedback_length,
- "feedback_truncated": truncated,
- }
- optional = {
- "repository": _first(env, "REPOSITORY"),
- "repository_id": _first(env, "REPOSITORY_ID"),
- "pr_number": _first(env, "PR_NUMBER", "ISSUE_NUMBER"),
- "comment_id": _first(env, "COMMENT_ID"),
- "comment_url": _first(env, "COMMENT_URL"),
- "author_association": _first(env, "AUTHOR_ASSOC", "AUTHOR_ASSOCIATION"),
- "sender_login": _first(env, "SENDER_LOGIN"),
- "sender_id": _first(env, "SENDER_ID"),
- "run_id": _first(env, "RUN_ID", "GITHUB_RUN_ID"),
- "run_attempt": _first(env, "RUN_ATTEMPT", "GITHUB_RUN_ATTEMPT"),
- "action_ref": _first(env, "ACTION_REF", "GITHUB_ACTION_REF", "GITHUB_SHA"),
- }
- props.update({key: value for key, value in optional.items() if value})
- return props
-
-
-def build_payload(env: dict) -> dict | None:
- """Build the PostHog event payload, or None when there is nothing to send."""
- command = resolve_command(env)
- feedback_text, feedback_length, truncated = cap_feedback(
- extract_feedback(env.get("COMMENT_BODY", ""), command), resolve_max_chars(env)
- )
- if not feedback_text:
- return None
- return {
- "api_key": resolve_key(env),
- "event": EVENT_NAME,
- "distinct_id": distinct_id(env),
- "properties": build_properties(env, command, feedback_text, feedback_length, truncated),
- }
-
-
-def post(payload: dict, host: str, timeout: int = 10) -> int:
- """POST one event to PostHog's ingest endpoint; return the HTTP status."""
- request = urllib.request.Request(
- f"{host}/i/v0/e/",
- data=json.dumps(payload).encode("utf-8"),
- headers={"Content-Type": "application/json"},
- method="POST",
- )
- with urllib.request.urlopen(request, timeout=timeout) as response:
- return response.status
-
-
-def main(env: dict | None = None) -> int:
- env = os.environ if env is None else env
-
- if telemetry_disabled(env):
- print("Feedback disabled via DO_NOT_TRACK / CODEBOARDING_TELEMETRY; not sending.")
- return 0
-
- payload = build_payload(env)
- if payload is None:
- print("No feedback text after the command; nothing to send.")
- return 0
- if not payload["api_key"]:
- print("No PostHog key configured; skipping feedback send.")
- return 0
-
- truncated = payload["properties"].get("feedback_truncated")
- try:
- status = post(payload, resolve_host(env))
- print(f"Feedback submitted (HTTP {status}, truncated={truncated}).")
- except urllib.error.HTTPError as exc:
- print(f"Feedback endpoint returned HTTP {exc.code}; ignoring.")
- except urllib.error.URLError as exc:
- print(f"Feedback endpoint unreachable ({type(exc.reason).__name__}); ignoring.")
- except Exception as exc: # never let feedback break the action
- print(f"Feedback send failed ({type(exc).__name__}); ignoring.")
- return 0
-
-
-if __name__ == "__main__":
- sys.exit(main())
diff --git a/tests/test_analyze_repository.py b/tests/test_analyze_repository.py
new file mode 100644
index 0000000..e0783de
--- /dev/null
+++ b/tests/test_analyze_repository.py
@@ -0,0 +1,175 @@
+"""Smoke tests for scripts/analyze_repository.py — JSON contract parsing and mode dispatch."""
+
+import io
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
+
+import analyze_repository as ar
+
+
+class AnalyzeRepositoryTests(unittest.TestCase):
+ def _analysis_json(self, base: Path) -> Path:
+ path = base / "analysis.json"
+ path.write_text(
+ json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 2}}),
+ encoding="utf-8",
+ )
+ return path
+
+ def test_parse_cli_response_accepts_contract_json(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ out = root / "analysis.json"
+ out.write_text("ok", encoding="utf-8")
+ payload = json.dumps({"analysis_path": "analysis.json", "requiresFullAnalysis": True})
+ requires_full, path, _ = ar._parse_cli_response(payload, str(root))
+ self.assertTrue(requires_full)
+ self.assertEqual(path, out)
+
+ def test_parse_cli_response_rejects_invalid_bool(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ payload = json.dumps({"analysis_path": "analysis.json", "requiresFullAnalysis": "maybe"})
+ root = Path(tmp)
+ (root / "analysis.json").write_text("x", encoding="utf-8")
+ with self.assertRaises(ar.AnalysisError):
+ ar._parse_cli_response(payload, str(root))
+
+ def test_parse_cli_response_accepts_full_fallback_without_analysis_path(self) -> None:
+ payload = json.dumps({"error": "baseline unavailable", "requiresFullAnalysis": True})
+ requires_full, path, _ = ar._parse_cli_response(payload, "/tmp/output")
+ self.assertTrue(requires_full)
+ self.assertIsNone(path)
+
+ def test_parse_cli_response_accepts_logs_before_json(self) -> None:
+ raw = "Analyzing repository...\n" + json.dumps(
+ {"error": "baseline unavailable", "requiresFullAnalysis": True}, indent=2
+ )
+ requires_full, path, _ = ar._parse_cli_response(raw, "/tmp/output")
+ self.assertTrue(requires_full)
+ self.assertIsNone(path)
+
+ def test_run_command_streams_stdout_to_action_logs(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ stderr = io.StringIO()
+ command = [
+ sys.executable,
+ "-c",
+ "print('Analyzing repository...'); print('{\"requiresFullAnalysis\": true}')",
+ ]
+
+ with patch("sys.stderr", stderr):
+ stdout = ar._run_command(command, Path(tmp) / "out")
+
+ self.assertIn("Analyzing repository...", stderr.getvalue())
+ self.assertIn('{"requiresFullAnalysis": true}', stderr.getvalue())
+ self.assertEqual(stdout, 'Analyzing repository...\n{"requiresFullAnalysis": true}\n')
+
+ def test_parse_main_incremental_success(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ checkout = root / "repo"
+ out_dir = root / "out"
+ checkout.mkdir()
+ out_dir.mkdir()
+ analysis_path = out_dir / "analysis.json"
+ analysis_path.write_text("ok", encoding="utf-8")
+
+ stdout = io.StringIO()
+ with unittest.mock.patch("sys.stdout", stdout):
+ with patch.object(
+ ar,
+ "_run_command",
+ return_value=json.dumps(
+ {
+ "analysis_path": str(analysis_path.relative_to(out_dir)),
+ "requiresFullAnalysis": False,
+ }
+ ),
+ ) as _mock:
+ ar.main(
+ [
+ "incremental",
+ "--checkout",
+ str(checkout),
+ "--output-dir",
+ str(out_dir),
+ ]
+ )
+ lines = dict(line.split("=", 1) for line in stdout.getvalue().splitlines() if "=" in line)
+ self.assertEqual(lines.get("analysis_mode"), "incremental")
+ self.assertEqual(lines.get("requires_full_analysis"), "false")
+ self.assertEqual(lines.get("analysis_path"), str(analysis_path))
+
+ def test_main_full_fails_without_depth(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ checkout = root / "repo"
+ out_dir = root / "out"
+ checkout.mkdir()
+ out_dir.mkdir()
+ with self.assertRaises(SystemExit):
+ ar.main(
+ [
+ "full",
+ "--checkout",
+ str(checkout),
+ "--output-dir",
+ str(out_dir),
+ ]
+ )
+
+ def test_main_full_uses_generated_analysis_file(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ checkout = root / "repo"
+ out_dir = root / "out"
+ checkout.mkdir()
+
+ def fake_run(_args, output_dir):
+ (output_dir / "analysis.json").write_text("ok", encoding="utf-8")
+ return "human-readable CLI output"
+
+ stdout = io.StringIO()
+ with patch("sys.stdout", stdout), patch.object(ar, "_run_command", side_effect=fake_run):
+ ar.main(
+ [
+ "full",
+ "--checkout",
+ str(checkout),
+ "--output-dir",
+ str(out_dir),
+ "--depth-level",
+ "1",
+ ]
+ )
+
+ self.assertIn(f"analysis_path={out_dir / 'analysis.json'}", stdout.getvalue())
+
+ def test_main_rejects_bad_cli_output(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ checkout = root / "repo"
+ out_dir = root / "out"
+ checkout.mkdir()
+ out_dir.mkdir()
+ with self.assertRaises(ar.AnalysisError):
+ with patch.object(ar, "_run_command", return_value="not-json"):
+ ar.main(
+ [
+ "incremental",
+ "--checkout",
+ str(checkout),
+ "--output-dir",
+ str(out_dir),
+ ]
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_build_component_files.py b/tests/test_build_component_files.py
deleted file mode 100644
index d4f36fc..0000000
--- a/tests/test_build_component_files.py
+++ /dev/null
@@ -1,321 +0,0 @@
-"""Unit tests for scripts/build_component_files.py — per-component changed-file dropdowns."""
-
-import json
-import os
-import re
-import subprocess
-import sys
-import tempfile
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
-import build_component_files as bcf # noqa: E402
-import diff_to_mermaid as dm # noqa: E402
-
-SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "build_component_files.py"
-
-
-def comp(name, files=None, subs=None, key_files=None):
- c = {
- "name": name,
- "component_id": name,
- "file_methods": [{"file_path": f, "methods": m} for f, m in (files or {}).items()],
- }
- if key_files is not None:
- c["key_entities"] = [{"reference_file": f} for f in key_files]
- if subs is not None:
- c["components"] = subs
- return c
-
-
-def render(base, head, changed_files=None, max_chars=bcf.MAX_TEXT):
- diff = dm.build_diff(base, head)
- return bcf.render_component_files(diff, base, changed_files, max_chars)
-
-
-class TestGitIntersection(unittest.TestCase):
- def test_modified_component_lists_only_touched_files(self):
- base = {"components": [comp("Auth", {"a.py": ["f"], "b.py": ["g"], "c.py": ["h"]})]}
- head = {"components": [comp("Auth", {"a.py": ["f", "f2"], "b.py": ["g"], "c.py": ["h"]})]}
- text, meta = render(base, head, changed_files={"a.py", "unrelated.py"})
- self.assertIn("
a.py", text)
- self.assertNotIn("b.py", text) # owned but untouched
- self.assertNotIn("unrelated.py", text) # touched but not owned
- self.assertIn("
Auth : 1 file changed", text)
- self.assertEqual(meta["n_components"], 1)
- self.assertEqual(meta["n_files"], 1)
-
- def test_added_component_wording(self):
- base = {"components": []}
- head = {"components": [comp("RateLimiter", {"rl/bucket.py": ["acquire"], "rl/config.py": ["load"]})]}
- text, _ = render(base, head, changed_files={"rl/bucket.py", "rl/config.py"})
- self.assertIn("
RateLimiter : 2 files added", text)
-
- def test_deleted_component_lists_base_files(self):
- base = {"components": [comp("Legacy", {"legacy/store.py": ["get"], "legacy/migrations.py": ["mig"]})]}
- head = {"components": []}
- text, _ = render(base, head, changed_files={"legacy/store.py", "legacy/migrations.py"})
- self.assertIn("
Legacy : 2 files removed", text)
- self.assertIn("
legacy/migrations.py", text)
-
- def test_unchanged_component_emits_nothing(self):
- base = {"components": [comp("A", {"a.py": ["f"]})]}
- head = {"components": [comp("A", {"a.py": ["f"]})]}
- text, meta = render(base, head, changed_files={"a.py"})
- self.assertEqual(text, "")
- self.assertFalse(meta["rendered"])
-
- def test_changed_component_with_no_touched_files_is_skipped(self):
- # Model reorg: file moved between components, but the PR's git diff is elsewhere.
- base = {"components": [comp("A", {"a.py": ["f"], "b.py": ["g"]})]}
- head = {"components": [comp("A", {"a.py": ["f"]})]}
- text, _ = render(base, head, changed_files={"elsewhere.py"})
- self.assertEqual(text, "")
-
- def test_empty_changed_files_set_means_no_dropdowns_not_fallback(self):
- # Empty git diff (net-zero PR / re-run): an empty set must NOT fall
- # back to analysis-derived changes — only None (flag omitted) does.
- base = {"components": [comp("A", {"a.py": ["f"]})]}
- head = {"components": [comp("A", {"a.py": ["f", "g"]})]}
- text, meta = render(base, head, changed_files=set())
- self.assertEqual(text, "")
- self.assertFalse(meta["rendered"])
-
- def test_nested_subtree_files_aggregate_to_top_level(self):
- base = {"components": [comp("Parent", {}, subs=[comp("Child", {"deep/x.py": ["f"]})])]}
- head = {"components": [comp("Parent", {}, subs=[comp("Child", {"deep/x.py": ["f", "g"]})])]}
- text, _ = render(base, head, changed_files={"deep/x.py"})
- self.assertIn("
Parent", text)
- self.assertIn("
deep/x.py", text)
- self.assertNotIn("
Child", text) # one dropdown per top-level component
-
- def test_rollup_parent_labels_changed_subcomponents(self):
- # Parent unchanged itself (display_status rollup): the summary carries the
- # recursive count the headline/diagram use, so counts don't contradict.
- base = {"components": [comp("Parent", {"p.py": ["f"]}, subs=[comp("Child", {"c.py": ["g"]})])]}
- head = {"components": [comp("Parent", {"p.py": ["f"]}, subs=[comp("Child", {"c.py": ["g", "g2"]})])]}
- text, _ = render(base, head, changed_files={"c.py"})
- self.assertIn("
Parent : 1 changed sub-component, 1 file changed", text)
-
- def test_deleted_nested_child_files_list_under_modified_parent(self):
- base = {"components": [comp("Parent", {"p.py": ["f"]}, subs=[comp("Child", {"child/x.py": ["g"]})])]}
- head = {"components": [comp("Parent", {"p.py": ["f"]}, subs=[])]}
- text, _ = render(base, head, changed_files={"child/x.py"})
- self.assertIn("
Parent", text)
- self.assertIn("
child/x.py", text)
-
- def test_key_entities_only_shape(self):
- # Some engine outputs have file_methods: [] everywhere and carry file
- # linkage only in key_entities[].reference_file (observed on TS repos).
- base = {"components": []}
- head = {"components": [comp("Webview", key_files=["src/panel.ts", "src/render.ts"])]}
- text, _ = render(base, head, changed_files={"src/panel.ts", "src/render.ts"})
- self.assertIn("
Webview : 2 files added", text)
- self.assertIn("
src/panel.ts", text)
-
- def test_duplicate_deleted_names_attribute_files_to_own_block(self):
- base = {"components": [comp("Dup", {"one.py": ["f"]}), comp("Dup", {"two.py": ["g"]})]}
- head = {"components": []}
- for changed in (None, {"one.py", "two.py"}):
- text, _ = render(base, head, changed_files=changed)
- self.assertEqual(text.count("one.py"), 1, text)
- self.assertEqual(text.count("two.py"), 1, text)
-
-
-class TestAnalysisFallback(unittest.TestCase):
- def test_fallback_lists_structural_and_method_changes(self):
- base = {"components": [comp("A", {"kept.py": ["f"], "gone.py": ["g"], "same.py": ["h"]})]}
- head = {"components": [comp("A", {"kept.py": ["f", "f2"], "new.py": ["n"], "same.py": ["h"]})]}
- text, _ = render(base, head, changed_files=None)
- self.assertIn("
kept.py", text) # method set changed
- self.assertIn("
gone.py", text) # removed from component
- self.assertIn("
new.py", text) # added to component
- self.assertNotIn("same.py", text)
-
- def test_fallback_deleted_component_lists_all_base_files(self):
- base = {"components": [comp("Legacy", {"l/a.py": ["f"], "l/b.py": ["g"]})]}
- head = {"components": []}
- text, _ = render(base, head, changed_files=None)
- self.assertIn("
Legacy : 2 files removed", text)
-
- def test_missing_file_path_entry_emits_no_phantom(self):
- base = {"components": [comp("A", {"a.py": ["f"]})]}
- head = {"components": [comp("A", {"a.py": ["f"]})]}
- head["components"][0]["file_methods"].append({"methods": ["orphan"]}) # no file_path
- text, _ = render(base, head, changed_files=None)
- self.assertNotIn("
", text)
-
-
-class TestOrdering(unittest.TestCase):
- def test_file_lists_are_sorted(self):
- files = {f: ["m"] for f in ["e.py", "b.py", "f.py", "a.py", "d.py", "c.py"]}
- base = {"components": []}
- head = {"components": [comp("A", files)]}
- text, _ = render(base, head, changed_files=set(files))
- paths = re.findall(r"
([^<]+)", text)
- self.assertEqual(paths, ["a.py", "b.py", "c.py", "d.py", "e.py", "f.py"])
-
- def test_blocks_follow_diagram_order_deleted_ghosts_last(self):
- # Head order first (matches Mermaid node emission), deleted ghosts appended
- # last — NOT alphabetical: Alpha is deleted and must render after Zeta.
- base = {"components": [comp("Zeta", {"z.py": ["f"]}), comp("Alpha", {"a.py": ["g"]})]}
- head = {"components": [comp("Zeta", {"z.py": ["f", "f2"]})]}
- text, _ = render(base, head, changed_files={"z.py", "a.py"})
- self.assertEqual(re.findall(r"
(\w+)", text), ["Zeta", "Alpha"])
-
-
-class TestCapsAndEscaping(unittest.TestCase):
- def test_per_component_file_cap(self):
- files = {f"src/f{i:02}.py": ["m"] for i in range(20)}
- base = {"components": []}
- head = {"components": [comp("Big", files)]}
- text, meta = render(base, head, changed_files=set(files))
- self.assertEqual(text.count("
"), bcf.MAX_FILES_PER_COMPONENT)
- self.assertIn("…and 5 more", text)
- self.assertIn(": 20 files added", text) # count reflects reality, list is capped
- self.assertTrue(meta["truncated"])
-
- def test_total_char_budget_drops_whole_components(self):
- base = {"components": []}
- head = {"components": [comp(f"C{i}", {f"c{i}/f.py": ["m"]}) for i in range(10)]}
- text, meta = render(base, head, changed_files={f"c{i}/f.py" for i in range(10)}, max_chars=300)
- self.assertIn("more changed components", text)
- self.assertTrue(meta["truncated"])
- # meta counts what actually rendered, not what the budget dropped
- self.assertEqual(meta["n_files"], text.count(""))
- self.assertEqual(meta["n_components"], text.count(""))
-
- def test_first_block_exceeding_budget_renders_nothing(self):
- # Never a dangling "…and N more" with no blocks above it.
- base = {"components": []}
- head = {"components": [comp("Big", {f"very/long/path/file{i}.py": ["m"] for i in range(15)})]}
- text, meta = render(base, head, changed_files={f"very/long/path/file{i}.py" for i in range(15)}, max_chars=100)
- self.assertEqual(text, "")
- self.assertFalse(meta["rendered"])
- self.assertEqual(meta["n_files"], 0)
- self.assertTrue(meta["truncated"])
-
- def test_html_escaping_of_names_and_paths(self):
- base = {"components": []}
- head = {"components": [comp("A <& B", {"weird/&.py": ["m"]})]}
- text, _ = render(base, head, changed_files={"weird/&.py"})
- self.assertIn("A <& B", text)
- self.assertIn("weird/<path>&.py", text)
- self.assertNotIn("", text)
-
- def test_blank_line_after_summary(self):
- # GitHub only renders markdown inside after a blank line.
- base = {"components": []}
- head = {"components": [comp("A", {"a.py": ["m"]})]}
- text, _ = render(base, head, changed_files={"a.py"})
- self.assertIn("\n\n-", text)
- self.assertIn("\n\n ", text)
-
-
-class TestCLI(unittest.TestCase):
- def _analyses(self, d):
- (d / "base.json").write_text(json.dumps({"components": [comp("Auth", {"a.py": ["f"], "b.py": ["g"]})]}))
- (d / "head.json").write_text(json.dumps({"components": [comp("Auth", {"a.py": ["f", "f2"], "b.py": ["g"]})]}))
-
- def _run(self, d, *extra):
- out = d / "out.md"
- core = d / "fake-core"
- (core / "codeboarding_workflows").mkdir(parents=True)
- (core / "diagram_analysis").mkdir()
- (core / "codeboarding_workflows" / "__init__.py").write_text("")
- (core / "diagram_analysis" / "__init__.py").write_text("")
- (core / "codeboarding_workflows" / "rendering.py").write_text(
- "def project_relations_to_level(relations, level_ids, id_to_name):\n"
- " return [r for r in relations if r.src_id in level_ids and r.dst_id in level_ids]\n"
- )
- (core / "diagram_analysis" / "analysis_json.py").write_text(
- "from types import SimpleNamespace\n"
- "def parse_unified_analysis(data):\n"
- " components = [SimpleNamespace(component_id=c['component_id']) for c in data.get('components', [])]\n"
- " relations = [SimpleNamespace(**r) for r in data.get('components_relations', [])]\n"
- " return SimpleNamespace(components=components, components_relations=relations), {}\n"
- "def build_id_to_name_map(root, subs):\n"
- " return {}\n"
- )
- args = [
- sys.executable,
- str(SCRIPT),
- "--base",
- str(d / "base.json"),
- "--head",
- str(d / "head.json"),
- "--out",
- str(out),
- ]
- env = {**os.environ, "PYTHONPATH": str(core)}
- return out, subprocess.run([*args, *extra], capture_output=True, text=True, env=env)
-
- def test_main_writes_out_file_and_prints_meta(self):
- with tempfile.TemporaryDirectory() as tmp:
- d = Path(tmp)
- self._analyses(d)
- (d / "changed.txt").write_text("a.py\nunrelated.py\n")
- out, r = self._run(d, "--changed-files", str(d / "changed.txt"))
- self.assertEqual(r.returncode, 0, r.stderr)
- content = out.read_text(encoding="utf-8")
- self.assertIn("a.py", content)
- self.assertTrue(content.endswith(" \n")) # trailing newline: see main()
- meta = json.loads(r.stdout)
- self.assertEqual(set(meta), {"rendered", "n_components", "n_files", "truncated"})
- self.assertTrue(meta["rendered"])
-
- def test_non_utf8_changed_files_does_not_crash(self):
- # core.quotepath=off emits raw filename bytes; a non-UTF-8 path must not
- # kill the section — it just can't intersect with the analysis's paths.
- with tempfile.TemporaryDirectory() as tmp:
- d = Path(tmp)
- self._analyses(d)
- (d / "changed.txt").write_bytes(b"r\xe9sum\xe9.py\na.py\n")
- out, r = self._run(d, "--changed-files", str(d / "changed.txt"))
- self.assertEqual(r.returncode, 0, r.stderr)
- self.assertIn("a.py", out.read_text(encoding="utf-8"))
-
- def test_empty_result_writes_zero_bytes(self):
- # The action gates the section on [ -s "$FILES_MD" ].
- with tempfile.TemporaryDirectory() as tmp:
- d = Path(tmp)
- self._analyses(d)
- (d / "changed.txt").write_text("elsewhere.py\n")
- out, r = self._run(d, "--changed-files", str(d / "changed.txt"))
- self.assertEqual(r.returncode, 0, r.stderr)
- self.assertEqual(out.read_bytes(), b"")
-
-
-class TestEngineGitPathContract(unittest.TestCase):
- """file_methods[].file_path must be repo-relative forward-slash paths identical
- to git --name-only output; pinned against the committed engine artifact (the
- dogfood workflows regenerate it on engine bumps, so format drift fails here)."""
-
- def test_committed_analysis_paths_are_git_name_only_format(self):
- root = Path(__file__).resolve().parent.parent
- analysis = json.loads((root / ".codeboarding" / "analysis.json").read_text())
- paths = set()
-
- def collect(c):
- for fm in c.get("file_methods") or []:
- paths.add(fm["file_path"])
- for ke in c.get("key_entities") or []:
- if ke.get("reference_file"):
- paths.add(ke["reference_file"])
- for s in c.get("components") or []:
- collect(s)
-
- for c in analysis.get("components") or []:
- collect(c)
- self.assertTrue(paths, "committed analysis.json has no file paths")
- tracked = set(
- subprocess.run(
- ["git", "-C", str(root), "ls-files"], capture_output=True, text=True, check=True
- ).stdout.splitlines()
- )
- self.assertLessEqual(paths, tracked, f"paths not in git --name-only format: {sorted(paths - tracked)[:5]}")
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/test_build_cta.py b/tests/test_build_cta.py
deleted file mode 100644
index e439d51..0000000
--- a/tests/test_build_cta.py
+++ /dev/null
@@ -1,195 +0,0 @@
-"""Unit tests for scripts/build_cta.py — editor detection + CTA footer."""
-
-import sys
-import tempfile
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
-import build_cta as bc # noqa: E402
-
-
-def repo_with(*dirs):
- d = Path(tempfile.mkdtemp())
- for x in dirs:
- (d / x).mkdir()
- return d
-
-
-class TestDetectEditors(unittest.TestCase):
- def test_neither_defaults_to_vscode(self):
- self.assertEqual(bc.detect_editors(repo_with()), ["vscode"])
-
- def test_vscode_only(self):
- self.assertEqual(bc.detect_editors(repo_with(".vscode")), ["vscode"])
-
- def test_cursor_only(self):
- self.assertEqual(bc.detect_editors(repo_with(".cursor")), ["cursor"])
-
- def test_both_vscode_first(self):
- self.assertEqual(bc.detect_editors(repo_with(".vscode", ".cursor")), ["vscode", "cursor"])
-
-
-class TestBuildCta(unittest.TestCase):
- def test_no_proxy_links_editor_to_https_listing_no_get_extension(self):
- out = bc.build_cta("", "o", "r", "1", repo_with(".cursor"), issues=3)
- self.assertIn("3 architecture issues found", out)
- # Cursor -> Open VSX https listing. A cursor: scheme would be stripped by GitHub.
- self.assertIn("[**Cursor**](https://open-vsx.org/extension/CodeBoarding/codeboarding)", out)
- self.assertNotIn("cursor:extension", out)
- self.assertNotIn("Get the extension", out) # dropped without a proxy
- self.assertNotIn("VS Code", out) # cursor-only repo
-
- def test_no_proxy_vscode_marketplace_https_no_banner_at_zero(self):
- out = bc.build_cta("", "o", "r", "1", repo_with()) # neither dir, no issues
- self.assertIn(
- "[**VS Code**](https://marketplace.visualstudio.com/items?itemName=Codeboarding.codeboarding)",
- out,
- )
- self.assertNotIn("vscode:extension", out) # custom scheme stripped by GitHub
- self.assertNotIn("Get the extension", out)
- self.assertNotIn("architecture issue", out) # banner suppressed at 0 issues
-
- def test_links_banner_and_cursor_only(self):
- out = bc.build_cta("https://x.dev/", "Org", "Repo", "9", repo_with(".cursor"), issues=2)
- self.assertIn("2 architecture issues found", out)
- self.assertIn("open-in-editor?owner=Org&repo=Repo&pr=9&editor=cursor", out)
- self.assertIn("use-marketplace?owner=Org&repo=Repo&pr=9", out) # proxy "Get the extension"
- self.assertNotIn("VS Code", out) # cursor-only repo
-
- def test_no_banner_when_zero_issues_and_default_vscode(self):
- out = bc.build_cta("https://x.dev", "o", "r", "1", repo_with(), issues=0)
- self.assertNotIn("architecture issue", out)
- self.assertIn("VS Code", out)
- self.assertNotIn("Cursor", out)
-
- def test_both_editors_singular_issue(self):
- out = bc.build_cta("https://x.dev", "o", "r", "1", repo_with(".vscode", ".cursor"), issues=1)
- self.assertIn("1 architecture issue found", out) # singular
- self.assertIn("VS Code", out)
- self.assertIn("Cursor", out)
-
- def test_trailing_slash_in_base_is_normalized(self):
- a = bc.build_cta("https://x.dev/", "o", "r", "1", repo_with())
- b = bc.build_cta("https://x.dev", "o", "r", "1", repo_with())
- self.assertNotIn("x.dev//", a)
- self.assertEqual(a, b)
-
-
-class TestWebviewUrl(unittest.TestCase):
- WV = "https://app.codeboarding.org"
-
- def test_url_is_github_style_pr_path(self):
- url = bc.webview_url(self.WV, "Org", "Repo", pr="9", run_id="123")
- self.assertEqual(url, "https://app.codeboarding.org/Org/Repo/pull/9?run=123")
-
- def test_url_carries_only_pr_path_and_run(self):
- # Head/base SHAs and the artifact name/url are re-derived by the webview, so
- # none of them appear in the short link.
- url = bc.webview_url(self.WV, "o", "r", pr="9", run_id="123")
- self.assertIn("/o/r/pull/9", url)
- self.assertIn("run=123", url)
- self.assertNotIn("ref=", url)
- self.assertNotIn("compare=", url)
- self.assertNotIn("artifact", url)
- self.assertNotIn("repo=o%2Fr", url) # not the old query-style link
-
- def test_url_none_without_pr_or_run(self):
- self.assertIsNone(bc.webview_url(self.WV, "o", "r", pr="9")) # no run
- self.assertIsNone(bc.webview_url(self.WV, "o", "r", run_id="123")) # no pr
- self.assertIsNone(bc.webview_url("", "o", "r", pr="9", run_id="123")) # no base
-
- def test_trailing_slash_in_webview_base_is_normalized(self):
- a = bc.webview_url("https://app.codeboarding.org/", "o", "r", pr="9", run_id="1")
- b = bc.webview_url("https://app.codeboarding.org", "o", "r", pr="9", run_id="1")
- self.assertEqual(a, b)
- self.assertNotIn(".org//", a)
-
- def test_cta_includes_browser_link_when_ready(self):
- out = bc.build_cta(
- "",
- "Org",
- "Repo",
- "9",
- repo_with(),
- issues=0,
- webview_base=self.WV,
- webview_ready=True,
- run_id="123",
- )
- self.assertIn("Explore this PR", out)
- self.assertIn("your [**browser**](", out)
- self.assertIn("/Org/Repo/pull/9?run=123", out)
- self.assertIn("VS Code", out) # editor merged into the same line
-
- def test_cta_omits_browser_link_when_not_ready(self):
- # No uploaded analysis artifact -> webview can't fetch PR-specific data.
- out = bc.build_cta(
- "",
- "Org",
- "Repo",
- "9",
- repo_with(),
- issues=0,
- webview_base=self.WV,
- webview_ready=False,
- run_id="123",
- )
- self.assertNotIn("/pull/", out) # no browser link
- self.assertNotIn("[**browser**]", out)
- self.assertIn("Explore this PR", out) # the line is still there, editor-only
- self.assertIn("VS Code", out)
-
- def test_cta_omits_browser_link_when_ready_but_no_base_url(self):
- out = bc.build_cta(
- "",
- "Org",
- "Repo",
- "9",
- repo_with(),
- issues=0,
- webview_base="",
- webview_ready=True,
- run_id="123",
- )
- self.assertNotIn("[**browser**]", out)
- self.assertNotIn("/pull/", out)
-
-
-class TestJoinOr(unittest.TestCase):
- def test_join_shapes(self):
- self.assertEqual(bc._join_or(["a"]), "a")
- self.assertEqual(bc._join_or(["a", "b"]), "a or b")
- self.assertEqual(bc._join_or(["a", "b", "c"]), "a, b, or c")
-
-
-class TestMergedExploreLine(unittest.TestCase):
- WV = "https://app.codeboarding.org"
-
- def _ready(self, repo, cta=""):
- return bc.build_cta(cta, "o", "r", "1", repo, webview_base=self.WV, webview_ready=True, run_id="123")
-
- def test_browser_and_single_editor_joined_with_or(self):
- out = self._ready(repo_with()) # default VS Code
- self.assertIn("in your [**browser**](", out)
- self.assertIn(") or [**VS Code**](", out) # browser editor on one line
-
- def test_editor_only_has_no_your_and_no_browser(self):
- out = bc.build_cta("", "o", "r", "1", repo_with()) # no webview
- self.assertIn("architecture in [**VS Code**](", out) # "in " with no "your"
- self.assertNotIn("browser", out)
-
- def test_browser_and_two_editors_use_oxford_or(self):
- out = self._ready(repo_with(".vscode", ".cursor"))
- self.assertIn("your [**browser**](", out)
- self.assertIn(", or [**Cursor**](", out) # 3 targets -> ", or" before the last
-
- def test_two_editors_no_browser_joined_with_or(self):
- out = bc.build_cta("", "o", "r", "1", repo_with(".vscode", ".cursor"))
- self.assertIn(" or [**Cursor**](", out)
- self.assertNotIn(", or [**Cursor**]", out) # 2 targets -> plain "or", no Oxford comma
- self.assertNotIn("browser", out)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/test_engine_adapter.py b/tests/test_engine_adapter.py
deleted file mode 100644
index 433e7f9..0000000
--- a/tests/test_engine_adapter.py
+++ /dev/null
@@ -1,998 +0,0 @@
-"""Smoke tests for scripts/engine_adapter.py — verify it calls the engine API correctly,
-using stub modules so no real engine venv is needed."""
-
-import json
-import os
-import subprocess
-import sys
-import tempfile
-import types
-import unittest
-from contextlib import redirect_stderr, redirect_stdout
-from io import StringIO
-from pathlib import Path
-from unittest.mock import patch
-
-
-def _preload(name, **attrs):
- m = types.ModuleType(name)
- for k, v in attrs.items():
- setattr(m, k, v)
- sys.modules[name] = m
- return m
-
-
-class _InitialBaselineUnavailableError(Exception):
- pass
-
-
-class _InitialIncrementalCacheMissingError(Exception):
- pass
-
-
-class _InitialSeverity:
- WARNING, CRITICAL = "warning", "critical"
-
-
-class _InitialStaticAnalysisCache:
- def __init__(self, *args, **kwargs):
- pass
-
- def get(self):
- return None
-
- def save(self, *args, **kwargs):
- pass
-
-
-class _RunPaths:
- def __init__(self, repo_path=None, output_dir=None, project_name=None):
- self.repo_path, self.output_dir, self.project_name = repo_path, output_dir, project_name
-
-
-class _RunContext:
- def __init__(self, run_id=None, log_path=None, repo_dir=None):
- self.run_id, self.log_path, self.repo_dir = run_id, log_path, repo_dir
-
-
-class _InitialUnifiedAnalysisJson:
- def __init__(self, data):
- self.data = data
-
- @classmethod
- def model_validate(cls, data):
- return cls(data)
-
- def model_dump(self, **kwargs):
- return self.data
-
-
-class _RejectingUnifiedAnalysisJson:
- @classmethod
- def model_validate(cls, data):
- raise ValueError("incompatible analysis schema")
-
-
-class _LossyUnifiedAnalysisJson(_InitialUnifiedAnalysisJson):
- def model_dump(self, **kwargs):
- return {"normalized": True}
-
-
-analysis = _preload(
- "codeboarding_workflows.analysis",
- run_full=lambda *a, **k: "OUT",
- run_incremental=lambda *a, **k: "OUT",
- BaselineUnavailableError=_InitialBaselineUnavailableError,
-)
-pkg = _preload("codeboarding_workflows")
-pkg.analysis = analysis
-exc = _preload("diagram_analysis.exceptions", IncrementalCacheMissingError=_InitialIncrementalCacheMissingError)
-da = _preload("diagram_analysis", RunPaths=_RunPaths, RunContext=_RunContext)
-da.exceptions = exc
-_preload("diagram_analysis.analysis_json", UnifiedAnalysisJson=_InitialUnifiedAnalysisJson)
-_preload("diagram_analysis.io_utils", write_fingerprint=lambda *a, **k: None)
-_preload("logging_config", setup_logging=lambda **kwargs: None)
-_preload("agents.content_hash", hash_repo_source_files=lambda *a, **k: {})
-_preload("agents")
-_preload("codeboarding_workflows.rendering", render_docs=lambda *args, **kwargs: None)
-_preload("health.models", Severity=_InitialSeverity)
-_preload("health.runner", run_health_checks=lambda *args, **kwargs: None)
-_preload("health")
-_preload("static_analyzer", get_static_analysis=lambda *args, **kwargs: {})
-_preload("static_analyzer.analysis_cache", StaticAnalysisCache=_InitialStaticAnalysisCache)
-_preload("static_analyzer.cluster_helpers", build_all_cluster_results=lambda *args, **kwargs: {})
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
-import engine_adapter # noqa: E402
-
-_STUBBED = [
- "agents",
- "agents.content_hash",
- "codeboarding_workflows",
- "codeboarding_workflows.analysis",
- "diagram_analysis",
- "diagram_analysis.analysis_json",
- "diagram_analysis.exceptions",
- "diagram_analysis.io_utils",
- "logging_config",
- "health",
- "health.models",
- "health.runner",
- "static_analyzer",
- "static_analyzer.analysis_cache",
- "static_analyzer.cluster_helpers",
-]
-
-
-class _Rec:
- def __init__(self, ret="OUT", raises=None):
- self.calls = [] # kwargs of each call
- self.args = [] # positional args of each call
- self._ret, self._raises = ret, raises
-
- def __call__(self, *a, **k):
- self.calls.append(k)
- self.args.append(a)
- if self._raises:
- raise self._raises("boom")
- return self._ret
-
-
-def _mod(name, **attrs):
- m = types.ModuleType(name)
- for k, v in attrs.items():
- setattr(m, k, v)
- sys.modules[name] = m
- return m
-
-
-def _write_model_valid_analysis(output_dir, **metadata):
- path = Path(output_dir)
- path.mkdir(parents=True, exist_ok=True)
- (path / "analysis.json").write_text(
- json.dumps({"metadata": metadata}),
- encoding="utf-8",
- )
-
-
-class _Base(unittest.TestCase):
- def tearDown(self):
- for n in _STUBBED:
- sys.modules.pop(n, None)
-
-
-class TestAnalysis(_Base):
- def _install(self, run_full=None, run_incremental=None):
- class BaselineUnavailableError(Exception):
- pass
-
- class IncrementalCacheMissingError(Exception):
- pass
-
- analysis = _mod(
- "codeboarding_workflows.analysis",
- run_full=run_full or _Rec(),
- run_incremental=run_incremental or _Rec(),
- BaselineUnavailableError=BaselineUnavailableError,
- )
- pkg = _mod("codeboarding_workflows")
- pkg.analysis = analysis
- exc = _mod("diagram_analysis.exceptions", IncrementalCacheMissingError=IncrementalCacheMissingError)
- da = _mod("diagram_analysis")
- da.exceptions = exc
- engine_adapter.run_full = analysis.run_full
- engine_adapter.run_incremental = analysis.run_incremental
- engine_adapter.BaselineUnavailableError = BaselineUnavailableError
- engine_adapter.IncrementalCacheMissingError = IncrementalCacheMissingError
- return analysis, IncrementalCacheMissingError, BaselineUnavailableError
-
- def test_base_calls_run_full(self):
- rf = _Rec()
- self._install(run_full=rf)
- engine_adapter.run_base("/repo", "/out", "myrepo", "rid-base", 2, "abc123")
- self.assertEqual(len(rf.calls), 1)
- run_paths, run_context = rf.args[0]
- self.assertEqual(run_paths.project_name, "myrepo")
- self.assertEqual(str(run_paths.repo_path), "/repo")
- self.assertEqual(str(run_paths.output_dir), "/out")
- self.assertEqual(run_context.run_id, "rid-base")
- self.assertEqual(rf.calls[0]["depth_level"], 2)
- self.assertEqual(rf.calls[0]["source_sha"], "abc123")
-
- def test_main_parses_depth_as_int(self):
- rf = _Rec()
- self._install(run_full=rf)
- engine_adapter.main(
- [
- "base",
- "--repo",
- "/repo",
- "--out",
- "/out",
- "--name",
- "myrepo",
- "--run-id",
- "rid-base",
- "--depth",
- "2",
- "--source-sha",
- "abc123",
- ]
- )
- self.assertEqual(rf.calls[0]["depth_level"], 2)
-
- def test_main_enables_engine_console_logging(self):
- self._install()
- setup_logging = _Rec()
- with (
- patch.object(engine_adapter, "setup_logging", setup_logging),
- patch.dict(os.environ, {"CODEBOARDING_LOG_LEVEL": "DEBUG"}),
- ):
- engine_adapter.main(
- [
- "base",
- "--repo",
- "/repo",
- "--out",
- "/out",
- "--name",
- "myrepo",
- "--run-id",
- "rid-base",
- "--depth",
- "2",
- "--source-sha",
- "abc123",
- ]
- )
-
- self.assertEqual(setup_logging.calls, [{"default_level": "DEBUG"}])
-
- def test_main_sets_github_action_source(self):
- rf = _Rec()
- self._install(run_full=rf)
- with patch.dict(os.environ, {}, clear=True):
- engine_adapter.main(
- [
- "base",
- "--repo",
- "/repo",
- "--out",
- "/out",
- "--name",
- "myrepo",
- "--run-id",
- "rid-base",
- "--depth",
- "2",
- "--source-sha",
- "abc123",
- ]
- )
- self.assertEqual(os.environ["CODEBOARDING_SOURCE"], "github_action")
-
- def test_main_rejects_invalid_depth(self):
- # argparse enforces the structural range 1-10; the per-tier cap (free=3)
- # is applied later by the action/resolver, not here.
- for depth in ("0", "11", "x"):
- with self.subTest(depth=depth):
- with redirect_stderr(StringIO()):
- with self.assertRaises(SystemExit):
- engine_adapter.main(
- [
- "base",
- "--repo",
- "/repo",
- "--out",
- "/out",
- "--name",
- "myrepo",
- "--run-id",
- "rid-base",
- "--depth",
- depth,
- "--source-sha",
- "abc123",
- ]
- )
-
- def test_main_accepts_depth_four(self):
- # The action's accepted depth ceiling is 4 so a committed depth-4 baseline
- # is a first-class value review can inherit (the engine has no depth cap).
- rf = _Rec()
- self._install(run_full=rf)
- engine_adapter.main(
- [
- "base",
- "--repo",
- "/repo",
- "--out",
- "/out",
- "--name",
- "myrepo",
- "--run-id",
- "rid-base",
- "--depth",
- "4",
- "--source-sha",
- "abc123",
- ]
- )
- self.assertEqual(rf.calls[0]["depth_level"], 4)
-
- def test_head_uses_incremental(self):
- ri, rf = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = tempfile.mkdtemp()
- _write_model_valid_analysis(out, depth_level=1)
- buf = StringIO()
- with redirect_stdout(buf):
- engine_adapter.run_head("/repo", out, "r", "rid", 1, "head")
- self.assertEqual(len(ri.calls), 1)
- self.assertEqual(len(rf.calls), 0) # no fallback
- # Git-free: no base/target ref — Core diffs the seeded fingerprint itself.
- run_paths, run_context = ri.args[0]
- self.assertEqual(str(run_paths.repo_path), "/repo")
- self.assertEqual(str(run_paths.output_dir), out)
- self.assertEqual(run_context.run_id, "rid")
- self.assertIn("head_analysis_mode=incremental", buf.getvalue())
-
- def test_head_force_full_skips_incremental(self):
- ri, rf = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = tempfile.mkdtemp()
- (Path(out) / "stale.json").write_text("{}")
- buf = StringIO()
-
- with redirect_stdout(buf):
- engine_adapter.run_head("/repo", out, "r", "rid", 2, "head", force_full=True)
-
- self.assertEqual(len(ri.calls), 0)
- self.assertEqual(len(rf.calls), 1)
- self.assertEqual(rf.calls[0]["depth_level"], 2)
- self.assertEqual(rf.calls[0]["source_sha"], "head")
- self.assertFalse((Path(out) / "stale.json").exists())
- self.assertIn("head_analysis_mode=full", buf.getvalue())
-
- def test_head_falls_back_to_full_on_cache_miss(self):
- analysis, IncMiss, _ = self._install() # install once so the exception class identity matches
- rf = _Rec()
- analysis.run_full = rf
- analysis.run_incremental = _Rec(raises=IncMiss)
- engine_adapter.run_full = analysis.run_full
- engine_adapter.run_incremental = analysis.run_incremental
- out = tempfile.mkdtemp()
- _write_model_valid_analysis(out, depth_level=3)
- (Path(out) / "stale.json").write_text("{}") # must be wiped before the full run
- (Path(out) / "health").mkdir()
- (Path(out) / "health" / "stale.json").write_text("{}")
- buf = StringIO()
- with redirect_stdout(buf):
- engine_adapter.run_head("/repo", out, "r", "rid", 3, "head")
- self.assertEqual(len(rf.calls), 1) # fell back to full
- self.assertEqual(rf.calls[0]["depth_level"], 3)
- self.assertFalse((Path(out) / "stale.json").exists()) # head dir wiped before full
- self.assertFalse((Path(out) / "health").exists()) # nested stale artifacts wiped too
- self.assertIn("head_analysis_mode=full", buf.getvalue())
-
- def test_head_falls_back_to_full_on_baseline_unavailable(self):
- analysis, _, BaseUnavail = self._install() # the other warm-start failure must also fall back
- rf = _Rec()
- analysis.run_full = rf
- analysis.run_incremental = _Rec(raises=BaseUnavail)
- engine_adapter.run_full = analysis.run_full
- engine_adapter.run_incremental = analysis.run_incremental
- out = tempfile.mkdtemp()
- _write_model_valid_analysis(out, depth_level=1)
- engine_adapter.run_head("/repo", out, "r", "rid", 1, "head")
- self.assertEqual(len(rf.calls), 1) # BaselineUnavailableError also triggers the full re-run
-
- def test_head_rebuilds_analysis_rejected_by_core_model(self):
- ri, rf = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = Path(tempfile.mkdtemp())
- (out / "analysis.json").write_text(
- json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 3}}),
- encoding="utf-8",
- )
- (out / "stale.json").write_text("{}", encoding="utf-8")
- buf = StringIO()
-
- with patch.object(engine_adapter, "UnifiedAnalysisJson", _RejectingUnifiedAnalysisJson):
- with redirect_stdout(buf):
- engine_adapter.run_head("/repo", str(out), "r", "rid", 1, "head")
-
- self.assertEqual(len(ri.calls), 0)
- self.assertEqual(len(rf.calls), 1)
- self.assertEqual(rf.calls[0]["depth_level"], 3)
- self.assertFalse((out / "stale.json").exists())
- self.assertIn("could not load baseline analysis.json", buf.getvalue())
- self.assertIn("head_analysis_mode=full", buf.getvalue())
-
-
-class TestValidateBase(_Base):
- def test_validate_base_accepts_matching_commit(self):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(json.dumps({"metadata": {"commit_hash": "abc123"}}), encoding="utf-8")
-
- ok, message = engine_adapter.validate_base_analysis(path, "abc123")
-
- self.assertTrue(ok)
- self.assertIn("matches", message)
-
- def test_validate_base_accepts_mismatched_commit(self):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(json.dumps({"metadata": {"commit_hash": "old"}}), encoding="utf-8")
-
- ok, message = engine_adapter.validate_base_analysis(path, "new")
-
- self.assertTrue(ok)
- self.assertIn("old", message)
- self.assertIn("new", message)
-
- def test_validate_base_accepts_docs_only_bot_commit(self):
- with tempfile.TemporaryDirectory() as tmp:
- repo = Path(tmp) / "repo"
- repo.mkdir()
- self._git(repo, "init")
- self._git(repo, "config", "user.name", "Test")
- self._git(repo, "config", "user.email", "test@example.com")
- (repo / "app.py").write_text("print('base')\n", encoding="utf-8")
- self._git(repo, "add", "app.py")
- self._git(repo, "commit", "-m", "base")
- base_sha = self._git(repo, "rev-parse", "HEAD").stdout.strip()
-
- (repo / ".codeboarding").mkdir()
- (repo / ".codeboarding" / "analysis.json").write_text(
- json.dumps({"metadata": {"commit_hash": base_sha}}),
- encoding="utf-8",
- )
- (repo / ".codeboarding" / "overview.md").write_text("overview\n", encoding="utf-8")
- (repo / "docs" / "development").mkdir(parents=True)
- (repo / "docs" / "development" / "architecture.md").write_text("overview\n", encoding="utf-8")
- self._git(repo, "add", ".codeboarding", "docs/development/architecture.md")
- self._git(repo, "commit", "-m", "docs bot")
- docs_sha = self._git(repo, "rev-parse", "HEAD").stdout.strip()
-
- cwd = os.getcwd()
- try:
- os.chdir(repo)
- ok, message = engine_adapter.validate_base_analysis(repo / ".codeboarding" / "analysis.json", docs_sha)
- finally:
- os.chdir(cwd)
-
- self.assertTrue(ok)
- self.assertIn("Using committed baseline", message)
-
- def test_validate_base_accepts_committed_baseline_even_after_code_drift(self):
- with tempfile.TemporaryDirectory() as tmp:
- repo = Path(tmp) / "repo"
- repo.mkdir()
- self._git(repo, "init")
- self._git(repo, "config", "user.name", "Test")
- self._git(repo, "config", "user.email", "test@example.com")
- (repo / "app.py").write_text("print('base')\n", encoding="utf-8")
- self._git(repo, "add", "app.py")
- self._git(repo, "commit", "-m", "base")
- base_sha = self._git(repo, "rev-parse", "HEAD").stdout.strip()
- (repo / ".codeboarding").mkdir()
- analysis_path = repo / ".codeboarding" / "analysis.json"
- analysis_path.write_text(json.dumps({"metadata": {"commit_hash": base_sha}}), encoding="utf-8")
- (repo / "app.py").write_text("print('changed')\n", encoding="utf-8")
- self._git(repo, "add", "app.py", ".codeboarding/analysis.json")
- self._git(repo, "commit", "-m", "code change")
- code_sha = self._git(repo, "rev-parse", "HEAD").stdout.strip()
-
- cwd = os.getcwd()
- try:
- os.chdir(repo)
- ok, message = engine_adapter.validate_base_analysis(analysis_path, code_sha)
- finally:
- os.chdir(cwd)
-
- self.assertTrue(ok)
- self.assertIn("Using committed baseline", message)
-
- def _git(self, repo, *args):
- return subprocess.run(
- ["git", *args],
- cwd=repo,
- check=True,
- text=True,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- )
-
- def test_validate_base_accepts_missing_commit(self):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(json.dumps({"metadata": {}}), encoding="utf-8")
-
- ok, message = engine_adapter.validate_base_analysis(path, "abc123")
-
- self.assertTrue(ok)
- self.assertIn("commit_hash", message)
-
- def test_validate_base_rejects_lossy_model_load(self):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(
- json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 2}}),
- encoding="utf-8",
- )
-
- with patch.object(engine_adapter, "UnifiedAnalysisJson", _LossyUnifiedAnalysisJson):
- ok, message = engine_adapter.validate_base_analysis(path, "abc123")
-
- self.assertFalse(ok)
- self.assertIn("could not load baseline analysis.json", message)
- self.assertIn("without schema changes", message)
- self.assertIn("full analysis", message)
-
- def test_main_validate_base_exit_codes(self):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(json.dumps({"metadata": {"commit_hash": "abc123"}}), encoding="utf-8")
-
- self.assertEqual(
- engine_adapter.main(["validate-base", "--analysis", str(path), "--expected-sha", "abc123"]),
- 0,
- )
- self.assertEqual(
- engine_adapter.main(["validate-base", "--analysis", str(path), "--expected-sha", "def456"]),
- 0,
- )
-
- def test_validate_base_accepts_matching_depth(self):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(
- json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 2}}),
- encoding="utf-8",
- )
-
- ok, message = engine_adapter.validate_base_analysis(path, "abc123", expected_depth=2)
-
- self.assertTrue(ok)
- self.assertIn("matches", message)
-
- def test_validate_base_rejects_deeper_baseline(self):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(
- json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 3}}),
- encoding="utf-8",
- )
-
- ok, message = engine_adapter.validate_base_analysis(path, "abc123", expected_depth=1)
-
- self.assertFalse(ok)
- self.assertIn("3", message) # baseline depth
- self.assertIn("1", message) # expected depth
-
- def test_validate_base_accepts_shallower_baseline(self):
- # The engine records the depth REACHED, not requested: a depth-2 run on
- # a repo that never expands persists depth_level 1. Rejecting it would
- # regenerate (computing 1 again) on every PR without converging.
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(
- json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 1}}),
- encoding="utf-8",
- )
-
- ok, _ = engine_adapter.validate_base_analysis(path, "abc123", expected_depth=3)
-
- self.assertTrue(ok)
-
- def test_validate_base_depth_checked_on_drift_path(self):
- # The deeper-baseline rejection must also apply when the commit matched
- # only via the docs-only-drift allowance, not just on exact SHA match.
- with tempfile.TemporaryDirectory() as tmp:
- repo = Path(tmp) / "repo"
- repo.mkdir()
- self._git(repo, "init")
- self._git(repo, "config", "user.name", "Test")
- self._git(repo, "config", "user.email", "test@example.com")
- (repo / "app.py").write_text("print('base')\n", encoding="utf-8")
- self._git(repo, "add", "app.py")
- self._git(repo, "commit", "-m", "base")
- base_sha = self._git(repo, "rev-parse", "HEAD").stdout.strip()
-
- (repo / ".codeboarding").mkdir()
- analysis_path = repo / ".codeboarding" / "analysis.json"
- analysis_path.write_text(
- json.dumps({"metadata": {"commit_hash": base_sha, "depth_level": 3}}),
- encoding="utf-8",
- )
- self._git(repo, "add", ".codeboarding")
- self._git(repo, "commit", "-m", "docs bot")
- docs_sha = self._git(repo, "rev-parse", "HEAD").stdout.strip()
-
- cwd = os.getcwd()
- try:
- os.chdir(repo)
- ok_drift, _ = engine_adapter.validate_base_analysis(analysis_path, docs_sha)
- ok_depth, message = engine_adapter.validate_base_analysis(analysis_path, docs_sha, expected_depth=1)
- finally:
- os.chdir(cwd)
-
- self.assertTrue(ok_drift) # drift alone is accepted...
- self.assertFalse(ok_depth) # ...but the depth check still applies
- self.assertIn("deeper", message)
-
- def test_validate_base_accepts_legacy_baseline_without_depth(self):
- # Missing or unparseable depth_level remains acceptable when the
- # installed Core model accepts the document.
- for metadata in (
- {"commit_hash": "abc123"},
- {"commit_hash": "abc123", "depth_level": "not-a-number"},
- ):
- with self.subTest(metadata=metadata):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(json.dumps({"metadata": metadata}), encoding="utf-8")
-
- ok, _ = engine_adapter.validate_base_analysis(path, "abc123", expected_depth=2)
-
- self.assertTrue(ok)
-
- def test_validate_base_without_expected_depth_ignores_depth(self):
- # No --expected-depth -> behavior unchanged even when depth_level disagrees.
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(
- json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 3}}),
- encoding="utf-8",
- )
-
- ok, message = engine_adapter.validate_base_analysis(path, "abc123")
-
- self.assertTrue(ok)
- self.assertIn("matches", message)
-
- def test_validate_base_accepts_depth_four_baseline(self):
- # The core fix: review inherits the committed baseline's depth, so a
- # depth-4 baseline validated at --expected-depth 4 is accepted (reused,
- # not regenerated). Validated at a shallower expected depth it is still
- # rejected (an explicit shallower depth_level input).
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(
- json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 4}}),
- encoding="utf-8",
- )
-
- ok_same, _ = engine_adapter.validate_base_analysis(path, "abc123", expected_depth=4)
- ok_shallower, message = engine_adapter.validate_base_analysis(path, "abc123", expected_depth=2)
-
- self.assertTrue(ok_same)
- self.assertFalse(ok_shallower)
- self.assertIn("deeper", message)
-
- def test_main_validate_base_expected_depth_exit_codes(self):
- # patch.dict: main() setdefaults CODEBOARDING_SOURCE; don't leak it.
- with patch.dict(os.environ), tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(
- json.dumps({"metadata": {"commit_hash": "abc123", "depth_level": 2}}),
- encoding="utf-8",
- )
-
- self.assertEqual(
- engine_adapter.main(
- ["validate-base", "--analysis", str(path), "--expected-sha", "abc123", "--expected-depth", "2"]
- ),
- 0,
- )
- self.assertEqual(
- engine_adapter.main(
- ["validate-base", "--analysis", str(path), "--expected-sha", "abc123", "--expected-depth", "1"]
- ),
- 1,
- )
- # depth 4 is now an accepted value (against a depth-2 baseline a
- # shallower-or-equal expected depth passes the depth check).
- self.assertEqual(
- engine_adapter.main(
- ["validate-base", "--analysis", str(path), "--expected-sha", "abc123", "--expected-depth", "4"]
- ),
- 0,
- )
- with redirect_stderr(StringIO()):
- with self.assertRaises(SystemExit): # depth outside 1-10 rejected by argparse
- engine_adapter.main(
- ["validate-base", "--analysis", str(path), "--expected-sha", "abc123", "--expected-depth", "11"]
- )
-
-
-class TestSeed(_Base):
- """run_seed must analyze, cluster, then save — in that order, same results object.
-
- The save-after-clustering order is the point of the subcommand: the engine
- persists a pkl on LSP teardown BEFORE clustering, and a pkl saved then has
- no cluster baseline, which is exactly the state that forces the head run
- into a full-analysis fallback.
- """
-
- def _install(self, fail_at=None):
- log = []
- results = object()
-
- def get_static_analysis(repo_path, cache_dir, skip_cache=False, source_sha=None):
- log.append(("analyze", str(repo_path), str(cache_dir), source_sha))
- if fail_at == "analyze":
- raise RuntimeError("boom")
- return results
-
- def build_all_cluster_results(res):
- log.append(("cluster", res))
- if fail_at == "cluster":
- raise RuntimeError("boom")
- return {"python": types.SimpleNamespace(clusters={1: {"a"}, 2: {"b"}})}
-
- class _Cache:
- def __init__(self, artifact_dir, repo_root):
- log.append(("cache_init", str(artifact_dir), str(repo_root)))
-
- def save(self, res, source_sha=None):
- log.append(("save", res, source_sha))
-
- sa = _mod("static_analyzer", get_static_analysis=get_static_analysis)
- sa.cluster_helpers = _mod(
- "static_analyzer.cluster_helpers", build_all_cluster_results=build_all_cluster_results
- )
- sa.analysis_cache = _mod("static_analyzer.analysis_cache", StaticAnalysisCache=_Cache)
- engine_adapter.get_static_analysis = get_static_analysis
- engine_adapter.build_all_cluster_results = build_all_cluster_results
- engine_adapter.StaticAnalysisCache = _Cache
- return log, results
-
- def test_seed_analyzes_clusters_then_saves(self):
- log, results = self._install()
- engine_adapter.run_seed("/repo", "/out", "abc123")
- self.assertEqual(
- log,
- [
- ("analyze", "/repo", "/out", "abc123"),
- ("cluster", results),
- ("cache_init", "/out", "/repo"),
- ("save", results, "abc123"),
- ],
- )
-
- def test_seed_propagates_engine_errors(self):
- # Fail-open lives in the action step; run_seed itself must not swallow.
- for stage in ("analyze", "cluster"):
- with self.subTest(stage=stage):
- log, _ = self._install(fail_at=stage)
- with self.assertRaises(RuntimeError):
- engine_adapter.run_seed("/repo", "/out", "abc123")
- self.assertNotIn("save", [e[0] for e in log])
- self.tearDown()
-
- def test_main_seed_wires_args(self):
- log, _ = self._install()
- rc = engine_adapter.main(["seed", "--repo", "/r", "--out", "/o", "--source-sha", "s1"])
- self.assertEqual(rc, 0)
- self.assertEqual(log[0], ("analyze", "/r", "/o", "s1"))
- self.assertEqual(log[-1][0], "save")
-
-
-class TestHealth(_Base):
- def _install_health(self, report):
- class Severity:
- WARNING, CRITICAL = "warning", "critical"
-
- class _Cache:
- def __init__(self, artifact_dir, repo_root):
- pass
-
- def get(self):
- return object() # non-None static analysis
-
- _mod("health.models", Severity=Severity)
- _mod("health.runner", run_health_checks=lambda sa, repo_name, repo_path: report)
- _mod(
- "health",
- )
- _mod("static_analyzer.analysis_cache", StaticAnalysisCache=_Cache)
- _mod(
- "static_analyzer",
- )
- engine_adapter.Severity = Severity
- engine_adapter.run_health_checks = lambda sa, repo_name, repo_path: report
- engine_adapter.StaticAnalysisCache = _Cache
- return Severity
-
- def test_counts_warning_and_critical(self):
- Sev = self._install_health(report=None)
-
- class FG:
- def __init__(self, sev, n):
- self.severity, self.entities = sev, list(range(n))
-
- class CS:
- finding_groups = [FG(Sev.WARNING, 2), FG(Sev.CRITICAL, 1), FG("info", 5)]
-
- report = types.SimpleNamespace(check_summaries=[CS()])
- self._install_health(report=report)
- self.assertEqual(engine_adapter.run_health("/art", "/repo", "r"), 3) # 2 warnings + 1 critical, info ignored
-
- def test_prefers_written_health_report(self):
- artifact_dir = Path(tempfile.mkdtemp())
- report_dir = artifact_dir / "health"
- report_dir.mkdir()
- (report_dir / "health_report.json").write_text(
- """
- {
- "check_summaries": [
- {"finding_groups": [
- {"severity": "warning", "entities": [{}, {}]},
- {"severity": "critical", "entities": [{}]},
- {"severity": "info", "entities": [{}, {}, {}, {}, {}]}
- ]}
- ]
- }
- """,
- encoding="utf-8",
- )
- self.assertEqual(engine_adapter.run_health(str(artifact_dir), "/repo", "r"), 3)
-
- def test_malformed_health_report_falls_back(self):
- self._install_health(report=None)
- artifact_dir = Path(tempfile.mkdtemp())
- report_dir = artifact_dir / "health"
- report_dir.mkdir()
- (report_dir / "health_report.json").write_text("[]", encoding="utf-8")
- self.assertEqual(engine_adapter.run_health(str(artifact_dir), "/repo", "r"), 0)
-
- def test_missing_module_yields_zero(self):
- # Health failures are best-effort: return 0, never raise.
- class _BrokenCache:
- def __init__(self, *args, **kwargs):
- raise ImportError("missing health dependency")
-
- old_cache = engine_adapter.StaticAnalysisCache
- engine_adapter.StaticAnalysisCache = _BrokenCache
- try:
- self.assertEqual(engine_adapter.run_health("/art", "/repo", "r"), 0)
- finally:
- engine_adapter.StaticAnalysisCache = old_cache
-
-
-class TestQuotaExhausted(_Base):
- def test_detects_402_status_attr(self):
- class APIErr(Exception):
- status_code = 402
-
- self.assertTrue(engine_adapter._is_quota_exhausted(APIErr("nope")))
-
- def test_detects_status_attr(self):
- class FunctionUrlErr(Exception):
- status = 402
-
- self.assertTrue(engine_adapter._is_quota_exhausted(FunctionUrlErr("nope")))
-
- def test_detects_marker_string(self):
- exc = RuntimeError("upstream said: Resource exhausted: token limit reached")
- self.assertTrue(engine_adapter._is_quota_exhausted(exc))
-
- def test_detects_in_cause_chain(self):
- inner = RuntimeError("Resource exhausted: token limit reached")
- try:
- raise ValueError("wrapped") from inner
- except ValueError as e:
- self.assertTrue(engine_adapter._is_quota_exhausted(e))
-
- def test_other_errors_not_flagged(self):
- self.assertFalse(engine_adapter._is_quota_exhausted(RuntimeError("disk full")))
-
- class OtherStatus(Exception):
- status_code = 500
-
- self.assertFalse(engine_adapter._is_quota_exhausted(OtherStatus("boom")))
-
- def _install_raising(self, exc):
- analysis = _mod(
- "codeboarding_workflows.analysis",
- run_full=_Rec(raises=exc),
- run_incremental=_Rec(),
- BaselineUnavailableError=type("BaselineUnavailableError", (Exception,), {}),
- )
- pkg = _mod("codeboarding_workflows")
- pkg.analysis = analysis
- excmod = _mod(
- "diagram_analysis.exceptions",
- IncrementalCacheMissingError=type("IncrementalCacheMissingError", (Exception,), {}),
- )
- da = _mod("diagram_analysis")
- da.exceptions = excmod
- engine_adapter.run_full = analysis.run_full
- engine_adapter.run_incremental = analysis.run_incremental
- engine_adapter.BaselineUnavailableError = analysis.BaselineUnavailableError
- engine_adapter.IncrementalCacheMissingError = excmod.IncrementalCacheMissingError
-
- def _run_base(self):
- return engine_adapter.main(
- [
- "base",
- "--repo",
- "/r",
- "--out",
- "/o",
- "--name",
- "n",
- "--run-id",
- "rid",
- "--depth",
- "2",
- "--source-sha",
- "abc123",
- ]
- )
-
- def test_main_drops_sentinel_on_quota_error(self):
- class APIErr(Exception):
- status_code = 402
-
- self._install_raising(APIErr)
- sentinel = Path(tempfile.mkdtemp()) / "cb-quota-exhausted"
- with patch.dict(os.environ, {"CB_QUOTA_SENTINEL": str(sentinel)}):
- with redirect_stderr(StringIO()):
- with self.assertRaises(APIErr): # re-raised so the step still fails
- self._run_base()
- self.assertTrue(sentinel.exists(), "quota sentinel should be written")
-
- def test_main_no_sentinel_on_other_error(self):
- self._install_raising(RuntimeError)
- sentinel = Path(tempfile.mkdtemp()) / "cb-quota-exhausted"
- with patch.dict(os.environ, {"CB_QUOTA_SENTINEL": str(sentinel)}):
- with redirect_stderr(StringIO()):
- with self.assertRaises(RuntimeError):
- self._run_base()
- self.assertFalse(sentinel.exists(), "non-quota errors must not write the sentinel")
-
-
-class TestEngineRequired(_Base):
- """A missing/too-old engine (RunPaths imported as None) fails the analysis
- subcommands with a clear message, while metadata-only subcommands still run."""
-
- def _argv(self, cmd):
- run = ["--repo", "/r", "--out", "/o", "--name", "n", "--run-id", "id", "--source-sha", "s"]
- return {
- "base": [cmd, *run, "--depth", "2"],
- "seed": [cmd, "--repo", "/r", "--out", "/o", "--source-sha", "s"],
- "head": [cmd, *run, "--depth", "2"],
- "validate-base": [cmd, "--analysis", "/a.json", "--expected-sha", "abc123"],
- "analyze": [cmd, *run, "--depth", "2"],
- "render": [cmd, "--analysis", "/a.json", "--out", "/o", "--repo-name", "n", "--repo-ref", "r"],
- }[cmd]
-
- def test_engine_commands_fail_clearly_when_engine_missing(self):
- for cmd in engine_adapter._ENGINE_COMMANDS:
- with (
- self.subTest(cmd=cmd),
- patch.object(engine_adapter, "RunPaths", None),
- patch.object(engine_adapter, "UnifiedAnalysisJson", None),
- ):
- with self.assertRaises(RuntimeError) as ctx:
- engine_adapter.main(self._argv(cmd))
- msg = str(ctx.exception)
- self.assertIn(cmd, msg)
- self.assertIn("too old", msg)
- self.assertIn("codeboarding_version", msg)
-
- def test_metadata_command_runs_without_engine(self):
- with tempfile.TemporaryDirectory() as d:
- path = Path(d) / "analysis.json"
- path.write_text(json.dumps({"metadata": {"commit_hash": "abc1234"}}))
- with patch.object(engine_adapter, "RunPaths", None), redirect_stdout(StringIO()):
- rc = engine_adapter.main(["baseline-info", "--analysis", str(path)])
- self.assertEqual(rc, 0)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/test_install_sync_artifacts.py b/tests/test_install_sync_artifacts.py
new file mode 100644
index 0000000..d2b3dde
--- /dev/null
+++ b/tests/test_install_sync_artifacts.py
@@ -0,0 +1,94 @@
+"""Regression tests for selective sync-artifact installation."""
+
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
+
+import install_sync_artifacts as isa
+
+
+class InstallSyncArtifactsTests(unittest.TestCase):
+ def test_preserves_user_config_while_replacing_generated_files(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ output = root / ".codeboarding"
+ health = output / "health"
+ docs = root / "docs"
+ analysis = root / "analysis"
+ analysis_health = analysis / "health"
+ for directory in (health, docs, analysis_health):
+ directory.mkdir(parents=True)
+
+ preserved = {
+ output / ".codeboardingignore": "ignore me\n",
+ output / "health_config.json": '{"root": true}\n',
+ health / ".healthignore": "known issue\n",
+ health / "health_config.json": '{"health": true}\n',
+ output / "notes.txt": "user notes\n",
+ }
+ for path, content in preserved.items():
+ path.write_text(content, encoding="utf-8")
+
+ (output / "stale-component.md").write_text("stale\n", encoding="utf-8")
+ (output / "codeboarding_version.json").write_text("stale\n", encoding="utf-8")
+ (health / "health_report.json").write_text("old report\n", encoding="utf-8")
+ (docs / "overview.md").write_text("# New overview\n", encoding="utf-8")
+ analysis_path = analysis / "analysis.json"
+ analysis_path.write_text('{"new": true}\n', encoding="utf-8")
+ (analysis / "fingerprint.json").write_text('{"fingerprint": true}\n', encoding="utf-8")
+ (analysis_health / "health_report.json").write_text("new report\n", encoding="utf-8")
+
+ stage_paths = isa.install_sync_artifacts(
+ output_dir=output,
+ docs_dir=docs,
+ analysis_path=analysis_path,
+ analysis_dir=analysis,
+ )
+
+ for path, content in preserved.items():
+ self.assertEqual(path.read_text(encoding="utf-8"), content)
+ self.assertFalse((output / "stale-component.md").exists())
+ self.assertFalse((output / "codeboarding_version.json").exists())
+ self.assertEqual((output / "overview.md").read_text(encoding="utf-8"), "# New overview\n")
+ self.assertEqual((output / "analysis.json").read_text(encoding="utf-8"), '{"new": true}\n')
+ self.assertEqual((health / "health_report.json").read_text(encoding="utf-8"), "new report\n")
+
+ staged = set(stage_paths)
+ self.assertIn(output / "stale-component.md", staged)
+ self.assertIn(output / "codeboarding_version.json", staged)
+ self.assertIn(output / "overview.md", staged)
+ self.assertIn(output / "analysis.json", staged)
+ self.assertNotIn(output / ".codeboardingignore", staged)
+ self.assertNotIn(health / ".healthignore", staged)
+ self.assertNotIn(health / "health_config.json", staged)
+
+ def test_rejects_empty_render_output_before_modifying_destination(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ output = root / ".codeboarding"
+ docs = root / "docs"
+ analysis = root / "analysis"
+ output.mkdir()
+ docs.mkdir()
+ analysis.mkdir()
+ existing = output / "overview.md"
+ existing.write_text("keep on failure\n", encoding="utf-8")
+ analysis_path = analysis / "analysis.json"
+ analysis_path.write_text("{}\n", encoding="utf-8")
+
+ with self.assertRaises(isa.ArtifactInstallError):
+ isa.install_sync_artifacts(
+ output_dir=output,
+ docs_dir=docs,
+ analysis_path=analysis_path,
+ analysis_dir=analysis,
+ )
+
+ self.assertEqual(existing.read_text(encoding="utf-8"), "keep on failure\n")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_render_sync_docs.py b/tests/test_render_sync_docs.py
new file mode 100644
index 0000000..766b0f8
--- /dev/null
+++ b/tests/test_render_sync_docs.py
@@ -0,0 +1,105 @@
+"""Smoke tests for scripts/render_sync_docs.py."""
+
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from types import ModuleType
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
+
+stub_pkg = ModuleType("codeboarding_workflows")
+stub_rendering = ModuleType("codeboarding_workflows.rendering")
+stub_rendering.render_docs = lambda *args, **kwargs: None
+stub_pkg.rendering = stub_rendering
+sys.modules["codeboarding_workflows"] = stub_pkg
+sys.modules["codeboarding_workflows.rendering"] = stub_rendering
+
+import render_sync_docs as rsd # noqa: E402
+
+
+class RenderSyncDocsTests(unittest.TestCase):
+ def setUp(self) -> None:
+ self.render_calls = []
+
+ def _make_fake_render(self, with_overview: bool = True):
+ def _render(
+ analysis,
+ repo_name,
+ repo_ref,
+ temp_dir,
+ format=".md",
+ root_name="overview",
+ ):
+ self.render_calls.append((analysis, repo_name, repo_ref, temp_dir, format, root_name))
+ out = Path(temp_dir)
+ out.mkdir(parents=True, exist_ok=True)
+ if with_overview:
+ (out / "overview.md").write_text("# Overview\n", encoding="utf-8")
+ (out / "api.md").write_text("# API\n", encoding="utf-8")
+ (out / "zeta.md").write_text("# Zeta\n", encoding="utf-8")
+
+ return _render
+
+ def test_concat_prefers_overview_first_and_appends_sorted(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ analysis = root / "analysis.json"
+ analysis.write_text("{}", encoding="utf-8")
+ output = root / "docs"
+ architecture = root / "architecture.md"
+ with patch.object(rsd, "render_docs", new=self._make_fake_render(True)):
+ rsd.main(
+ [
+ "--analysis",
+ str(analysis),
+ "--output-dir",
+ str(output),
+ "--repo-name",
+ "org/repo",
+ "--repo-ref",
+ "abc123",
+ "--format",
+ ".md",
+ "--architecture-file",
+ str(architecture),
+ ]
+ )
+ result = architecture.read_text(encoding="utf-8")
+ self.assertIn("# Overview", result)
+ self.assertIn("# API", result)
+ self.assertIn("# Zeta", result)
+ self.assertLess(result.index("# Overview"), result.index("# API"))
+ self.assertLess(result.index("# API"), result.index("# Zeta"))
+ self.assertEqual(len(self.render_calls), 1)
+
+ def test_missing_overview_fails(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ analysis = root / "analysis.json"
+ analysis.write_text("{}", encoding="utf-8")
+ output = root / "docs"
+ architecture = root / "architecture.md"
+ with patch.object(rsd, "render_docs", new=self._make_fake_render(False)):
+ with self.assertRaises(SystemExit):
+ rsd.main(
+ [
+ "--analysis",
+ str(analysis),
+ "--output-dir",
+ str(output),
+ "--repo-name",
+ "org/repo",
+ "--repo-ref",
+ "abc123",
+ "--format",
+ ".md",
+ "--architecture-file",
+ str(architecture),
+ ]
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_submit_feedback.py b/tests/test_submit_feedback.py
deleted file mode 100644
index 29e1e75..0000000
--- a/tests/test_submit_feedback.py
+++ /dev/null
@@ -1,196 +0,0 @@
-"""Unit tests for scripts/submit_feedback.py — /codeboarding-feedback capture."""
-
-import io
-import json
-import sys
-import unittest
-from contextlib import redirect_stdout
-from pathlib import Path
-from unittest import mock
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
-import submit_feedback as sf # noqa: E402
-
-COMMAND = "/codeboarding-feedback"
-HOST = "https://us.i.posthog.com"
-
-
-def base_env(**overrides):
- env = {
- "COMMENT_BODY": f"{COMMAND} the diagram is great",
- "FEEDBACK_COMMAND": COMMAND,
- "REPOSITORY": "octo/repo",
- "REPOSITORY_ID": "555",
- "ISSUE_NUMBER": "42",
- "COMMENT_ID": "99",
- "COMMENT_URL": "https://github.com/octo/repo/pull/42#issuecomment-99",
- "AUTHOR_ASSOC": "CONTRIBUTOR",
- "SENDER_LOGIN": "octocat",
- "SENDER_ID": "1234",
- "GITHUB_RUN_ID": "777",
- "RUN_ATTEMPT": "1",
- "ACTION_REF": "v1",
- }
- env.update(overrides)
- return env
-
-
-class TestExtractFeedback(unittest.TestCase):
- def test_extracts_text_after_command(self):
- self.assertEqual(sf.extract_feedback(f"{COMMAND} hello there", COMMAND), "hello there")
-
- def test_preserves_multiline_feedback(self):
- body = f"{COMMAND} first line\nsecond line\n\nfourth"
- self.assertEqual(sf.extract_feedback(body, COMMAND), "first line\nsecond line\n\nfourth")
-
- def test_command_only_yields_empty(self):
- self.assertEqual(sf.extract_feedback(COMMAND, COMMAND), "")
- self.assertEqual(sf.extract_feedback(f"{COMMAND} ", COMMAND), "")
-
- def test_command_on_its_own_line_then_body(self):
- self.assertEqual(sf.extract_feedback(f"{COMMAND}\nthe body", COMMAND), "the body")
-
- def test_leading_whitespace_and_crlf_normalized(self):
- self.assertEqual(sf.extract_feedback(f" {COMMAND} a\r\nb\r\n", COMMAND), "a\nb")
-
- def test_wrong_command_yields_empty(self):
- self.assertEqual(sf.extract_feedback("/codeboarding run it", COMMAND), "")
- self.assertEqual(sf.extract_feedback(f"{COMMAND}-typo hi", COMMAND), "")
-
-
-class TestCapFeedback(unittest.TestCase):
- def test_short_text_not_truncated(self):
- self.assertEqual(sf.cap_feedback("abc", 10), ("abc", 3, False))
-
- def test_long_text_capped_and_marked(self):
- capped, length, truncated = sf.cap_feedback("x" * 50, 10)
- self.assertEqual(capped, "x" * 10)
- self.assertEqual(length, 50)
- self.assertTrue(truncated)
-
-
-class TestOptOut(unittest.TestCase):
- def test_do_not_track_disables(self):
- self.assertTrue(sf.telemetry_disabled({"DO_NOT_TRACK": "1"}))
- self.assertTrue(sf.telemetry_disabled({"DO_NOT_TRACK": "true"}))
-
- def test_codeboarding_telemetry_false_disables(self):
- self.assertTrue(sf.telemetry_disabled({"CODEBOARDING_TELEMETRY": "false"}))
-
- def test_default_enabled(self):
- self.assertFalse(sf.telemetry_disabled({}))
-
-
-class TestResolvers(unittest.TestCase):
- def test_key_and_host_defaults(self):
- self.assertEqual(sf.resolve_key({}), sf.DEFAULT_POSTHOG_KEY)
- self.assertEqual(sf.resolve_host({}), sf.DEFAULT_POSTHOG_HOST)
-
- def test_host_override_strips_trailing_slash(self):
- self.assertEqual(
- sf.resolve_host({"CODEBOARDING_POSTHOG_HOST": "https://eu.example.com/"}), "https://eu.example.com"
- )
-
- def test_max_chars_invalid_falls_back(self):
- self.assertEqual(sf.resolve_max_chars({"FEEDBACK_MAX_CHARS": "nope"}), sf.DEFAULT_MAX_CHARS)
- self.assertEqual(sf.resolve_max_chars({"FEEDBACK_MAX_CHARS": "0"}), sf.DEFAULT_MAX_CHARS)
- self.assertEqual(sf.resolve_max_chars({"FEEDBACK_MAX_CHARS": "25"}), 25)
-
- def test_distinct_id_prefers_sender_then_run(self):
- self.assertEqual(sf.distinct_id({"SENDER_ID": "5"}), "github-user:5")
- self.assertEqual(sf.distinct_id({"GITHUB_RUN_ID": "9"}), "github-run:9")
-
-
-class TestBuildPayload(unittest.TestCase):
- def test_empty_feedback_returns_none(self):
- self.assertIsNone(sf.build_payload(base_env(COMMENT_BODY=COMMAND)))
-
- def test_payload_shape(self):
- payload = sf.build_payload(base_env())
- self.assertEqual(payload["event"], "codeboarding_feedback_submitted")
- self.assertEqual(payload["distinct_id"], "github-user:1234")
- self.assertEqual(payload["api_key"], sf.DEFAULT_POSTHOG_KEY)
- props = payload["properties"]
- self.assertEqual(props["source"], "github_action_feedback")
- self.assertEqual(props["command"], COMMAND)
- self.assertEqual(props["feedback_text"], "the diagram is great")
- self.assertEqual(props["feedback_length"], len("the diagram is great"))
- self.assertFalse(props["feedback_truncated"])
- self.assertEqual(props["repository"], "octo/repo")
- self.assertEqual(props["repository_id"], "555")
- self.assertEqual(props["pr_number"], "42")
- self.assertEqual(props["comment_id"], "99")
- self.assertEqual(props["author_association"], "CONTRIBUTOR")
- self.assertEqual(props["sender_login"], "octocat")
- self.assertEqual(props["run_id"], "777")
-
- def test_truncation_recorded_in_payload(self):
- payload = sf.build_payload(base_env(COMMENT_BODY=f"{COMMAND} " + "y" * 50, FEEDBACK_MAX_CHARS="10"))
- props = payload["properties"]
- self.assertEqual(len(props["feedback_text"]), 10)
- self.assertEqual(props["feedback_length"], 50)
- self.assertTrue(props["feedback_truncated"])
-
- def test_optional_props_omitted_when_absent(self):
- payload = sf.build_payload({"COMMENT_BODY": f"{COMMAND} hi", "SENDER_ID": "1"})
- self.assertNotIn("repository", payload["properties"])
- self.assertNotIn("comment_url", payload["properties"])
-
-
-class TestMain(unittest.TestCase):
- def _run(self, env):
- with mock.patch.object(sf.urllib.request, "urlopen") as urlopen:
- urlopen.return_value.__enter__.return_value.status = 200
- buf = io.StringIO()
- with redirect_stdout(buf):
- rc = sf.main(env)
- return rc, urlopen, buf.getvalue()
-
- def test_sends_expected_json_shape(self):
- rc, urlopen, _ = self._run(base_env())
- self.assertEqual(rc, 0)
- urlopen.assert_called_once()
- request = urlopen.call_args.args[0]
- self.assertEqual(request.full_url, f"{HOST}/i/v0/e/")
- self.assertEqual(request.get_method(), "POST")
- self.assertEqual(request.headers.get("Content-type"), "application/json")
- body = json.loads(request.data)
- self.assertEqual(body["event"], "codeboarding_feedback_submitted")
- self.assertEqual(body["distinct_id"], "github-user:1234")
- self.assertEqual(body["properties"]["feedback_text"], "the diagram is great")
-
- def test_host_override_used(self):
- _, urlopen, _ = self._run(base_env(CODEBOARDING_POSTHOG_HOST="https://eu.example.com"))
- request = urlopen.call_args.args[0]
- self.assertEqual(request.full_url, "https://eu.example.com/i/v0/e/")
-
- def test_do_not_track_skips_sending(self):
- _, urlopen, out = self._run(base_env(DO_NOT_TRACK="1"))
- urlopen.assert_not_called()
- self.assertIn("disabled", out)
-
- def test_telemetry_false_skips_sending(self):
- _, urlopen, _ = self._run(base_env(CODEBOARDING_TELEMETRY="false"))
- urlopen.assert_not_called()
-
- def test_empty_feedback_not_sent(self):
- _, urlopen, out = self._run(base_env(COMMENT_BODY=COMMAND))
- urlopen.assert_not_called()
- self.assertIn("nothing to send", out)
-
- def test_does_not_print_feedback_text(self):
- secret = "PLEASE_DO_NOT_LEAK_THIS_abc123"
- _, _, out = self._run(base_env(COMMENT_BODY=f"{COMMAND} {secret}"))
- self.assertNotIn(secret, out)
-
- def test_network_failure_is_swallowed(self):
- with mock.patch.object(sf.urllib.request, "urlopen", side_effect=sf.urllib.error.URLError("down")):
- buf = io.StringIO()
- with redirect_stdout(buf):
- rc = sf.main(base_env())
- self.assertEqual(rc, 0)
- self.assertIn("ignoring", buf.getvalue())
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/test_sync_subcommands.py b/tests/test_sync_subcommands.py
deleted file mode 100644
index bad730b..0000000
--- a/tests/test_sync_subcommands.py
+++ /dev/null
@@ -1,719 +0,0 @@
-"""Smoke tests for the sync-mode subcommands of scripts/engine_adapter.py (analyze,
-render, concat) with stubbed engine modules — ported from the standalone
-docs-action's test_docs_engine.py. Seed tests are not ported: engine_adapter's seed
-is byte-identical and already covered by tests/test_engine_adapter.py."""
-
-import json
-import os
-import subprocess
-import sys
-import tempfile
-import types
-import unittest
-from contextlib import redirect_stderr, redirect_stdout
-from io import StringIO
-from pathlib import Path
-from unittest.mock import patch
-
-
-def _preload(name, **attrs):
- module = types.ModuleType(name)
- for key, value in attrs.items():
- setattr(module, key, value)
- sys.modules[name] = module
- return module
-
-
-class _InitialBaselineUnavailableError(Exception):
- pass
-
-
-class _InitialIncrementalCacheMissingError(Exception):
- pass
-
-
-class _InitialSeverity:
- WARNING, CRITICAL = "warning", "critical"
-
-
-class _InitialStaticAnalysisCache:
- def __init__(self, *args, **kwargs):
- pass
-
- def get(self):
- return None
-
- def save(self, *args, **kwargs):
- pass
-
-
-class _RunPaths:
- def __init__(self, repo_path=None, output_dir=None, project_name=None):
- self.repo_path, self.output_dir, self.project_name = repo_path, output_dir, project_name
-
-
-class _RunContext:
- def __init__(self, run_id=None, log_path=None, repo_dir=None):
- self.run_id, self.log_path, self.repo_dir = run_id, log_path, repo_dir
-
-
-class _InitialUnifiedAnalysisJson:
- def __init__(self, data):
- self.data = data
-
- @classmethod
- def model_validate(cls, data):
- return cls(data)
-
- def model_dump(self, **kwargs):
- return self.data
-
-
-class _LossyUnifiedAnalysisJson(_InitialUnifiedAnalysisJson):
- def model_dump(self, **kwargs):
- return {"normalized": True}
-
-
-analysis = _preload(
- "codeboarding_workflows.analysis",
- run_full=lambda *a, **k: "OUT",
- run_incremental=lambda *a, **k: "OUT",
- BaselineUnavailableError=_InitialBaselineUnavailableError,
-)
-pkg = _preload("codeboarding_workflows")
-pkg.analysis = analysis
-rendering = _preload("codeboarding_workflows.rendering", render_docs=lambda *args, **kwargs: None)
-pkg.rendering = rendering
-exc = _preload("diagram_analysis.exceptions", IncrementalCacheMissingError=_InitialIncrementalCacheMissingError)
-da = _preload("diagram_analysis", RunPaths=_RunPaths, RunContext=_RunContext)
-da.exceptions = exc
-_preload("diagram_analysis.analysis_json", UnifiedAnalysisJson=_InitialUnifiedAnalysisJson)
-_preload("diagram_analysis.io_utils", write_fingerprint=lambda *a, **k: None)
-_preload("logging_config", setup_logging=lambda **kwargs: None)
-_preload("agents.content_hash", hash_repo_source_files=lambda *a, **k: {})
-_preload("agents")
-_preload("health.models", Severity=_InitialSeverity)
-_preload("health.runner", run_health_checks=lambda *args, **kwargs: None)
-_preload("health")
-_preload("static_analyzer", get_static_analysis=lambda *args, **kwargs: {})
-_preload("static_analyzer.analysis_cache", StaticAnalysisCache=_InitialStaticAnalysisCache)
-_preload("static_analyzer.cluster_helpers", build_all_cluster_results=lambda *args, **kwargs: {})
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
-import engine_adapter # noqa: E402
-
-_STUBBED = [
- "agents",
- "agents.content_hash",
- "codeboarding_workflows",
- "codeboarding_workflows.analysis",
- "codeboarding_workflows.rendering",
- "diagram_analysis",
- "diagram_analysis.analysis_json",
- "diagram_analysis.exceptions",
- "diagram_analysis.io_utils",
- "logging_config",
- "static_analyzer",
- "static_analyzer.analysis_cache",
- "static_analyzer.cluster_helpers",
-]
-
-
-class _Rec:
- def __init__(self, ret="OUT", raises=None):
- self.calls = []
- self._ret = ret
- self._raises = raises
-
- def __call__(self, *args, **kwargs):
- self.calls.append((args, kwargs))
- if self._raises:
- raise self._raises("boom")
- return self._ret
-
-
-def _mod(name, **attrs):
- module = types.ModuleType(name)
- for key, value in attrs.items():
- setattr(module, key, value)
- sys.modules[name] = module
- return module
-
-
-def _write_analysis(out, *, commit="base123", depth=2):
- path = Path(out)
- path.mkdir(parents=True, exist_ok=True)
- (path / "analysis.json").write_text(
- json.dumps({"metadata": {"commit_hash": commit, "depth_level": depth}}),
- encoding="utf-8",
- )
-
-
-class _Base(unittest.TestCase):
- def tearDown(self):
- for name in _STUBBED:
- sys.modules.pop(name, None)
-
-
-class TestAnalyze(_Base):
- def _install(self, run_full=None, run_incremental=None):
- class BaselineUnavailableError(Exception):
- pass
-
- class IncrementalCacheMissingError(Exception):
- pass
-
- analysis = _mod(
- "codeboarding_workflows.analysis",
- run_full=run_full or _Rec(),
- run_incremental=run_incremental or _Rec(),
- BaselineUnavailableError=BaselineUnavailableError,
- )
- pkg = _mod("codeboarding_workflows")
- pkg.analysis = analysis
- exc = _mod("diagram_analysis.exceptions", IncrementalCacheMissingError=IncrementalCacheMissingError)
- da = _mod("diagram_analysis")
- da.exceptions = exc
- engine_adapter.run_full = analysis.run_full
- engine_adapter.run_incremental = analysis.run_incremental
- engine_adapter.BaselineUnavailableError = BaselineUnavailableError
- engine_adapter.IncrementalCacheMissingError = IncrementalCacheMissingError
- return analysis, IncrementalCacheMissingError, BaselineUnavailableError
-
- def test_no_baseline_runs_full(self):
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = tempfile.mkdtemp()
-
- mode = engine_adapter.run_analyze("/repo", out, "myrepo", "rid", "head123", 2)
-
- self.assertEqual(mode, "full")
- self.assertEqual(len(rf.calls), 1)
- self.assertEqual(len(ri.calls), 0)
- run_paths, run_context = rf.calls[0][0]
- self.assertEqual(run_paths.project_name, "myrepo")
- self.assertEqual(str(run_paths.repo_path), "/repo")
- self.assertEqual(rf.calls[0][1]["depth_level"], 2)
- self.assertEqual(rf.calls[0][1]["source_sha"], "head123")
-
- def test_committed_baseline_runs_incremental(self):
- # Git-free: a committed analysis.json is the baseline; incremental runs
- # (Core diffs the committed fingerprint itself), with no commit_hash gate.
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = tempfile.mkdtemp()
- _write_analysis(out, depth=2)
-
- mode = engine_adapter.run_analyze("/repo", out, "myrepo", "rid", "head123", 2)
-
- self.assertEqual(mode, "incremental")
- self.assertEqual(len(rf.calls), 0)
- self.assertEqual(len(ri.calls), 1)
- run_paths, run_context = ri.calls[0][0]
- self.assertEqual(str(run_paths.repo_path), "/repo")
- self.assertEqual(run_context.run_id, "rid")
-
- def test_incompatible_baseline_runs_full_at_baseline_depth(self):
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = Path(tempfile.mkdtemp())
- _write_analysis(out, depth=3)
- (out / "stale.json").write_text("{}", encoding="utf-8")
- buf = StringIO()
-
- with patch.object(engine_adapter, "UnifiedAnalysisJson", _LossyUnifiedAnalysisJson):
- with redirect_stdout(buf):
- mode = engine_adapter.run_analyze("/repo", str(out), "myrepo", "rid", "head123", 1)
-
- self.assertEqual(mode, "full")
- self.assertEqual(len(ri.calls), 0)
- self.assertEqual(len(rf.calls), 1)
- self.assertEqual(rf.calls[0][1]["depth_level"], 3)
- self.assertFalse((out / "stale.json").exists())
- self.assertIn("could not load baseline analysis.json", buf.getvalue())
- self.assertEqual(self._markers(buf), ["analysis_mode=full"])
-
- def test_deeper_baseline_still_runs_incremental(self):
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = Path(tempfile.mkdtemp())
- _write_analysis(out, commit="metadata-base", depth=3)
- (out / "stale.json").write_text("{}", encoding="utf-8")
- (out / "health").mkdir()
- (out / "health" / "stale.json").write_text("{}", encoding="utf-8")
-
- mode = engine_adapter.run_analyze("/repo", str(out), "myrepo", "rid", "head123", 2)
-
- self.assertEqual(mode, "incremental")
- self.assertEqual(len(rf.calls), 0)
- self.assertEqual(len(ri.calls), 1)
- self.assertTrue((out / "stale.json").exists())
- self.assertTrue((out / "health").exists())
-
- def test_deep_baseline_runs_incremental_regardless_of_tier(self):
- # A committed depth-7 baseline still runs incremental (the depth value
- # doesn't gate incremental — baseline presence does); on the free tier the
- # depth is clamped for any eventual run, but incremental is unaffected.
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = Path(tempfile.mkdtemp())
- _write_analysis(out, commit="metadata-base", depth=7)
-
- mode = engine_adapter.run_analyze("/repo", str(out), "myrepo", "rid", "head123", 2)
-
- self.assertEqual(mode, "incremental")
- self.assertEqual(len(rf.calls), 0)
- self.assertEqual(len(ri.calls), 1)
-
- def test_over_cap_depth_clamped_on_forced_full(self):
- # When a full run happens (here: force_full), the requested depth is
- # clamped to the tier ceiling: free clamps 7 -> 3, licensed keeps 7.
- for licensed, expected in ((False, 3), (True, 7)):
- with self.subTest(licensed=licensed):
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = Path(tempfile.mkdtemp())
- _write_analysis(out, depth=2)
-
- mode = engine_adapter.run_analyze(
- "/repo", str(out), "myrepo", "rid", "head123", 7, force_full=True, licensed=licensed
- )
-
- self.assertEqual(mode, "full")
- self.assertEqual(rf.calls[0][1]["depth_level"], expected)
-
- def test_shallower_baseline_runs_incremental(self):
- # The engine records the depth REACHED, not requested: a depth-2 push on
- # a repo that never expands keeps writing depth_level 1, so a strict !=
- # gate would run full on every push forever.
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = tempfile.mkdtemp()
- _write_analysis(out, commit="metadata-base", depth=1)
-
- mode = engine_adapter.run_analyze("/repo", out, "myrepo", "rid", "head123", 2)
-
- self.assertEqual(mode, "incremental")
- self.assertEqual(len(rf.calls), 0)
- self.assertEqual(len(ri.calls), 1)
-
- def test_missing_depth_still_runs_incremental(self):
- # A missing/unparseable depth_level is not a reason to force a full: the
- # baseline (analysis.json) is present, so incremental runs; the depth
- # resolves to a default and Core falls back to full itself if the cache
- # is actually absent.
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = Path(tempfile.mkdtemp())
- out.joinpath("analysis.json").write_text(json.dumps({"metadata": {}}), encoding="utf-8")
-
- mode = engine_adapter.run_analyze("/repo", str(out), "myrepo", "rid", "head123", 3)
-
- self.assertEqual(mode, "incremental")
- self.assertEqual(len(rf.calls), 0)
- self.assertEqual(len(ri.calls), 1)
-
- def test_baseline_without_commit_still_runs_incremental(self):
- # commit_hash is gone from #401 metadata, so its absence no longer forces
- # a full rebuild — a present analysis.json runs incremental git-free.
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = Path(tempfile.mkdtemp())
- out.joinpath("analysis.json").write_text(json.dumps({"metadata": {"depth_level": 3}}), encoding="utf-8")
-
- mode = engine_adapter.run_analyze("/repo", str(out), "myrepo", "rid", "head123", 1)
-
- self.assertEqual(mode, "incremental")
- self.assertEqual(len(ri.calls), 1)
- self.assertEqual(len(rf.calls), 0)
-
- def test_falls_back_to_full_on_cache_miss(self):
- analysis, IncMiss, _ = self._install()
- rf = _Rec()
- analysis.run_full = rf
- analysis.run_incremental = _Rec(raises=IncMiss)
- engine_adapter.run_full = analysis.run_full
- engine_adapter.run_incremental = analysis.run_incremental
- out = Path(tempfile.mkdtemp())
- _write_analysis(out, commit="metadata-base", depth=3)
- (out / "stale.json").write_text("{}", encoding="utf-8")
-
- mode = engine_adapter.run_analyze("/repo", str(out), "myrepo", "rid", "head123", 1)
-
- self.assertEqual(mode, "full")
- self.assertEqual(len(rf.calls), 1)
- self.assertEqual(rf.calls[0][1]["depth_level"], 3)
- self.assertFalse((out / "stale.json").exists())
-
- def test_falls_back_to_full_on_baseline_unavailable(self):
- analysis, _, BaseUnavailable = self._install()
- rf = _Rec()
- analysis.run_full = rf
- analysis.run_incremental = _Rec(raises=BaseUnavailable)
- engine_adapter.run_full = analysis.run_full
- engine_adapter.run_incremental = analysis.run_incremental
- out = tempfile.mkdtemp()
- _write_analysis(out, commit="metadata-base", depth=2)
-
- mode = engine_adapter.run_analyze("/repo", out, "myrepo", "rid", "head123", 1)
-
- self.assertEqual(mode, "full")
- self.assertEqual(len(rf.calls), 1)
- self.assertEqual(rf.calls[0][1]["depth_level"], 2)
-
- def _markers(self, buf):
- return [line for line in buf.getvalue().splitlines() if line.startswith("analysis_mode=")]
-
- def test_stdout_marker_full_printed_exactly_once(self):
- # The action reads the mode from stdout (tee + sed 's/^analysis_mode=//p');
- # main() discards run_analyze's return value, so the print IS the interface.
- self._install()
- buf = StringIO()
- with redirect_stdout(buf):
- engine_adapter.run_analyze("/repo", tempfile.mkdtemp(), "myrepo", "rid", "head123", 2)
- self.assertEqual(self._markers(buf), ["analysis_mode=full"])
-
- def test_stdout_marker_incremental_printed_exactly_once(self):
- self._install()
- out = tempfile.mkdtemp()
- _write_analysis(out, commit="metadata-base", depth=2)
- buf = StringIO()
- with redirect_stdout(buf):
- engine_adapter.run_analyze("/repo", out, "myrepo", "rid", "head123", 2)
- self.assertEqual(self._markers(buf), ["analysis_mode=incremental"])
-
- def test_stdout_marker_fallback_prints_full_exactly_once(self):
- analysis, IncMiss, _ = self._install()
- analysis.run_full = _Rec()
- analysis.run_incremental = _Rec(raises=IncMiss)
- engine_adapter.run_full = analysis.run_full
- engine_adapter.run_incremental = analysis.run_incremental
- out = tempfile.mkdtemp()
- _write_analysis(out, commit="metadata-base", depth=2)
- buf = StringIO()
- with redirect_stdout(buf):
- engine_adapter.run_analyze("/repo", out, "myrepo", "rid", "head123", 2)
- self.assertEqual(self._markers(buf), ["analysis_mode=full"])
-
- def test_force_full_ignores_valid_baseline(self):
- # force_full must run a full analysis even when a reusable baseline is
- # present (the escape hatch that replaces refresh-baseline.yml).
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = tempfile.mkdtemp()
- _write_analysis(out, commit="metadata-base", depth=2) # a perfectly reusable baseline
- buf = StringIO()
- with redirect_stdout(buf):
- mode = engine_adapter.run_analyze("/repo", out, "myrepo", "rid", "head123", 2, force_full=True)
- self.assertEqual(mode, "full")
- self.assertEqual(len(rf.calls), 1)
- self.assertEqual(len(ri.calls), 0) # baseline never consulted
- self.assertEqual(self._markers(buf), ["analysis_mode=full"])
-
- def test_main_force_full_flag_wires_through(self):
- rf, ri = _Rec(), _Rec()
- self._install(run_full=rf, run_incremental=ri)
- out = tempfile.mkdtemp()
- _write_analysis(out, commit="metadata-base", depth=2)
- with patch.dict(os.environ, {}, clear=True):
- engine_adapter.main(
- [
- "analyze",
- "--repo",
- "/r",
- "--out",
- out,
- "--name",
- "n",
- "--run-id",
- "rid",
- "--source-sha",
- "head123",
- "--depth",
- "2",
- "--force-full",
- ]
- )
- self.assertEqual(len(rf.calls), 1)
- self.assertEqual(len(ri.calls), 0)
-
- def test_main_parses_depth_as_int_and_sets_sync_source(self):
- rf = _Rec()
- self._install(run_full=rf)
- with patch.dict(os.environ, {}, clear=True):
- engine_adapter.main(
- [
- "analyze",
- "--repo",
- "/repo",
- "--out",
- tempfile.mkdtemp(),
- "--name",
- "myrepo",
- "--run-id",
- "rid",
- "--source-sha",
- "head123",
- "--depth",
- "2",
- ]
- )
- self.assertEqual(rf.calls[0][1]["depth_level"], 2)
- self.assertEqual(os.environ["CODEBOARDING_SOURCE"], "sync")
-
- def test_main_rejects_invalid_depth(self):
- # argparse enforces the structural range 1-10; the per-tier cap is applied
- # later by the action/resolver, not here.
- for depth in ("0", "11", "x"):
- with self.subTest(depth=depth):
- with redirect_stderr(StringIO()):
- with self.assertRaises(SystemExit):
- engine_adapter.main(
- [
- "analyze",
- "--repo",
- "/repo",
- "--out",
- "/out",
- "--name",
- "myrepo",
- "--run-id",
- "rid",
- "--source-sha",
- "head123",
- "--depth",
- depth,
- ]
- )
-
-
-class TestRenderAndConcat(_Base):
- def _install_rendering(self, render_docs=None):
- rec = render_docs or _Rec()
- rendering = _mod("codeboarding_workflows.rendering", render_docs=rec)
- pkg = _mod("codeboarding_workflows")
- pkg.rendering = rendering
- engine_adapter.render_docs = rec
- return rec
-
- def test_render_calls_engine_with_overview_root(self):
- rec = self._install_rendering()
-
- engine_adapter.run_render(
- "/tmp/analysis.json", "/tmp/docs", "repo", "https://example/repo/.codeboarding", ".md"
- )
-
- args, kwargs = rec.calls[0]
- self.assertEqual(str(args[0]), "/tmp/analysis.json")
- self.assertEqual(kwargs["repo_name"], "repo")
- self.assertEqual(kwargs["repo_ref"], "https://example/repo/.codeboarding")
- self.assertEqual(str(kwargs["temp_dir"]), "/tmp/docs")
- self.assertEqual(kwargs["format"], ".md")
- self.assertEqual(kwargs["root_name"], "overview")
-
- def test_concat_orders_overview_first_then_sorted_markdown(self):
- docs_dir = Path(tempfile.mkdtemp())
- (docs_dir / "z_component.md").write_text("z", encoding="utf-8")
- (docs_dir / "overview.md").write_text("overview", encoding="utf-8")
- (docs_dir / "a_component.md").write_text("a", encoding="utf-8")
- (docs_dir / "notes.txt").write_text("ignored", encoding="utf-8")
- out = Path(tempfile.mkdtemp()) / "docs" / "development" / "architecture.md"
-
- engine_adapter.run_concat(str(docs_dir), str(out))
-
- self.assertEqual(out.read_text(encoding="utf-8"), "overview\n\na\n\nz\n")
-
-
-class TestSourceDispatch(_Base):
- """CODEBOARDING_SOURCE is setdefault'ed after argparse: sync for
- analyze/render/concat, github_action for everything else (base/seed/head/
- health/validate-base — base is asserted in test_engine_adapter.py)."""
-
- def test_main_render_sets_sync_source(self):
- rec = _Rec()
- rendering = _mod("codeboarding_workflows.rendering", render_docs=rec)
- pkg = _mod("codeboarding_workflows")
- pkg.rendering = rendering
- engine_adapter.render_docs = rec
- with patch.dict(os.environ, {}, clear=True):
- rc = engine_adapter.main(
- [
- "render",
- "--analysis",
- "/tmp/analysis.json",
- "--out",
- tempfile.mkdtemp(),
- "--repo-name",
- "repo",
- "--repo-ref",
- "ref",
- ]
- )
- self.assertEqual(rc, 0)
- self.assertEqual(rec.calls[0][1]["format"], ".md") # default --format
- self.assertEqual(os.environ["CODEBOARDING_SOURCE"], "sync")
-
- def test_main_concat_sets_sync_source(self):
- docs_dir = Path(tempfile.mkdtemp())
- (docs_dir / "overview.md").write_text("overview", encoding="utf-8")
- out = Path(tempfile.mkdtemp()) / "architecture.md"
- with patch.dict(os.environ, {}, clear=True):
- rc = engine_adapter.main(["concat", "--docs-dir", str(docs_dir), "--out", str(out)])
- self.assertEqual(rc, 0)
- self.assertEqual(os.environ["CODEBOARDING_SOURCE"], "sync")
-
- def test_main_validate_base_keeps_github_action_source(self):
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "analysis.json"
- path.write_text(json.dumps({"metadata": {"commit_hash": "abc123"}}), encoding="utf-8")
- with patch.dict(os.environ, {}, clear=True):
- engine_adapter.main(["validate-base", "--analysis", str(path), "--expected-sha", "abc123"])
- self.assertEqual(os.environ["CODEBOARDING_SOURCE"], "github_action")
-
- def test_main_does_not_override_existing_source(self):
- docs_dir = Path(tempfile.mkdtemp())
- (docs_dir / "overview.md").write_text("overview", encoding="utf-8")
- out = Path(tempfile.mkdtemp()) / "architecture.md"
- with patch.dict(os.environ, {"CODEBOARDING_SOURCE": "custom"}, clear=True):
- engine_adapter.main(["concat", "--docs-dir", str(docs_dir), "--out", str(out)])
- self.assertEqual(os.environ["CODEBOARDING_SOURCE"], "custom")
-
-
-class TestBaselineInfo(_Base):
- """baseline-info replaces the sync_seed step's inline heredoc: it returns the
- committed baseline's commit_hash only when present and SHA-shaped."""
-
- def _write(self, metadata):
- out = Path(tempfile.mkdtemp())
- (out / "analysis.json").write_text(json.dumps({"metadata": metadata}), encoding="utf-8")
- return out / "analysis.json"
-
- def test_returns_sha_shaped_commit(self):
- path = self._write({"commit_hash": "a1b2c3d4e5f6"})
- self.assertEqual(engine_adapter.baseline_info(path), "a1b2c3d4e5f6")
-
- def test_rejects_non_sha_commit(self):
- # A non-SHA value must not flow into GITHUB_OUTPUT / cache keys / git.
- for bad in ("not-a-sha", "abc\ncb_dir=/evil", "ABC123", "", "12345"): # too short / wrong charset / injection
- with self.subTest(commit=bad):
- self.assertEqual(engine_adapter.baseline_info(self._write({"commit_hash": bad})), "")
-
- def test_missing_metadata_or_file(self):
- self.assertEqual(engine_adapter.baseline_info(self._write({})), "")
- self.assertEqual(engine_adapter.baseline_info(Path(tempfile.mkdtemp()) / "absent.json"), "")
-
- def test_main_prints_commit_hash_line(self):
- path = self._write({"commit_hash": "deadbeef1234"})
- buf = StringIO()
- with patch.dict(os.environ, {}, clear=True), redirect_stdout(buf):
- rc = engine_adapter.main(["baseline-info", "--analysis", str(path)])
- self.assertEqual(rc, 0)
- self.assertIn("commit_hash=deadbeef1234", buf.getvalue())
-
- def test_main_prints_empty_for_bad_baseline(self):
- path = self._write({"commit_hash": "nope"})
- buf = StringIO()
- with patch.dict(os.environ, {}, clear=True), redirect_stdout(buf):
- engine_adapter.main(["baseline-info", "--analysis", str(path)])
- self.assertIn("commit_hash=", buf.getvalue())
- self.assertNotIn("nope", buf.getvalue())
-
-
-class TestBaselineDepth(_Base):
- """baseline-depth lets review inherit the committed baseline's depth_level
- (clamped to the tier ceiling) so the PR head is analyzed at the same depth as
- the base it is diffed against. It returns a usable number for any present
- baseline, and None only when there is no baseline at all (cold start)."""
-
- def _write(self, metadata):
- out = Path(tempfile.mkdtemp())
- (out / "analysis.json").write_text(json.dumps({"metadata": metadata}), encoding="utf-8")
- return out / "analysis.json"
-
- def test_in_range_passes_through(self):
- for depth in (1, 2, 3): # within the free cap
- with self.subTest(depth=depth):
- self.assertEqual(engine_adapter.baseline_depth(self._write({"depth_level": depth}), False), depth)
-
- def test_clamps_over_cap_per_tier(self):
- # depth 4-10 exceed the free cap (3) -> clamp to 3; licensed cap is 10.
- self.assertEqual(engine_adapter.baseline_depth(self._write({"depth_level": 7}), False), 3)
- self.assertEqual(engine_adapter.baseline_depth(self._write({"depth_level": 7}), True), 7)
- self.assertEqual(engine_adapter.baseline_depth(self._write({"depth_level": 4}), False), 3)
- self.assertEqual(engine_adapter.baseline_depth(self._write({"depth_level": 4}), True), 4)
- self.assertEqual(engine_adapter.baseline_depth(self._write({"depth_level": 99}), True), 10)
-
- def test_invalid_depth_uses_default(self):
- # Every spec violation that isn't an over-cap clamp falls back to the
- # default depth (2): a non-positive depth, an unparseable value, or a
- # missing depth_level are all handled the same way.
- for metadata in (
- {"depth_level": 0},
- {"depth_level": -3},
- {"depth_level": "x"},
- {"commit_hash": "deadbeef1234"},
- ):
- with self.subTest(metadata=metadata):
- self.assertEqual(engine_adapter.baseline_depth(self._write(metadata), False), 2)
-
- def test_none_only_when_no_baseline(self):
- # No file, or an empty/no-metadata object -> cold start (caller defaults).
- self.assertIsNone(engine_adapter.baseline_depth(Path(tempfile.mkdtemp()) / "absent.json", False))
- self.assertIsNone(engine_adapter.baseline_depth(self._write({}), False))
-
- def test_main_prints_depth_line(self):
- path = self._write({"depth_level": 3})
- buf = StringIO()
- with patch.dict(os.environ, {}, clear=True), redirect_stdout(buf):
- rc = engine_adapter.main(["baseline-depth", "--analysis", str(path)])
- self.assertEqual(rc, 0)
- self.assertIn("depth_level=3", buf.getvalue())
-
- def test_main_licensed_raises_ceiling(self):
- path = self._write({"depth_level": 7})
- free, lic = StringIO(), StringIO()
- with patch.dict(os.environ, {}, clear=True), redirect_stdout(free):
- engine_adapter.main(["baseline-depth", "--analysis", str(path)])
- with patch.dict(os.environ, {}, clear=True), redirect_stdout(lic):
- engine_adapter.main(["baseline-depth", "--analysis", str(path), "--licensed"])
- self.assertIn("depth_level=3", free.getvalue()) # clamped
- self.assertIn("depth_level=7", lic.getvalue()) # within licensed cap
-
- def test_main_prints_empty_for_no_baseline(self):
- buf = StringIO()
- with patch.dict(os.environ, {}, clear=True), redirect_stdout(buf):
- engine_adapter.main(["baseline-depth", "--analysis", str(Path(tempfile.mkdtemp()) / "absent.json")])
- self.assertIn("depth_level=", buf.getvalue())
- self.assertNotIn("depth_level=None", buf.getvalue())
-
- def test_diagnostics_go_to_stderr_not_stdout(self):
- # Clamp/default messages must not pollute the machine-readable stdout line.
- path = self._write({"depth_level": 7})
- adapter = Path(__file__).resolve().parent.parent / "scripts" / "engine_adapter.py"
- result = subprocess.run(
- [sys.executable, str(adapter), "baseline-depth", "--analysis", str(path)],
- capture_output=True,
- text=True,
- cwd=tempfile.mkdtemp(),
- )
- self.assertEqual(result.returncode, 0, result.stderr)
- self.assertEqual(result.stdout.strip(), "depth_level=3") # stdout is JUST the value
- self.assertIn("clamping to 3", result.stderr) # the log is on stderr
-
- def test_runs_without_engine_installed(self):
- # The action calls baseline-depth BEFORE the engine package is installed,
- # so it must work as a subprocess with no engine modules on sys.path.
- path = self._write({"depth_level": 3})
- adapter = Path(__file__).resolve().parent.parent / "scripts" / "engine_adapter.py"
- result = subprocess.run(
- [sys.executable, str(adapter), "baseline-depth", "--analysis", str(path)],
- capture_output=True,
- text=True,
- cwd=tempfile.mkdtemp(), # not the repo: no stub engine modules importable
- )
- self.assertEqual(result.returncode, 0, result.stderr)
- self.assertIn("depth_level=3", result.stdout)
-
-
-if __name__ == "__main__":
- unittest.main()