Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 32 additions & 1 deletion scripts/formatters/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,33 @@ def _normalize_to_ai(data: Any, command: str = "") -> Dict[str, Any]:
"suggestion": data.get("suggestion", ""),
})

# Umbrella envelope {s, st, r:[...]} (the #195 consolidation): normalize
# each sub-result and merge, so `ai` consumers get the data instead of an
# empty items list. Stats are namespaced per sub-check to avoid collisions
# when several checks run at once (issue #306).
sub_results = data.get("r")
if isinstance(sub_results, list) and "st" in data:
merged_items = []
merged_stats = {}
for sub in sub_results:
if not isinstance(sub, dict):
continue
check = sub.get("_check", "")
norm = _normalize_to_ai(sub, check or command)
merged_items.extend(norm.get("items", []))
sub_stats = norm.get("stats", {})
if sub_stats:
merged_stats[check or "result"] = sub_stats
return stamp_schema_version({
"status": "ok" if data.get("s", "ok") != "error" else "error",
"command": command,
"stats": merged_stats,
"items": merged_items,
"truncated": False,
"recommendations": [],
"metadata": {"checks": data.get("st", {})},
})

result = {
"status": status,
"command": command,
Expand Down Expand Up @@ -79,6 +106,10 @@ def _normalize_to_ai(data: Any, command: str = "") -> Dict[str, Any]:
elif "identity" in data and "registry_stats" in data:
# summary
stats.update(data["registry_stats"])
elif isinstance(data.get("summary"), dict):
# Generic: many sub-checks (tags, diff) carry a flat `summary` dict.
# Last resort so it never shadows a more specific stats source above.
stats.update(data["summary"])

result["stats"] = stats

Expand All @@ -90,7 +121,7 @@ def _normalize_to_ai(data: Any, command: str = "") -> Dict[str, Any]:
_ITEM_KEYS = [
"functions", "findings", "leaks", "hints", "issues",
"matches", "violations", "entrypoints", "routes", "stores",
"results", "ownership_summary", "chains",
"results", "ownership_summary", "chains", "flows",
"by_category", "top_priority", "actionable_items",
]

Expand Down
61 changes: 61 additions & 0 deletions scripts/formatters/markdown.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,21 @@
lines = []
status = data.get("status", "")

# Umbrella envelope {s, st, r:[...]} (the #195 command consolidation):
# unwrap and render each sub-result through its own `_check` handler.
# Without this, every umbrella sub-check rendered empty or "Symbol not
# found" because the old flat handlers never saw the shape they expect
# (issue #306). Recursion reuses every existing per-command renderer.
sub_results = data.get("r")
if isinstance(sub_results, list) and "st" in data:
parts = []
for sub in sub_results:
if isinstance(sub, dict):
rendered = to_markdown(sub, sub.get("_check", "")).strip()
if rendered:
parts.append(rendered)
return "\n\n".join(parts) if parts else "_No results._"

# Error output
if status == "error":
lines.append(f"## Error")
Expand Down Expand Up @@ -124,13 +139,59 @@
_md_summary(data, lines)
elif command == "analyze":
_md_analyze(data, lines)
elif command == "tags":
_md_tags(data, lines)
else:
# Generic markdown for any command
_md_generic(data, lines)

return "\n".join(lines)


def _md_tags(data: Dict, lines: list) -> None:

Check failure on line 151 in scripts/formatters/markdown.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 19 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=Wolfvin_CodeLens&issues=AZ9wvid-bmHhZdCEt-It&open=AZ9wvid-bmHhZdCEt-It&pullRequest=308
"""Markdown for the doc-tag audit (`context --check tags`, issue #305)."""
s = data.get("summary", {})
lines.append("## Doc-Tag Audit")
lines.append("")
lines.append(
f"**Header coverage:** {s.get('header_coverage_pct', 0)}% "
f"({s.get('with_full_header', 0)} full, "
f"{s.get('with_partial_header', 0)} partial, "
f"{s.get('without_header', 0)} untagged of "
f"{s.get('files_scanned', 0)} files)"
)
lines.append(f"**Named flows:** {s.get('distinct_flows', 0)}")
lines.append("")

flows = data.get("flows", [])
if flows:
lines.append("### Flows")
for f in flows:
locs = f.get("locations", [])
where = locs[0] if locs else ""
extra = f" (+{len(locs) - 1} more)" if len(locs) > 1 else ""
lines.append(f"- `{f.get('name', '')}` — {where}{extra}")
lines.append("")

partial = data.get("partial_headers", [])
if partial:
lines.append("### Partial headers (missing tags)")
for p in partial:
lines.append(f"- `{p.get('file', '')}` — missing {', '.join(p.get('missing', []))}")
if data.get("partial_headers_truncated"):
lines.append("- …")
lines.append("")

untagged = data.get("untagged_files", [])
if untagged:
lines.append(f"### Untagged files ({s.get('without_header', 0)})")
for u in untagged:
lines.append(f"- `{u}`")
if data.get("untagged_files_truncated"):
lines.append("- …")
lines.append("")


def _md_generic(data: Dict, lines: list) -> None:
"""Generic markdown output for any command."""
lines.append(f"## Result")
Expand Down
100 changes: 100 additions & 0 deletions tests/test_umbrella_formats.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
"""
Tests for umbrella envelope rendering in markdown + ai formats (issue #306).

The #195 command consolidation wraps sub-check output in an envelope
``{s, st, r:[...]}``. The markdown and ai formatters never learned that shape,
so every umbrella sub-check rendered empty ("Symbol not found") or with an
empty ``items`` list. These tests pin the unwrap.
"""

import os
import sys
import unittest

SCRIPT_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts")
sys.path.insert(0, SCRIPT_DIR)

from formatters.markdown import to_markdown # noqa: E402
from formatters import _normalize_to_ai # noqa: E402


def _envelope(sub):
"""Wrap a sub-result the way an umbrella command does."""
return {"s": "ok", "st": {"checks_requested": 1, "checks_run": 1, "checks_failed": 0}, "r": [sub]}


_TAGS_SUB = {
"status": "ok",
"_check": "tags",
"summary": {
"files_scanned": 10, "with_full_header": 4, "with_partial_header": 1,
"without_header": 5, "header_coverage_pct": 50.0, "distinct_flows": 2,
"total_flow_declarations": 2,
},
"flows": [
{"name": "PAYMENT", "count": 2, "locations": ["a.py:1", "b.py:9"]},
{"name": "AUTH", "count": 1, "locations": ["c.py:3"]},
],
"partial_headers": [{"file": "d.py", "present": ["WHO"], "missing": ["WHAT", "PART", "ENTRY"]}],
"partial_headers_truncated": False,
"untagged_files": ["e.py", "f.py"],
"untagged_files_truncated": False,
}


class TestMarkdownEnvelope(unittest.TestCase):
def test_tags_envelope_renders_content_not_symbol_not_found(self):
out = to_markdown(_envelope(_TAGS_SUB), "context")

self.assertNotIn("Symbol not found", out)
self.assertIn("Doc-Tag Audit", out)
self.assertIn("PAYMENT", out)
self.assertIn("a.py:1", out)

def test_tags_envelope_lists_partial_and_untagged(self):
out = to_markdown(_envelope(_TAGS_SUB), "context")

self.assertIn("d.py", out) # partial header
self.assertIn("WHAT", out) # its missing tag
self.assertIn("e.py", out) # untagged file

def test_envelope_dispatches_sub_to_its_own_handler(self):
"""A dead-code sub-result must reach the dead-code renderer, not generic."""
sub = {"status": "ok", "_check": "dead-code", "dead_functions": [], "summary": {}}
out = to_markdown(_envelope(sub), "audit")

self.assertIn("Dead Code Analysis", out)

def test_empty_envelope_is_not_a_crash(self):
out = to_markdown({"s": "ok", "st": {}, "r": []}, "context")

self.assertIn("No results", out)

def test_non_umbrella_output_not_regressed(self):
"""A flat (non-envelope) result must still use its own handler."""
out = to_markdown({"status": "ok", "dead_functions": [], "summary": {}}, "dead-code")

self.assertIn("Dead Code Analysis", out)


class TestAiEnvelope(unittest.TestCase):
def test_tags_envelope_populates_items(self):
out = _normalize_to_ai(_envelope(_TAGS_SUB), "context")

self.assertEqual(len(out["items"]), 2) # the two flows
self.assertEqual({i["name"] for i in out["items"]}, {"PAYMENT", "AUTH"})

def test_tags_envelope_populates_stats(self):
out = _normalize_to_ai(_envelope(_TAGS_SUB), "context")

self.assertIn("tags", out["stats"])
self.assertEqual(out["stats"]["tags"]["distinct_flows"], 2)

def test_envelope_records_check_metadata(self):
out = _normalize_to_ai(_envelope(_TAGS_SUB), "context")

self.assertIn("checks", out["metadata"])


if __name__ == "__main__":
unittest.main()
Loading