107 lines
4.7 KiB
Python
107 lines
4.7 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def _mk_factory_module(monkeypatch, tmp_path):
|
|
from products import product_factory
|
|
|
|
monkeypatch.setattr(product_factory, "ROOT", tmp_path)
|
|
monkeypatch.setattr(product_factory, "EVENT_DIR", tmp_path / ".events")
|
|
monkeypatch.setattr(product_factory, "PRODUCTS_DIR", tmp_path / "generated")
|
|
monkeypatch.setattr(product_factory, "DASHBOARD", tmp_path / "DASHBOARD.md")
|
|
monkeypatch.setattr(product_factory, "PLANS_DIR", tmp_path / "plans")
|
|
monkeypatch.setattr(product_factory, "RECEIPTS_DIR", tmp_path / "receipts")
|
|
monkeypatch.setattr(product_factory, "run_adw_pipeline", lambda product_id, zte=False: {"ok": True})
|
|
return product_factory
|
|
|
|
|
|
def test_product_factory_creates_spec(tmp_path, monkeypatch):
|
|
product_factory = _mk_factory_module(monkeypatch, tmp_path)
|
|
|
|
result = product_factory.create_product("Agent powered support triage")
|
|
|
|
assert result["product_id"] == "agent-powered-support-triage"
|
|
assert (tmp_path / "generated" / result["product_id"] / "spec.json").exists()
|
|
assert Path(result["plan"]).exists()
|
|
assert Path(result["receipt"]).exists()
|
|
|
|
spec = json.loads((tmp_path / "generated" / result["product_id"] / "spec.json").read_text())
|
|
assert spec["living_plan"]["deploy_endpoint"] == "git-proxy:8099/deploy"
|
|
assert spec["living_plan"]["rsi_decision"] == "healthy_but_autopatch_unproven"
|
|
assert spec["living_plan"]["auto_patch_proven"] is False
|
|
assert spec["living_plan"]["rsi_canary_recovery_evidence"] is True
|
|
assert spec["living_plan"]["rsi_evidence_receipt"].endswith("rsi-proof-20260619T093724Z.md")
|
|
assert spec["living_plan"]["local_rsi_evidence_receipt"].endswith("rsi-proof-20260619T093724Z.md")
|
|
assert set(spec["living_plan"]["loop_engineering"]) == {
|
|
"agent_loop",
|
|
"verification_loop",
|
|
"event_driven_loop",
|
|
"hill_climbing_loop",
|
|
}
|
|
|
|
receipt = json.loads(Path(result["receipt"]).read_text())
|
|
assert receipt["deploy_route"] == "git-proxy:8099/deploy"
|
|
assert receipt["rsi_decision"] == "healthy_but_autopatch_unproven"
|
|
assert receipt["auto_patch_proven"] is False
|
|
assert receipt["rsi_canary_recovery_evidence"] is True
|
|
assert receipt["rsi_evidence_receipt"].endswith("rsi-proof-20260619T093724Z.md")
|
|
assert receipt["local_rsi_evidence_receipt"].endswith("rsi-proof-20260619T093724Z.md")
|
|
assert receipt["loop_engineering"]["hill_climbing_loop"].startswith("skill_health")
|
|
|
|
log = next((tmp_path / ".events").glob("*.jsonl"))
|
|
events = [json.loads(line) for line in log.read_text().splitlines()]
|
|
assert [e["kind"] for e in events] == [
|
|
"idea_received",
|
|
"product_spec_created",
|
|
"plan_written",
|
|
"adw_pipeline_started",
|
|
"checks_run",
|
|
"verifier_result",
|
|
"deploy_action_requested",
|
|
"outcome",
|
|
]
|
|
checks = next(e for e in events if e["kind"] == "checks_run")
|
|
assert checks["checks"] == ["plan", "build", "test", "review", "document", "ship"]
|
|
assert checks["checks_ok"] is True
|
|
assert checks["plan_id"] == "agent-powered-support-triage"
|
|
assert checks["plan_ref"].endswith("plan.md")
|
|
|
|
plan_event = next(e for e in events if e["kind"] == "plan_written")
|
|
assert plan_event["plan_id"] == "agent-powered-support-triage"
|
|
assert plan_event["plan_ref"].endswith("plan.md")
|
|
|
|
dashboard = (tmp_path / "DASHBOARD.md").read_text()
|
|
assert "## Last check" in dashboard
|
|
assert "## Deploy readiness" in dashboard
|
|
assert "git-proxy:8099/deploy" in dashboard
|
|
assert "auto_patch_proven=false" in dashboard
|
|
assert "rsi-proof-20260619T093724Z.md" in dashboard
|
|
assert "## Loop engineering" in dashboard
|
|
assert "Loop 4 hill-climbing" in dashboard
|
|
assert "## Open risks" in dashboard
|
|
|
|
|
|
def test_product_factory_blocks_overwrite_when_exists(tmp_path, monkeypatch):
|
|
product_factory = _mk_factory_module(monkeypatch, tmp_path)
|
|
|
|
first = product_factory.create_product("Duplication safety check")
|
|
|
|
existing_spec = Path(first["path"]) / "spec.json"
|
|
stamp = existing_spec.stat().st_mtime
|
|
|
|
assert first["product_id"] == "duplication-safety-check"
|
|
assert existing_spec.exists()
|
|
|
|
import pytest
|
|
|
|
with pytest.raises(FileExistsError):
|
|
product_factory.create_product("Duplication safety check")
|
|
|
|
# same product still re-runs with explicit overwrite
|
|
second = product_factory.create_product("Duplication safety check", overwrite=True)
|
|
assert second["product_id"] == first["product_id"]
|
|
assert second["plan"] != ""
|
|
assert Path(second["plan"]).exists()
|
|
assert Path(second["plan"]).resolve() == (tmp_path / "plans" / "duplication-safety-check" / "plan.md").resolve()
|
|
assert existing_spec.stat().st_mtime >= stamp
|