File size: 7,587 Bytes
26de23c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """No-GPU tests for decisions_api / decisions_server: schema of the OpenRouter Decisions API, error codes, auth, windowing.
cd code && python -m pytest test_decisions_api.py -q
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) # code/: eval, modeling_openjev, openjev_decide
import numpy as np
import pytest
from fastapi.testclient import TestClient
import decisions_api as da
import decisions_server as ds
TUTORIAL = {
"model": "typesafe/jev-1.13",
"state": {"customer_tier": "enterprise",
"ticket": "My checkout page shows a blank screen after I click Pay. I have tried two browsers."},
"questions": {
"is_bug": {"type": "noul", "instructions": "Is the customer reporting a software defect?",
"criteria": {"true": "The customer describes broken or unexpected product behavior.",
"false": "The customer is asking a question or requesting a feature."}},
"team": {"type": "choice", "instructions": "Which team should own this ticket?",
"criteria": {"payments": "Checkout, billing, or payment processing issues.",
"frontend": "Rendering, layout, or browser compatibility issues.",
"account": "Login, permissions, or profile issues."}},
"urgency": {"type": "score", "instructions": "How urgent is this ticket?",
"criteria": ["Can wait for the next release", "Should be fixed this week",
"Blocking revenue right now"]},
},
}
def fake_scores(pairs):
"""Stand-in model: hypotheses naming payments / 'yes' / the last score level are entailed."""
return np.array([0.9 if any(k in h for k in (" is payments:", " is yes:", " is 2:")) else 0.1 for _, h in pairs])
@pytest.fixture
def api(monkeypatch):
async def fake_classify(pairs, image=None):
return fake_scores(pairs), 100 * len(pairs)
monkeypatch.setattr(ds, "classify", fake_classify)
monkeypatch.setattr(ds, "API_KEY", "")
return TestClient(ds.app)
def test_shared_strings_match_openjev_decide():
torch = pytest.importorskip("torch") # openjev_decide imports torch + transformers
import openjev_decide as od
assert od.TEMPLATE == da.HYPOTHESIS and od.WINDOW_CHARS == da.WINDOW_CHARS
def test_pairs_use_the_training_format():
plan = da.build_plan(TUTORIAL)
hyps = [h for _, h in plan.pairs]
assert hyps[0] == ('The answer to "Is the customer reporting a software defect?" is no: '
'The customer is asking a question or requesting a feature.')
assert hyps[1].startswith('The answer to "Is the customer reporting a software defect?" is yes: ')
assert any(h.startswith('The answer to "How urgent is this ticket?" is 2: Blocking') for h in hyps)
assert len(plan.pairs) == 2 + 3 + 3
assert plan.pairs[0][0].startswith('{"customer_tier": "enterprise"') # object state -> json
@pytest.mark.parametrize("path", ["/api/alpha/decisions", "/api/v1/systemone", "/v1/systemone"])
def test_tutorial_response_shape(api, path):
r = api.post(path, json=TUTORIAL)
assert r.status_code == 200, r.text
j = r.json()
assert j["id"].startswith("gen-dec-") and j["model"] and j["provider"]
assert set(j["usage"]) == {"input_tokens", "output_tokens", "cost"} and j["usage"]["output_tokens"] == 0
a = j["answers"]
assert a["is_bug"]["type"] == "noul" and a["is_bug"]["noul"] == pytest.approx(0.9, abs=1e-3)
t = a["team"]
assert t["type"] == "choice" and t["choice"] == "payments" and set(t["probabilities"]) == {"payments", "frontend", "account"}
assert sum(t["probabilities"].values()) == pytest.approx(1, abs=1e-3) and 0 <= t["confidence"] <= 1
u = a["urgency"]
assert u["type"] == "score" and set(u["probabilities"]) == {"0", "1", "2"}
assert u["legend"] == {"0": "Can wait for the next release", "1": "Should be fixed this week",
"2": "Blocking revenue right now"}
assert u["score"] == pytest.approx(0.1 / 1.1 * 1 + 0.9 / 1.1 * 2, abs=1e-3)
def test_confidence_bounds():
assert da.confidence([1.0, 0.0, 0.0]) == pytest.approx(1.0)
assert da.confidence([1 / 3] * 3) == pytest.approx(0.0, abs=1e-9)
assert da.confidence([1.0]) == 1.0
@pytest.mark.parametrize("mut,code", [
(lambda r: r.pop("model"), 400),
(lambda r: r.pop("state"), 400),
(lambda r: r.update(state=5), 400),
(lambda r: r.update(questions={}), 400),
(lambda r: r["questions"]["team"].update(type="rank"), 400),
(lambda r: r["questions"]["is_bug"].update(criteria={"true": "x"}), 400),
(lambda r: r["questions"]["team"].update(criteria=["a"]), 400),
(lambda r: r["questions"]["urgency"].update(criteria={"a": "b"}), 400),
(lambda r: r["questions"]["urgency"].update(criteria=[]), 400),
(lambda r: r.update(state="x" * (da.MAX_STATE_CHARS + 1)), 413),
])
def test_bad_requests(api, mut, code):
import copy
req = copy.deepcopy(TUTORIAL)
mut(req)
r = api.post("/api/alpha/decisions", json=req)
assert r.status_code == code
assert r.json()["error"]["code"] == code and r.json()["error"]["message"]
def test_invalid_json_and_auth(api, monkeypatch):
assert api.post("/api/alpha/decisions", content=b"{nope").status_code == 400
monkeypatch.setattr(ds, "API_KEY", "secret")
assert api.post("/api/alpha/decisions", json=TUTORIAL).status_code == 401
assert api.post("/api/alpha/decisions", json=TUTORIAL, headers={"Authorization": "Bearer wrong"}).status_code == 401
assert api.post("/api/alpha/decisions", json=TUTORIAL, headers={"Authorization": "Bearer secret"}).status_code == 200
def test_long_state_is_windowed_not_cut():
req = {"model": "m", "state": "a" * 60_000, "questions": {"q": {"type": "noul", "instructions": "i",
"criteria": {"true": "t", "false": "f"}}}}
plan = da.build_plan(req)
assert plan.questions[0].n_windows == 3 and len(plan.pairs) == 6
assert plan.pairs[-1][0][-1] == "a" and max(len(p) for p, _ in plan.pairs) == da.WINDOW_CHARS
# only window 0 supports "yes": max over windows per option, then normalise -> yes = 0.9 / (0.9 + 0.1)
ent = np.array([0.1, 0.9, 0.1, 0.1, 0.1, 0.1]) # per window: [no, yes]
assert da.assemble(plan, ent)["q"]["noul"] == pytest.approx(0.9, abs=1e-3)
ent = np.array([0.1, 0.2, 0.9, 0.1, 0.1, 0.1]) # window 1 supports "no" strongly
assert da.assemble(plan, ent)["q"]["noul"] == pytest.approx(0.2 / 1.1, abs=1e-3)
def test_softmax_ent_order():
p = da.softmax_ent([[0, 10, 0], [10, 0, 0]]) # label 1 = entailment
assert p[0] > 0.99 and p[1] < 0.01
def test_health_loading(monkeypatch):
async def boom(*a, **k):
raise __import__("httpx").ConnectError("down")
monkeypatch.setattr(ds.client(), "get", boom)
assert TestClient(ds.app).get("/health").status_code == 503
def test_noul_omitted_criteria_uses_existing_bare_yes_no_rubric(api):
req = {"model": "m", "state": "A duplicate invoice was submitted.",
"questions": {"duplicate": {"type": "noul", "instructions": "This invoice is a duplicate."}}}
plan = da.build_plan(req)
assert [h for _, h in plan.pairs] == [
'The answer to "This invoice is a duplicate." is no: no',
'The answer to "This invoice is a duplicate." is yes: yes',
]
response = api.post("/v1/systemone", json=req)
assert response.status_code == 200
assert response.json()["answers"]["duplicate"]["type"] == "noul"
|