tegwick
2711a3ebcc
Add from_metrics_store factory, OptimizerStore persistence, metrics optimize CLI, consolidate duplicate optimization agent, and add integration tests.
133 lines
4.0 KiB
Python
133 lines
4.0 KiB
Python
"""Tests for OptimizationLoop integration with MetricsStore."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from kaizen_agentic.cli import cli
|
|
from kaizen_agentic.metrics import MetricsStore, OptimizerStore
|
|
from kaizen_agentic.optimization import (
|
|
MIN_SAMPLES_FOR_RECOMMENDATIONS,
|
|
OptimizationLoop,
|
|
)
|
|
|
|
|
|
def _seed_executions(
|
|
store: MetricsStore,
|
|
count: int,
|
|
*,
|
|
success: bool = True,
|
|
execution_time_s: float = 5.0,
|
|
quality_score: float = 0.9,
|
|
) -> None:
|
|
for i in range(count):
|
|
store.append(
|
|
{
|
|
"success": success,
|
|
"execution_time_s": execution_time_s + i,
|
|
"quality_score": quality_score,
|
|
},
|
|
idempotency_key=f"run-{i}",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def project_dir(tmp_path: Path) -> Path:
|
|
root = tmp_path / "demo-project"
|
|
root.mkdir()
|
|
return root
|
|
|
|
|
|
class TestOptimizationFromMetricsStore:
|
|
def test_from_metrics_store_loads_execution_records(self, project_dir: Path):
|
|
store = MetricsStore(project_dir, "tdd-workflow")
|
|
_seed_executions(store, 3)
|
|
|
|
loop = OptimizationLoop.from_metrics_store(store)
|
|
|
|
assert len(loop.metrics_history) == 3
|
|
assert loop.metrics_history[0].success_rate == 1.0
|
|
|
|
def test_insufficient_data_recommendations(self, project_dir: Path):
|
|
store = MetricsStore(project_dir, "tdd-workflow")
|
|
loop = OptimizationLoop.from_metrics_store(store)
|
|
|
|
recommendations = loop.generate_improvement_recommendations()
|
|
|
|
assert recommendations[0]["type"] == "info"
|
|
assert "Insufficient data" in recommendations[0]["message"]
|
|
|
|
def test_sufficient_data_produces_performance_recommendations(
|
|
self, project_dir: Path
|
|
):
|
|
store = MetricsStore(project_dir, "tdd-workflow")
|
|
_seed_executions(
|
|
store,
|
|
MIN_SAMPLES_FOR_RECOMMENDATIONS,
|
|
success=False,
|
|
execution_time_s=60.0,
|
|
quality_score=0.4,
|
|
)
|
|
|
|
loop = OptimizationLoop.from_metrics_store(store)
|
|
recommendations = loop.generate_improvement_recommendations()
|
|
types = {item["type"] for item in recommendations}
|
|
|
|
assert "info" not in types
|
|
assert "reliability" in types or "quality" in types or "performance" in types
|
|
|
|
def test_get_optimization_report_json_is_serializable(self, project_dir: Path):
|
|
import json
|
|
|
|
store = MetricsStore(project_dir, "coach")
|
|
_seed_executions(store, 4)
|
|
|
|
report = OptimizationLoop.from_metrics_store(store).get_optimization_report_json()
|
|
json.dumps(report)
|
|
|
|
|
|
class TestMetricsOptimizeCli:
|
|
def test_optimize_insufficient_samples_writes_analysis_only(
|
|
self, project_dir: Path
|
|
):
|
|
store = MetricsStore(project_dir, "tdd-workflow")
|
|
_seed_executions(store, 2)
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
["metrics", "optimize", "tdd-workflow", "--target", str(project_dir)],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert "need 10" in result.output
|
|
optimizer = OptimizerStore(project_dir)
|
|
assert optimizer.analysis_path.exists()
|
|
assert not optimizer.recommendations_path.exists()
|
|
|
|
def test_optimize_sufficient_samples_writes_recommendations(
|
|
self, project_dir: Path
|
|
):
|
|
store = MetricsStore(project_dir, "tdd-workflow")
|
|
_seed_executions(
|
|
store,
|
|
MIN_SAMPLES_FOR_RECOMMENDATIONS,
|
|
success=False,
|
|
execution_time_s=60.0,
|
|
quality_score=0.4,
|
|
)
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
["metrics", "optimize", "tdd-workflow", "--target", str(project_dir)],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
optimizer = OptimizerStore(project_dir)
|
|
assert optimizer.analysis_path.exists()
|
|
assert optimizer.recommendations_path.exists()
|
|
assert '"type": "reliability"' in result.output or '"type": "quality"' in result.output |