115 lines
4.2 KiB
Python
115 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from tests.pipeline_setup_v4.core.models import V4CaseResult
|
|
|
|
|
|
class ArtifactLayout:
|
|
def __init__(self, *, run_name: str, started_at: datetime) -> None:
|
|
self._run_name = run_name
|
|
self._stamp = started_at.strftime("%Y%m%d_%H%M%S")
|
|
|
|
def run_dir_for(self, source_file: Path) -> Path:
|
|
path = source_file.parent / "test_runs" / self._run_name / self._stamp
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
|
|
class ArtifactWriter:
|
|
def __init__(self, layout: ArtifactLayout) -> None:
|
|
self._layout = layout
|
|
|
|
def write_case(self, result: V4CaseResult) -> Path:
|
|
run_dir = self._layout.run_dir_for(result.case.source_file)
|
|
stem = f"{result.case.source_file.stem}_{result.case.case_id}"
|
|
json_path = run_dir / f"{stem}.json"
|
|
md_path = run_dir / f"{stem}.md"
|
|
json_path.write_text(json.dumps(self._json_payload(result), ensure_ascii=False, indent=2), encoding="utf-8")
|
|
md_path.write_text(self._markdown_payload(result), encoding="utf-8")
|
|
return md_path
|
|
|
|
def write_summaries(self, results: list[V4CaseResult]) -> list[Path]:
|
|
grouped: dict[Path, list[V4CaseResult]] = defaultdict(list)
|
|
for result in results:
|
|
grouped[self._layout.run_dir_for(result.case.source_file)].append(result)
|
|
paths: list[Path] = []
|
|
for run_dir, items in sorted(grouped.items(), key=lambda item: item[0].as_posix()):
|
|
path = run_dir / "summary.md"
|
|
path.write_text(SummaryComposer().compose(items), encoding="utf-8")
|
|
paths.append(path)
|
|
return paths
|
|
|
|
def _json_payload(self, result: V4CaseResult) -> dict:
|
|
return {
|
|
"meta": {
|
|
"case_id": result.case.case_id,
|
|
"component": result.case.component,
|
|
"source_file": result.case.source_file.as_posix(),
|
|
"passed": result.passed,
|
|
"mismatches": result.mismatches,
|
|
},
|
|
"actual": result.actual,
|
|
"details": result.details,
|
|
}
|
|
|
|
def _markdown_payload(self, result: V4CaseResult) -> str:
|
|
lines = [
|
|
f"# {result.case.case_id}",
|
|
"",
|
|
f"- component: {result.case.component}",
|
|
f"- source_file: {result.case.source_file.as_posix()}",
|
|
f"- passed: {result.passed}",
|
|
"",
|
|
"## Query",
|
|
result.case.query,
|
|
"",
|
|
"## Actual",
|
|
"```json",
|
|
json.dumps(result.actual, ensure_ascii=False, indent=2),
|
|
"```",
|
|
"",
|
|
"## Details",
|
|
"```json",
|
|
json.dumps(result.details, ensure_ascii=False, indent=2),
|
|
"```",
|
|
"",
|
|
"## Mismatches",
|
|
*([f"- {item}" for item in result.mismatches] or ["- none"]),
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
class SummaryComposer:
|
|
def compose(self, results: list[V4CaseResult]) -> str:
|
|
passed = sum(1 for item in results if item.passed)
|
|
lines = [
|
|
"# pipeline_setup_v4 summary",
|
|
"",
|
|
f"Passed: {passed}/{len(results)}",
|
|
"",
|
|
"| Case | Component | Query | Intent | Sub-intent | Pass |",
|
|
"|------|-----------|-------|--------|------------|------|",
|
|
]
|
|
for item in results:
|
|
lines.append(
|
|
f"| {item.case.case_id} | {item.case.component} | {self._cell(item.case.query)} | "
|
|
f"{item.actual.get('intent') or '—'} | {item.actual.get('sub_intent') or '—'} | "
|
|
f"{'✓' if item.passed else '✗'} |"
|
|
)
|
|
failures = [item for item in results if not item.passed]
|
|
if failures:
|
|
lines.extend(["", "## Failures"])
|
|
for item in failures:
|
|
lines.append(f"- **{item.case.case_id}**: {'; '.join(item.mismatches)}")
|
|
return "\n".join(lines)
|
|
|
|
def _cell(self, text: str, limit: int = 140) -> str:
|
|
compact = " ".join(text.split()).replace("|", "\\|")
|
|
if len(compact) <= limit:
|
|
return compact
|
|
return compact[: limit - 1].rstrip() + "…"
|