70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
"""Spike A tests — verdict logic (varied inputs, non-tautological) + gated live arm.
|
|
|
|
The verdict function is exercised with *different* Arm inputs so a pass is genuine,
|
|
not a single scripted constant. The live arm is isolated behind ``skipif`` (no endpoint).
|
|
"""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
from agent_framework import Message
|
|
|
|
from spikes.a_groupchat import Arm, make_termination, run_live, verdict
|
|
|
|
_NO_ENDPOINT = not (
|
|
os.environ.get("PORTFOLIO_LOCAL_BASE_URL") and os.environ.get("PORTFOLIO_LOCAL_MODEL")
|
|
)
|
|
|
|
|
|
def _arm(label: str, *, tokens: int, caught: bool) -> Arm:
|
|
return Arm(label=label, rounds=3, stalls=0, tokens=tokens, caught_flaw=caught)
|
|
|
|
|
|
def test_verdict_passes_when_better_and_affordable() -> None:
|
|
# mc caught the flaw, single missed it, and mc spent 2x (<= 3x) the tokens.
|
|
mc = _arm("maker-checker", tokens=200, caught=True)
|
|
single = _arm("single-agent", tokens=100, caught=False)
|
|
v = verdict(mc, single)
|
|
assert v["better"] is True
|
|
assert v["affordable"] is True
|
|
assert v["passed"] is True
|
|
|
|
|
|
def test_verdict_fails_when_unaffordable_even_if_better() -> None:
|
|
# mc is better but spent 5x the tokens -> not affordable -> not passed.
|
|
mc = _arm("maker-checker", tokens=500, caught=True)
|
|
single = _arm("single-agent", tokens=100, caught=False)
|
|
v = verdict(mc, single)
|
|
assert v["better"] is True
|
|
assert v["affordable"] is False
|
|
assert v["passed"] is False
|
|
|
|
|
|
def test_verdict_fails_when_both_caught_so_not_better() -> None:
|
|
# Both arms caught the flaw -> maker-checker is not *better* -> not passed,
|
|
# even though it is affordable.
|
|
mc = _arm("maker-checker", tokens=150, caught=True)
|
|
single = _arm("single-agent", tokens=100, caught=True)
|
|
v = verdict(mc, single)
|
|
assert v["better"] is False
|
|
assert v["passed"] is False
|
|
|
|
|
|
def test_make_termination_stops_at_n_rounds() -> None:
|
|
term = make_termination(3)
|
|
msgs = [Message(role="assistant", contents=["x"]) for _ in range(3)]
|
|
assert term(msgs[:2]) is False
|
|
assert term(msgs[:3]) is True
|
|
|
|
|
|
def test_make_termination_rejects_non_positive() -> None:
|
|
with pytest.raises(ValueError):
|
|
make_termination(0)
|
|
|
|
|
|
@pytest.mark.skipif(_NO_ENDPOINT, reason="LOCAL endpoint not configured (PORTFOLIO_LOCAL_*)")
|
|
async def test_run_live_converges_within_cap_and_emits_table() -> None:
|
|
result = await run_live(write_findings=False)
|
|
mc = result["mc"]
|
|
assert mc.rounds <= 3 # cap respected
|
|
assert "verdict passed" in result["table"] # comparison table emitted
|