manual_slop/tests/test_code_path_audit_phase89.py

"""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_markdown,
 to_tree,
 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_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

# 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_2_files() -> None:
 """render_rollups produces summary.md + AUDIT_REPORT.md (MVP: single comprehensive document)."""
 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 "AUDIT_REPORT.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