Implement MetricsStore for project-scoped agent metrics.
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.
This commit is contained in:
107
tests/test_metrics.py
Normal file
107
tests/test_metrics.py
Normal file
@@ -0,0 +1,107 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user