code/serving/test_decisions_api.py

"""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"