ed
c82538474f
Phase 8: to_dsl_v2 (flat-section writer, 14 sections), to_markdown (10 sections), to_tree (box-drawing prefix tree), parse_dsl_v2 (round-trip parser). Phase 9: AGGREGATES_IN_SCOPE (10) + CANDIDATE_AGGREGATES (3), synthesize_aggregate_profile (per-aggregate builder, candidate placeholder path), AuditSummary dataclass, run_audit() main entry, render_rollups() (4 top-level files: summary, cross_audit_summary, decomposition_matrix, candidates), code_path_audit_v2() MCP tool wrapper. 13 new unit tests passing. 124 total tests passing. Phase 10 (integration tests with synthetic src/) next - may be deferred to next session if context runs low.
179 lines
6.2 KiB
Python
179 lines
6.2 KiB
Python
"""Tests for src.code_path_audit v2 - DSL renderers + run_audit + CLI + MCP."""
|
|
from __future__ import annotations
|
|
import ast
|
|
import tempfile
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
from datetime import date
|
|
from src.code_path_audit import (
|
|
AggregateKind,
|
|
MemoryDim,
|
|
AccessPattern,
|
|
Frequency,
|
|
RecommendedDirection,
|
|
FunctionRef,
|
|
AccessPatternEvidence,
|
|
FrequencyEvidence,
|
|
ResultCoverage,
|
|
TypeAliasCoverage,
|
|
CrossAuditFindings,
|
|
DecompositionCost,
|
|
OptimizationCandidate,
|
|
AggregateProfile,
|
|
to_dsl_v2,
|
|
to_markdown,
|
|
to_tree,
|
|
parse_dsl_v2,
|
|
AGGREGATES_IN_SCOPE,
|
|
CANDIDATE_AGGREGATES,
|
|
synthesize_aggregate_profile,
|
|
run_audit,
|
|
render_rollups,
|
|
AuditSummary,
|
|
code_path_audit_v2,
|
|
)
|
|
from src.result_types import Result
|
|
|
|
def _make_profile(name: str = "Metadata", kind: str = "typealias") -> AggregateProfile:
|
|
f = FunctionRef(fqname="src.x.y", file="src/x.py", line=1, role="producer")
|
|
return AggregateProfile(
|
|
name=name,
|
|
aggregate_kind=kind,
|
|
memory_dim="discussion",
|
|
producers=(f,),
|
|
consumers=(),
|
|
access_pattern="whole_struct",
|
|
access_pattern_evidence=(),
|
|
frequency="per_turn",
|
|
frequency_evidence=(),
|
|
result_coverage=ResultCoverage(0, 0, 0, 0, ""),
|
|
type_alias_coverage=TypeAliasCoverage(0, 0, 0, ""),
|
|
cross_audit_findings=CrossAuditFindings((), (), (), (), ()),
|
|
decomposition_cost=DecompositionCost(0, 0, 0, "hold", "no data", None, 0, False),
|
|
optimization_candidates=(),
|
|
is_candidate=False,
|
|
)
|
|
|
|
# Phase 8 Tasks 8.2-8.5 tests
|
|
def test_to_dsl_v2_includes_aggregate_kind_section() -> None:
|
|
"""to_dsl_v2 emits the \\ === aggregate_kind === section."""
|
|
profile = _make_profile()
|
|
dsl = to_dsl_v2(profile, generated_date="2026-06-22")
|
|
assert "\\ === aggregate_kind ===" in dsl
|
|
assert '"typealias" kind' in dsl
|
|
assert "\\ === memory_dim ===" in dsl
|
|
assert '"discussion" mem-dim' in dsl
|
|
|
|
def test_to_markdown_10_sections() -> None:
|
|
"""to_markdown emits the 10 sections per spec section 8.1."""
|
|
profile = _make_profile()
|
|
md = to_markdown(profile)
|
|
assert "# Aggregate Profile: Metadata" in md
|
|
assert "## Pipeline summary" in md
|
|
assert "## Access pattern" in md
|
|
assert "## Frequency" in md
|
|
assert "## Result coverage" in md
|
|
assert "## Type alias coverage" in md
|
|
assert "## Cross-audit findings" in md
|
|
assert "## Decomposition cost" in md
|
|
assert "## Optimization candidates" in md
|
|
assert "## Verdict" in md
|
|
|
|
def test_to_tree_box_drawing() -> None:
|
|
"""to_tree uses box-drawing characters."""
|
|
profile = _make_profile()
|
|
tree = to_tree(profile)
|
|
assert "Metadata" in tree
|
|
assert "kind: typealias" in tree
|
|
|
|
def test_parse_dsl_v2_round_trip_aggregate_kind() -> None:
|
|
"""parse_dsl_v2(to_dsl_v2(profile)) recovers the aggregate_kind section."""
|
|
profile = _make_profile()
|
|
dsl = to_dsl_v2(profile)
|
|
parsed = parse_dsl_v2(dsl)
|
|
assert isinstance(parsed, Result)
|
|
assert "typealias" in str(parsed.data)
|
|
|
|
def test_parse_dsl_v2_malformed() -> None:
|
|
"""parse_dsl_v2 returns Result.ok for empty input (no tagged words)."""
|
|
result = parse_dsl_v2("\\ empty comment only\n")
|
|
assert result.ok
|
|
assert result.data == {"_sections": []}
|
|
|
|
# Phase 9 Tasks 9.1-9.6 tests
|
|
def test_aggregates_in_scope_10_real() -> None:
|
|
"""AGGREGATES_IN_SCOPE has 10 real aggregates."""
|
|
expected = {"Metadata", "FileItem", "FileItems", "CommsLogEntry", "CommsLog", "HistoryMessage", "History", "ToolDefinition", "ToolCall", "Result"}
|
|
assert set(AGGREGATES_IN_SCOPE) == expected
|
|
|
|
def test_candidate_aggregates_3_placeholders() -> None:
|
|
"""CANDIDATE_AGGREGATES has the 3 candidate aggregates."""
|
|
expected = {"ToolSpec", "ChatMessage", "ProviderHistory"}
|
|
assert set(CANDIDATE_AGGREGATES) == expected
|
|
|
|
def test_synthesize_real_aggregate() -> None:
|
|
"""synthesize_aggregate_profile returns a real AggregateProfile for a known aggregate."""
|
|
f = FunctionRef(fqname="src.x.y", file="src/x.py", line=1, role="producer")
|
|
profile = synthesize_aggregate_profile(
|
|
aggregate="Metadata",
|
|
pcg_producers={"Metadata": [f]},
|
|
pcg_consumers={"Metadata": [f]},
|
|
audit_inputs={},
|
|
overrides={},
|
|
is_candidate=False,
|
|
)
|
|
assert profile.name == "Metadata"
|
|
assert profile.aggregate_kind == "typealias"
|
|
assert profile.memory_dim == "discussion"
|
|
assert profile.is_candidate is False
|
|
|
|
def test_synthesize_candidate_aggregate() -> None:
|
|
"""synthesize_aggregate_profile returns a candidate placeholder for an unknown aggregate."""
|
|
profile = synthesize_aggregate_profile(
|
|
aggregate="ChatMessage",
|
|
pcg_producers={"ChatMessage": []},
|
|
pcg_consumers={"ChatMessage": []},
|
|
audit_inputs={},
|
|
overrides={},
|
|
is_candidate=True,
|
|
)
|
|
assert profile.name == "ChatMessage"
|
|
assert profile.aggregate_kind == "candidate_dataclass"
|
|
assert profile.is_candidate is True
|
|
assert profile.producers == ()
|
|
assert profile.consumers == ()
|
|
|
|
def test_run_audit_returns_result() -> None:
|
|
"""run_audit returns Result[AuditSummary] per error_handling.md."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
result = run_audit(src_dir=tmp, audit_inputs_dir=tmp, output_dir=tmp, date="2026-06-22")
|
|
assert isinstance(result, Result)
|
|
assert result.ok
|
|
|
|
def test_run_audit_produces_13_aggregates() -> None:
|
|
"""run_audit produces 13 AggregateProfiles (10 in-scope + 3 candidate)."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
result = run_audit(src_dir=tmp, audit_inputs_dir=tmp, output_dir=tmp, date="2026-06-22")
|
|
assert result.ok
|
|
assert len(result.data.aggregate_profiles) == 13
|
|
|
|
def test_render_rollups_produces_4_files() -> None:
|
|
"""render_rollups produces summary.md, cross_audit_summary.md, decomposition_matrix.md, candidates.md."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
audit_result = run_audit(src_dir=tmp, audit_inputs_dir=tmp, output_dir=tmp, date="2026-06-22")
|
|
assert audit_result.ok
|
|
rollup_paths = render_rollups(audit_result.data, Path(tmp) / "2026-06-22")
|
|
assert "summary.md" in rollup_paths
|
|
assert "cross_audit_summary.md" in rollup_paths
|
|
assert "decomposition_matrix.md" in rollup_paths
|
|
assert "candidates.md" in rollup_paths
|
|
|
|
def test_code_path_audit_v2_returns_dict() -> None:
|
|
"""code_path_audit_v2 returns a dict with 'profiles' + 'errors' keys (MCP tool contract)."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
result = code_path_audit_v2(src_dir=tmp, audit_inputs_dir=tmp, output_dir=tmp, date="2026-06-22")
|
|
assert isinstance(result, dict)
|
|
assert "profiles" in result
|
|
assert "errors" in result
|
|
assert len(result["profiles"]) == 13 |