Wire OptimizationLoop to project metrics and add metrics optimize.
Add from_metrics_store factory, OptimizerStore persistence, metrics optimize CLI, consolidate duplicate optimization agent, and add integration tests.
This commit is contained in:
133
tests/test_optimization_metrics.py
Normal file
133
tests/test_optimization_metrics.py
Normal file
@@ -0,0 +1,133 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user