Private
Public Access
test(audit): Phase 10 - 7 integration tests against synthetic src/
Updated synthetic ai_client.py + aggregate.py to use proper return annotations (Metadata, FileItems, History) so P1 detects the producers. 7 integration tests: 1. synthetic src/ produces 10 real + 3 candidate profiles 2. Metadata has >=1 producer (after fixing fixture annotations) 3. Metadata memory_dim is 'discussion' (canonical) 4. FileItems memory_dim is 'curation' (canonical) 5. History memory_dim is 'discussion' (canonical) 6. Missing audit_inputs tolerated 7. render_rollups produces 4 non-empty rollup files 131 tests total passing.
This commit is contained in:
+5
-5
@@ -1,15 +1,15 @@
|
|||||||
"""Synthetic aggregate module for the v2 audit integration tests."""
|
"""Synthetic aggregate module for the v2 audit integration tests."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
from typing import Any
|
from typing import Any, List
|
||||||
|
|
||||||
FileItems = list[dict[str, Any]]
|
FileItems = List[dict[str, Any]]
|
||||||
|
|
||||||
def build_file_items() -> list[dict[str, Any]]:
|
def build_file_items() -> FileItems:
|
||||||
data: FileItems = [{"path": "a.py", "view_mode": "full"}, {"path": "b.py", "view_mode": "summary"}]
|
data: FileItems = [{"path": "a.py", "view_mode": "full"}, {"path": "b.py", "view_mode": "summary"}]
|
||||||
return data
|
return data
|
||||||
|
|
||||||
def format_paths(items: FileItems) -> list[str]:
|
def format_paths(items: FileItems) -> List[str]:
|
||||||
out: list[str] = []
|
out: List[str] = []
|
||||||
for item in items:
|
for item in items:
|
||||||
path = item["path"]
|
path = item["path"]
|
||||||
view_mode = item["view_mode"]
|
view_mode = item["view_mode"]
|
||||||
|
|||||||
+4
-4
@@ -1,15 +1,15 @@
|
|||||||
"""Synthetic AI client for the v2 audit integration tests."""
|
"""Synthetic AI client for the v2 audit integration tests."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
from typing import Any
|
from typing import Any, List
|
||||||
|
|
||||||
Metadata = dict[str, Any]
|
Metadata = dict[str, Any]
|
||||||
History = list[dict[str, Any]]
|
History = List[dict[str, Any]]
|
||||||
|
|
||||||
def send_result() -> dict[str, Any]:
|
def send_result() -> Metadata:
|
||||||
data: Metadata = {"role": "user", "content": "hello"}
|
data: Metadata = {"role": "user", "content": "hello"}
|
||||||
return data
|
return data
|
||||||
|
|
||||||
def append_history() -> list[dict[str, Any]]:
|
def append_history() -> History:
|
||||||
data: History = []
|
data: History = []
|
||||||
return data
|
return data
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,109 @@
|
|||||||
|
"""Integration tests for src.code_path_audit v2."""
|
||||||
|
from __future__ import annotations
|
||||||
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
|
from src.code_path_audit import (
|
||||||
|
run_audit,
|
||||||
|
render_rollups,
|
||||||
|
)
|
||||||
|
|
||||||
|
FIXTURE_SRC = "tests/fixtures/synthetic_src"
|
||||||
|
FIXTURE_INPUTS = "tests/fixtures/audit_inputs"
|
||||||
|
|
||||||
|
def test_integration_synthetic_src_produces_3_real_aggregates() -> None:
|
||||||
|
"""The full run_audit against the synthetic src/ produces 10 real + 3 candidate profiles."""
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
result = run_audit(
|
||||||
|
src_dir=FIXTURE_SRC,
|
||||||
|
audit_inputs_dir=FIXTURE_INPUTS,
|
||||||
|
output_dir=tmp,
|
||||||
|
date="2026-06-22",
|
||||||
|
)
|
||||||
|
assert result.ok
|
||||||
|
real_profiles = [p for p in result.data.aggregate_profiles if not p.is_candidate]
|
||||||
|
candidate_profiles = [p for p in result.data.aggregate_profiles if p.is_candidate]
|
||||||
|
assert len(real_profiles) == 10
|
||||||
|
assert len(candidate_profiles) == 3
|
||||||
|
assert any(p.name == "Metadata" for p in real_profiles)
|
||||||
|
assert any(p.name == "FileItems" for p in real_profiles)
|
||||||
|
assert any(p.name == "History" for p in real_profiles)
|
||||||
|
|
||||||
|
def test_integration_metadata_profile_has_producers_and_consumers() -> None:
|
||||||
|
"""The Metadata profile has at least 1 producer and 1 consumer from the synthetic src/."""
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
result = run_audit(
|
||||||
|
src_dir=FIXTURE_SRC,
|
||||||
|
audit_inputs_dir=FIXTURE_INPUTS,
|
||||||
|
output_dir=tmp,
|
||||||
|
date="2026-06-22",
|
||||||
|
)
|
||||||
|
assert result.ok
|
||||||
|
metadata = next(p for p in result.data.aggregate_profiles if p.name == "Metadata")
|
||||||
|
assert len(metadata.producers) >= 1
|
||||||
|
|
||||||
|
def test_integration_metadata_memory_dim_is_discussion() -> None:
|
||||||
|
"""The Metadata profile's memory_dim is 'discussion' (canonical mapping)."""
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
result = run_audit(
|
||||||
|
src_dir=FIXTURE_SRC,
|
||||||
|
audit_inputs_dir=FIXTURE_INPUTS,
|
||||||
|
output_dir=tmp,
|
||||||
|
date="2026-06-22",
|
||||||
|
)
|
||||||
|
assert result.ok
|
||||||
|
metadata = next(p for p in result.data.aggregate_profiles if p.name == "Metadata")
|
||||||
|
assert metadata.memory_dim == "discussion"
|
||||||
|
|
||||||
|
def test_integration_file_items_memory_dim_is_curation() -> None:
|
||||||
|
"""The FileItems profile's memory_dim is 'curation' (canonical mapping)."""
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
result = run_audit(
|
||||||
|
src_dir=FIXTURE_SRC,
|
||||||
|
audit_inputs_dir=FIXTURE_INPUTS,
|
||||||
|
output_dir=tmp,
|
||||||
|
date="2026-06-22",
|
||||||
|
)
|
||||||
|
assert result.ok
|
||||||
|
file_items = next(p for p in result.data.aggregate_profiles if p.name == "FileItems")
|
||||||
|
assert file_items.memory_dim == "curation"
|
||||||
|
|
||||||
|
def test_integration_history_memory_dim_is_discussion() -> None:
|
||||||
|
"""The History profile's memory_dim is 'discussion' (canonical mapping)."""
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
result = run_audit(
|
||||||
|
src_dir=FIXTURE_SRC,
|
||||||
|
audit_inputs_dir=FIXTURE_INPUTS,
|
||||||
|
output_dir=tmp,
|
||||||
|
date="2026-06-22",
|
||||||
|
)
|
||||||
|
assert result.ok
|
||||||
|
history = next(p for p in result.data.aggregate_profiles if p.name == "History")
|
||||||
|
assert history.memory_dim == "discussion"
|
||||||
|
|
||||||
|
def test_integration_missing_audit_inputs_tolerated() -> None:
|
||||||
|
"""run_audit tolerates missing audit inputs."""
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
inputs_dir = str(Path(tmp) / "nonexistent")
|
||||||
|
result = run_audit(
|
||||||
|
src_dir=FIXTURE_SRC,
|
||||||
|
audit_inputs_dir=inputs_dir,
|
||||||
|
output_dir=tmp,
|
||||||
|
date="2026-06-22",
|
||||||
|
)
|
||||||
|
assert result.ok
|
||||||
|
assert len(result.data.aggregate_profiles) == 13
|
||||||
|
|
||||||
|
def test_integration_produces_4_rollup_files() -> None:
|
||||||
|
"""run_audit + render_rollups produce the 4 top-level rollup files."""
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
result = run_audit(
|
||||||
|
src_dir=FIXTURE_SRC,
|
||||||
|
audit_inputs_dir=FIXTURE_INPUTS,
|
||||||
|
output_dir=tmp,
|
||||||
|
date="2026-06-22",
|
||||||
|
)
|
||||||
|
assert result.ok
|
||||||
|
rollup_paths = render_rollups(result.data, Path(tmp) / "2026-06-22")
|
||||||
|
for name, path in rollup_paths.items():
|
||||||
|
assert Path(path).exists()
|
||||||
|
assert Path(path).stat().st_size > 0
|
||||||
Reference in New Issue
Block a user