122 lines
5.2 KiB
Python
122 lines
5.2 KiB
Python
"""Quick smoke test for all pipeline modules."""
|
|
import json
|
|
import sys
|
|
import os
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
# Add pipeline dir to path
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from pipeline.confidence_ladder import Verdict, VerdictLevel, VerdictReport
|
|
from pipeline.rejection_envelope import RejectionEnvelope, Blocker
|
|
from pipeline.state_gate import StateGate, FieldSpec, PipelineState, GateResult
|
|
|
|
errors = 0
|
|
|
|
# ── Test 1: VerdictLevel ordering ──
|
|
assert VerdictLevel.PERFECT > VerdictLevel.VERIFIED, "PERFECT should be > VERIFIED"
|
|
assert VerdictLevel.PARTIAL > VerdictLevel.FEEDBACK, "PARTIAL should be > FEEDBACK"
|
|
assert VerdictLevel.FAILED < VerdictLevel.FEEDBACK, "FAILED should be < FEEDBACK"
|
|
assert VerdictLevel.PARTIAL.is_pass, "PARTIAL should be pass"
|
|
assert VerdictLevel.FEEDBACK.is_pass == False, "FEEDBACK should NOT be pass"
|
|
assert VerdictLevel.VERIFIED.is_blocking, "VERIFIED should be blocking"
|
|
assert VerdictLevel.PARTIAL.is_blocking == False, "PARTIAL should NOT be blocking"
|
|
print("✓ VerdictLevel ordering and predicates")
|
|
|
|
# ── Test 2: VerdictReport aggregation ──
|
|
report = VerdictReport(pipeline_id="test-001")
|
|
report.add(Verdict(VerdictLevel.PERFECT, "style check passed"))
|
|
report.add(Verdict(VerdictLevel.PARTIAL, "unit tests: can't test DB layer offline", ["no DB in CI"]))
|
|
report.add(Verdict(VerdictLevel.VERIFIED, "integration tests passed"))
|
|
assert report.aggregate == VerdictLevel.PARTIAL, f"Aggregate should be PARTIAL (lowest), got {report.aggregate}"
|
|
assert report.passed == True, "PARTIAL should be a pass"
|
|
print("✓ VerdictReport aggregation (PARTIAL is min, but still pass)")
|
|
|
|
# ── Test 3: VerdictReport JSON round-trip ──
|
|
serialized = report.to_json()
|
|
restored = VerdictReport.from_json(serialized)
|
|
assert restored.aggregate == report.aggregate, "JSON round-trip should preserve aggregate"
|
|
assert len(restored.verdicts) == len(report.verdicts), "JSON round-trip should preserve verdict count"
|
|
print("✓ JSON serialization round-trip")
|
|
|
|
# ── Test 4: RejectionEnvelope ──
|
|
reject = RejectionEnvelope.block(
|
|
received="write_file('foo.py')",
|
|
reason="LEADER_MUST_NOT_WRITE_FILES",
|
|
allowed="Delegate write to a writer subagent",
|
|
hint="Leaders read, delegate, synthesize. Never write files directly.",
|
|
)
|
|
assert "BLOCKED" in str(reject)
|
|
assert reject.received == "write_file('foo.py')"
|
|
reject_json = reject.to_json()
|
|
reject2 = RejectionEnvelope.from_json(reject_json)
|
|
assert reject2.reason == "LEADER_MUST_NOT_WRITE_FILES"
|
|
print("✓ RejectionEnvelope with 4-field schema")
|
|
|
|
# ── Test 5: Blocker rule ──
|
|
no_write = Blocker(
|
|
name="no_direct_write",
|
|
match=lambda a: "write_file" in str(a.get("tool", "")),
|
|
reject=lambda a: RejectionEnvelope.block(
|
|
received=str(a),
|
|
reason="LEADER_MUST_NOT_WRITE_FILES",
|
|
),
|
|
)
|
|
result = no_write.evaluate({"tool": "write_file", "path": "test.py"})
|
|
assert result is not None, "write_file should be blocked"
|
|
assert result.reason == "LEADER_MUST_NOT_WRITE_FILES"
|
|
result = no_write.evaluate({"tool": "read_file", "path": "test.py"})
|
|
assert result is None, "read_file should NOT be blocked"
|
|
print("✓ Blocker rule evaluation")
|
|
|
|
# ── Test 6: PipelineState with core_fields ──
|
|
state = PipelineState(
|
|
{"plan_file": "plan.md", "branch_name": "feat/x", "rogue_key": "should_be_dropped"},
|
|
core_fields=["plan_file", "branch_name", "backend_port"],
|
|
)
|
|
assert state.get("plan_file") == "plan.md"
|
|
assert state.get("branch_name") == "feat/x"
|
|
assert state.get("rogue_key") is None, "rogue_key should be silently dropped"
|
|
state.update(backend_port=9100, another_rogue="dropped")
|
|
assert state.get("backend_port") == 9100
|
|
assert state.get("another_rogue") is None
|
|
print("✓ PipelineState with core_fields filter")
|
|
|
|
# ── Test 7: StateGate validation ──
|
|
gate = StateGate(
|
|
FieldSpec("plan_file", required=True),
|
|
FieldSpec("branch_name", required=True),
|
|
FieldSpec("backend_port", required=True, type_hint=int),
|
|
)
|
|
incomplete = PipelineState({"plan_file": "plan.md"}, core_fields=["plan_file", "branch_name", "backend_port"])
|
|
result = gate.validate(incomplete)
|
|
assert not result.passed, "Missing fields should fail"
|
|
assert "branch_name" in result.missing
|
|
assert "backend_port" in result.missing
|
|
print("✓ StateGate catches missing fields")
|
|
|
|
complete = PipelineState(
|
|
{"plan_file": "plan.md", "branch_name": "feat/x", "backend_port": 9100},
|
|
core_fields=["plan_file", "branch_name", "backend_port"],
|
|
)
|
|
result = gate.validate(complete)
|
|
assert result.passed, "Complete state should pass"
|
|
print("✓ StateGate passes complete state")
|
|
|
|
# ── Test 8: Type validation ──
|
|
wrong_type = PipelineState(
|
|
{"plan_file": "plan.md", "branch_name": "feat/x", "backend_port": "not-an-int"},
|
|
core_fields=["plan_file", "branch_name", "backend_port"],
|
|
)
|
|
result = gate.validate(wrong_type)
|
|
assert not result.passed, "Type mismatch should fail"
|
|
assert any("backend_port" in e[0] for e in result.type_errors), "Should report type error for backend_port"
|
|
print("✓ StateGate type validation")
|
|
|
|
print(f"\n✅ All {8 - errors} tests passed!" if errors == 0 else f"\n❌ {errors} test(s) failed")
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(0 if errors == 0 else 1)
|