tegwick
5cd3da3166
Add ADR-004 storage layer with append-only executions, summary regeneration, idempotency keys, and retention pruning. Wire memory init to scaffold .kaizen/metrics/ by default and add unit tests.
107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
"""Tests for project-scoped metrics storage (ADR-004)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from kaizen_agentic.metrics import MetricsStore, DEFAULT_RETENTION_DAYS
|
|
|
|
|
|
def _old_timestamp(days: int) -> str:
|
|
dt = datetime.now(timezone.utc) - timedelta(days=days)
|
|
return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
@pytest.fixture
|
|
def project_dir(tmp_path: Path) -> Path:
|
|
root = tmp_path / "demo-project"
|
|
root.mkdir()
|
|
return root
|
|
|
|
|
|
class TestMetricsStore:
|
|
def test_scaffold_creates_directory_and_empty_executions(self, project_dir: Path):
|
|
store = MetricsStore(project_dir, "tdd-workflow")
|
|
path = store.scaffold()
|
|
|
|
assert path == project_dir / ".kaizen" / "metrics" / "tdd-workflow"
|
|
assert store.executions_path.exists()
|
|
assert store.executions_path.read_text() == ""
|
|
|
|
def test_append_and_read_executions(self, project_dir: Path):
|
|
store = MetricsStore(project_dir, "tdd-workflow")
|
|
|
|
assert store.append({"success": True, "quality_score": 0.9}) is True
|
|
assert store.append({"success": False, "execution_time_s": 12.5}) is True
|
|
|
|
records = store.read_executions()
|
|
assert len(records) == 2
|
|
assert records[0]["agent"] == "tdd-workflow"
|
|
assert records[0]["success"] is True
|
|
assert "timestamp" in records[0]
|
|
|
|
def test_idempotency_key_rejects_duplicate(self, project_dir: Path):
|
|
store = MetricsStore(project_dir, "coach")
|
|
|
|
assert store.append({"success": True}, idempotency_key="sess-1") is True
|
|
assert store.append({"success": True}, idempotency_key="sess-1") is False
|
|
assert len(store.read_executions()) == 1
|
|
|
|
def test_write_summary_regenerates_summary_json(self, project_dir: Path):
|
|
store = MetricsStore(project_dir, "tdd-workflow")
|
|
store.append({"success": True, "quality_score": 0.8, "execution_time_s": 10})
|
|
store.append({"success": True, "quality_score": 1.0, "execution_time_s": 20})
|
|
|
|
summary = store.write_summary()
|
|
|
|
assert summary["execution_count"] == 2
|
|
assert summary["success_rate"] == 1.0
|
|
assert summary["avg_quality_score"] == 0.9
|
|
assert summary["avg_execution_time_s"] == 15.0
|
|
assert store.summary_path.exists()
|
|
on_disk = json.loads(store.summary_path.read_text())
|
|
assert on_disk["execution_count"] == 2
|
|
|
|
def test_prune_removes_expired_records(self, project_dir: Path):
|
|
store = MetricsStore(project_dir, "tdd-workflow", retention_days=30)
|
|
store.scaffold()
|
|
|
|
old = {
|
|
"timestamp": _old_timestamp(45),
|
|
"agent": "tdd-workflow",
|
|
"success": False,
|
|
}
|
|
recent = {
|
|
"timestamp": _old_timestamp(1),
|
|
"agent": "tdd-workflow",
|
|
"success": True,
|
|
"quality_score": 0.7,
|
|
}
|
|
with store.executions_path.open("w", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(old) + "\n")
|
|
handle.write(json.dumps(recent) + "\n")
|
|
|
|
removed = store.prune()
|
|
|
|
assert removed == 1
|
|
records = store.read_executions()
|
|
assert len(records) == 1
|
|
assert records[0]["success"] is True
|
|
summary = store.read_summary()
|
|
assert summary is not None
|
|
assert summary["execution_count"] == 1
|
|
|
|
def test_list_agents_with_metrics(self, project_dir: Path):
|
|
MetricsStore(project_dir, "tdd-workflow").scaffold()
|
|
MetricsStore(project_dir, "coach").append({"success": True})
|
|
|
|
agents = MetricsStore.list_agents(project_dir)
|
|
|
|
assert agents == ["coach", "tdd-workflow"]
|
|
|
|
def test_default_retention_matches_adr(self):
|
|
assert DEFAULT_RETENTION_DAYS == 180 |