| """Contract tests for the HTTP API (deg.api / deg.server / deg.client). |
| |
| Golden fixtures: |
| * ``fixtures/openai_decisions_docs.json``: the OpenAI Decisions guide's requests and |
| illustrative responses (openai-python 3.26.1 types); |
| * ``fixtures/jev_systemone_live.json``: live api.typesafe.ai (jev-1.13.0) requests, |
| statuses and responses. |
| |
| The model is :class:`fake_fill.FakeFill` (no torch); responses still go through |
| ``deg.response.assemble`` and its schema validation. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import threading |
| import time |
| from pathlib import Path |
| from typing import Any |
|
|
| import pytest |
| from fake_fill import FakeFill |
|
|
| from deg.api import parse_decisions, parse_systemone |
| from deg.client import Deg, DegAPIError |
| from deg.response import MODEL_ID, confidence |
| from deg.server import Batcher, Engine, make_server |
|
|
| FIXTURES = Path(__file__).parent / "fixtures" |
| DOCS = json.loads((FIXTURES / "openai_decisions_docs.json").read_text())["cases"] |
| JEV = json.loads((FIXTURES / "jev_systemone_live.json").read_text())["cases"] |
|
|
| EVENT_INPUT = ("Hi, order AB-1234 arrived with a cracked screen on 2026-10-01. " |
| "I want a refund, not a replacement.") |
| EVENT_QUESTIONS = [ |
| {"type": "choice", "name": "event", "instructions": "Which event should be dispatched?", |
| "choices": [{"value": "refund", "description": "Customer asks for money back."}, |
| {"value": "replacement", "description": "Customer wants a new unit."}, |
| {"value": "other"}]}, |
| {"type": "predicate", "name": "damaged", "instructions": "cracked or broken item?"}, |
| {"type": "extract", "name": "order_id", "instructions": "The order id."}, |
| {"type": "extract", "name": "delivered", "instructions": "Delivery date.", |
| "value_type": "date"}, |
| {"type": "extract", "name": "tracking", "instructions": "Tracking number.", |
| "pattern": "^1Z\\w+$"}, |
| {"type": "generate", "name": "note", "instructions": "A one-line note for the agent.", |
| "max_length": 3, "max_tokens": 64}, |
| ] |
|
|
|
|
| class Server: |
| def __init__(self, *, api_key: str | None = None, fill: Any = None, |
| fill_many: Any = None, window_s: float = 0.001, max_pending: int = 256): |
| self.fill = fill or FakeFill() |
| self.batcher = Batcher(self.fill, fill_many=fill_many, window_s=window_s, |
| max_pending=max_pending) |
| self.httpd = make_server(Engine(self.batcher), "127.0.0.1", 0, api_key=api_key) |
| self.url = f"http://127.0.0.1:{self.httpd.server_address[1]}" |
| threading.Thread(target=self.httpd.serve_forever, daemon=True).start() |
|
|
| def close(self) -> None: |
| self.httpd.shutdown() |
| self.httpd.server_close() |
|
|
|
|
| @pytest.fixture(scope="module") |
| def server(): |
| srv = Server() |
| yield srv |
| srv.close() |
|
|
|
|
| @pytest.fixture(scope="module") |
| def client(server): |
| return Deg(server.url) |
|
|
|
|
| def shape(value: Any) -> Any: |
| """Keys (in order) and JSON types, recursively; lists by their first item.""" |
| if isinstance(value, dict): |
| return {k: shape(v) for k, v in value.items()} |
| if isinstance(value, list): |
| return [shape(value[0])] if value else [] |
| if isinstance(value, bool): |
| return "bool" |
| if isinstance(value, (int, float)): |
| return "number" |
| return type(value).__name__ |
|
|
|
|
| def raw_post(url: str, path: str, body: Any, headers: dict | None = None) -> tuple[int, Any]: |
| import urllib.error |
| import urllib.request |
|
|
| data = body if isinstance(body, bytes) else json.dumps(body).encode() |
| req = urllib.request.Request(url + path, data=data, method="POST", |
| headers={"Content-Type": "application/json", **(headers or {})}) |
| try: |
| with urllib.request.urlopen(req, timeout=30) as r: |
| return r.status, json.loads(r.read()) |
| except urllib.error.HTTPError as exc: |
| return exc.code, json.loads(exc.read()) |
|
|
|
|
| |
|
|
|
|
| @pytest.mark.parametrize("case", ["choice", "score"]) |
| def test_decisions_docs_golden_shape(client, case): |
| golden = DOCS[case] |
| got = client.decisions.create(**golden["request"]) |
| excerpt = golden["response_excerpt"] |
| assert shape(got["answers"]) == shape(excerpt["answers"]) |
| assert list(got) == ["answers", "model", "usage"] |
| assert got.model == MODEL_ID |
| usage = got.usage |
| assert list(usage) == ["input_tokens", "input_tokens_details", "output_tokens", |
| "output_tokens_details", "total_tokens"] |
| assert usage.total_tokens == usage.input_tokens + usage.output_tokens > 0 |
| assert usage.output_tokens == 0 |
| ans = got.answers[0] |
| probs = [p.probability for p in ans.probabilities] |
| assert abs(sum(probs) - 1) < 1e-9 |
| assert ans.confidence == pytest.approx(confidence(probs)) |
| if case == "choice": |
| assert [p.value for p in ans.probabilities] == [ |
| c["value"] for c in golden["request"]["questions"][0]["choices"]] |
| assert ans.choice == max(ans.probabilities, key=lambda p: p.probability).value |
| else: |
| assert [(p.value, p.label) for p in ans.probabilities] == [ |
| (0, "Cosmetic"), (1, "Workaround available"), (2, "Fully blocked")] |
| assert ans.score == pytest.approx(sum(p.value * p.probability |
| for p in ans.probabilities)) |
|
|
|
|
| def test_decisions_docs_score_formula(): |
| """The guide's own numbers: probabilities .1/.7/.2 over indices 0/1/2 -> score 1.1.""" |
| from deg.api import _expected |
|
|
| assert _expected([0.1, 0.7, 0.2]) == pytest.approx(1.1) |
|
|
|
|
| def test_decisions_predicate_text_and_messages(client): |
| req = json.loads(json.dumps(DOCS["predicate_image"]["request"])) |
| req["input"][0]["content"] = [{"type": "input_text", "text": "Does the box show damage?"}] |
| got = client.decisions.create(**req) |
| assert shape(got.answers) == shape(DOCS["predicate_image"]["response_excerpt"]["answers"]) |
| assert 0 <= got.answers[0].probability <= 1 |
|
|
|
|
| def test_decisions_image_refused(server): |
| status, body = raw_post(server.url, "/v1/decisions", DOCS["predicate_image"]["request"]) |
| assert status == 400 |
| assert body["error"]["type"] == "invalid_request_error" |
| assert "input_image is not supported" in body["error"]["message"] |
|
|
|
|
| def test_decisions_extensions_event_dispatch(server, client): |
| got = client.decisions.create(model="deg-latest", input=EVENT_INPUT, |
| questions=EVENT_QUESTIONS) |
| a = {x.name: x for x in got.answers} |
| assert [x.name for x in got.answers] == [q["name"] for q in EVENT_QUESTIONS] |
| assert a["event"].type == "choice" and a["event"].choice == "refund" |
| assert a["damaged"].type == "predicate" |
| assert a["order_id"] == {"name": "order_id", "type": "extract", "value": "AB-1234", |
| "span": [10, 17], "confidence": 0.9} |
| assert EVENT_INPUT[10:17] == "AB-1234" |
| assert a["delivered"].value == "2026-10-01" |
| assert a["tracking"].value is None and a["tracking"].null_reason == "not_found" |
| assert a["tracking"].span is None |
| assert a["note"] == {"name": "note", "type": "generate", "text": "not", |
| "output_tokens": 2} |
| assert got.usage.output_tokens == 2 |
| |
| _, schema, options = server.fill.calls[-1] |
| assert list(schema["properties"]) == [q["name"] for q in EVENT_QUESTIONS] |
| assert schema["properties"]["note"] == {"type": "string", "x-generate": True, |
| "description": "A one-line note for the agent.", |
| "maxLength": 3, "x-max-tokens": 64} |
| assert schema["properties"]["delivered"]["format"] == "date" |
| assert schema["properties"]["order_id"]["type"] == ["string", "null"] |
| assert options.max_new_tokens == 256 |
|
|
|
|
| def test_decisions_score_levels_go_to_x_levels(server, client): |
| got = client.decisions.create(model="deg-latest", input="I am furious", questions=[ |
| {"type": "score", "name": "anger", "instructions": "How angry is the customer?", |
| "levels": [{"label": "calm", "description": "no complaint"}, {"label": "annoyed"}, |
| {"label": "furious", "description": "threatens to leave"}]}]) |
| _, schema, _ = server.fill.calls[-1] |
| assert schema["properties"]["anger"] == { |
| "type": "integer", "minimum": 0, "maximum": 2, |
| "description": "How angry is the customer?", |
| "x-levels": {"0": "calm: no complaint", "1": "annoyed", |
| "2": "furious: threatens to leave"}} |
| assert [p.label for p in got.answers[0].probabilities] == ["calm", "annoyed", "furious"] |
|
|
|
|
| def test_generate_questions_mix_max_tokens(server, client): |
| got = client.decisions.create(model="deg-latest", input="refund please", questions=[ |
| {"type": "generate", "name": "a", "instructions": "x", "max_tokens": 8}, |
| {"type": "generate", "name": "b", "instructions": "y", "max_tokens": 9}, |
| {"type": "generate", "name": "c", "instructions": "z"}]) |
| assert [a.type for a in got.answers] == ["generate"] * 3 |
| _, schema, options = server.fill.calls[-1] |
| props = schema["properties"] |
| assert props["a"]["x-max-tokens"] == 8 and props["b"]["x-max-tokens"] == 9 |
| assert "x-max-tokens" not in props["c"] and options.max_new_tokens == 256 |
| status, _ = raw_post(server.url, "/v1/systemone", { |
| "model": "jev-latest", "state": "refund please", "questions": { |
| "a": {"type": "generate", "instructions": "x", "max_tokens": 4}, |
| "b": {"type": "generate", "instructions": "y", "max_tokens": 12}}}) |
| assert status == 200 |
| _, schema, _ = server.fill.calls[-1] |
| assert [schema["properties"][k]["x-max-tokens"] for k in "ab"] == [4, 12] |
|
|
|
|
| def test_decisions_unnamed_questions_get_null_name(client): |
| got = client.decisions.create(model="deg-latest", input="refund please", questions=[ |
| {"type": "predicate", "instructions": "refund requested?"}, |
| {"type": "predicate", "instructions": "angry?"}]) |
| assert [a.name for a in got.answers] == [None, None] |
|
|
|
|
| def test_decisions_single_choice_and_level_without_model(server, client): |
| before = len(server.fill.calls) |
| got = client.decisions.create(model="deg-latest", input="x", questions=[ |
| {"type": "choice", "name": "c", "instructions": "pick", "choices": [{"value": True}]}, |
| {"type": "score", "name": "s", "instructions": "rate", "levels": [{"label": "only"}]}]) |
| assert len(server.fill.calls) == before |
| assert got.answers[0].choice is True and got.answers[0].confidence == 1.0 |
| assert got.answers[1].score == 0 and got.usage.input_tokens == 0 |
|
|
|
|
| @pytest.mark.parametrize("questions,needle", [ |
| ([{"type": "classify", "instructions": "x"}], "type must be one of"), |
| ([{"type": "choice", "instructions": "x", |
| "choices": [{"value": f"o{i}"} for i in range(256)]}], "Too many choices"), |
| ([{"type": "score", "instructions": "x", |
| "levels": [{"label": str(i)} for i in range(11)]}], "Too many score levels"), |
| ([{"type": "score", "instructions": "x", "levels": []}], "levels must be a non-empty"), |
| ([{"type": "choice", "instructions": "x", "choices": []}], "choices must be a non-empty"), |
| ([{"type": "choice", "instructions": "x", |
| "choices": [{"value": "a"}, {"value": "a"}]}], "duplicate choice value"), |
| ([{"type": "predicate", "name": "a", "instructions": "x"}, |
| {"type": "predicate", "name": "a", "instructions": "y"}], "names must be unique"), |
| ([{"type": "predicate", "instructions": "x", "choices": []}], "Unrecognized question"), |
| ([{"type": "generate", "instructions": "x", "max_tokens": 0}], "max_tokens must be"), |
| ([{"type": "generate", "instructions": "x", "pattern": "("}], "pattern"), |
| ([{"type": "extract", "instructions": "x", "value_type": "date", |
| "max_length": 3}], "applies only to string"), |
| ([{"type": "extract", "instructions": "x", "value_type": "phone"}], "value_type"), |
| ([], "non-empty array"), |
| ]) |
| def test_decisions_validation_errors(server, questions, needle): |
| status, body = raw_post(server.url, "/v1/decisions", |
| {"model": "gpt-6-luna", "input": "x", "questions": questions}) |
| assert status == 400, body |
| assert set(body["error"]) == {"message", "type", "param", "code"} |
| assert needle in body["error"]["message"] |
|
|
|
|
| def test_decisions_choice_values_are_typed(client): |
| got = client.decisions.create(model="deg-latest", input="true", questions=[ |
| {"type": "choice", "name": "c", "instructions": "pick", |
| "choices": [{"value": "true"}, {"value": True}]}]) |
| assert [p.value for p in got.answers[0].probabilities] == ["true", True] |
|
|
|
|
| def test_decisions_unknown_model_and_fields(server): |
| status, body = raw_post(server.url, "/v1/decisions", { |
| "model": "gpt-5", "input": "x", "questions": [{"type": "predicate", |
| "instructions": "y"}], "foo": 1}) |
| assert status == 400 |
| msg = body["error"]["message"] |
| assert "Unrecognized request argument supplied: foo" in msg |
| assert "The model `gpt-5` does not exist" in msg |
|
|
|
|
| def test_decisions_request_options(server, client): |
| with pytest.raises(DegAPIError) as exc: |
| client.decisions.create(model="deg-latest", input="x", options={"temperature": 0.5}, |
| questions=[{"type": "generate", "instructions": "y"}]) |
| assert exc.value.status == 400 and "requires a seed" in exc.value.body["error"]["message"] |
| client.decisions.create(model="deg-latest", input="x", |
| options={"temperature": 0.5, "seed": 3}, |
| questions=[{"type": "generate", "instructions": "y"}]) |
| assert server.fill.calls[-1][2].seed == 3 |
|
|
|
|
| |
|
|
|
|
| def test_stock_openai_sdk_round_trip(server): |
| openai = pytest.importorskip("openai") |
| from openai.types.decision import ( |
| AnswerAnswerResourceChoice, |
| AnswerAnswerResourcePredicate, |
| AnswerAnswerResourceScore, |
| Decision, |
| ) |
|
|
| oai = openai.OpenAI(base_url=server.url + "/v1", api_key="unused", max_retries=0) |
| for case in ("choice", "score"): |
| got = oai.decisions.create(**DOCS[case]["request"]) |
| assert isinstance(got, Decision) |
| want = {"choice": AnswerAnswerResourceChoice, "score": AnswerAnswerResourceScore}[case] |
| assert isinstance(got.answers[0], want) |
| assert got.answers[0].name == DOCS[case]["request"]["questions"][0]["name"] |
| assert got.usage.total_tokens > 0 |
| got = oai.decisions.create(model="gpt-6-luna", input=[ |
| {"role": "user", "content": [{"type": "input_text", "text": "cracked lid"}]}], |
| questions=[{"type": "predicate", "name": "visible_damage", |
| "instructions": "cracked anything?"}]) |
| assert isinstance(got.answers[0], AnswerAnswerResourcePredicate) |
| assert got.answers[0].probability == pytest.approx(0.8) |
|
|
|
|
| def test_stock_openai_sdk_extensions_and_errors(server): |
| openai = pytest.importorskip("openai") |
| oai = openai.OpenAI(base_url=server.url + "/v1", api_key="unused", max_retries=0) |
| got = oai.decisions.create(model="deg-latest", input=EVENT_INPUT, |
| questions=EVENT_QUESTIONS, |
| extra_body={"options": {"date_order": "MDY"}}) |
| by_name = {a.name: a for a in got.answers} |
| assert by_name["event"].choice == "refund" |
| |
| assert by_name["order_id"].type == "extract" |
| assert by_name["order_id"].value == "AB-1234" |
| assert by_name["note"].text == "not" |
| assert server.fill.calls[-1][2].date_order == "MDY" |
| with pytest.raises(openai.BadRequestError) as exc: |
| oai.decisions.create(model="gpt-6-luna", input="x", |
| questions=[{"type": "score", "instructions": "y", |
| "levels": [{"label": str(i)} for i in range(11)]}]) |
| assert "Too many score levels" in str(exc.value) |
|
|
|
|
| def test_stock_openai_sdk_auth(): |
| openai = pytest.importorskip("openai") |
| srv = Server(api_key="k1") |
| try: |
| ok = openai.OpenAI(base_url=srv.url + "/v1", api_key="k1", max_retries=0) |
| assert ok.decisions.create(**DOCS["choice"]["request"]).answers[0].type == "choice" |
| bad = openai.OpenAI(base_url=srv.url + "/v1", api_key="nope", max_retries=0) |
| with pytest.raises(openai.AuthenticationError): |
| bad.decisions.create(**DOCS["choice"]["request"]) |
| status, body = raw_post(srv.url, "/v1/systemone", JEV["ok_all"]["request"]) |
| assert status == 403 |
| assert body == {"detail": {"error_type": "authentication_error", "message": |
| "Must supply an API key! Check your request and try again."}} |
| finally: |
| srv.close() |
|
|
|
|
| |
|
|
|
|
| @pytest.mark.parametrize("case", ["ok_all", "uncertain", "obj_state", "extra_q_name", |
| "one_option", "one_level"]) |
| def test_systemone_live_shapes(server, case): |
| live = JEV[case] |
| status, body = raw_post(server.url, "/v1/systemone", live["request"]) |
| assert status == live["status"] == 200 |
| assert shape(body) == shape(live["response"]) |
| assert list(body["answers"]) == list(live["request"]["questions"]) |
| for qid, ans in body["answers"].items(): |
| if ans["type"] in ("choice", "score"): |
| probs = list(ans["probabilities"].values()) |
| assert abs(sum(probs) - 1) < 1e-9 |
| assert ans["confidence"] == pytest.approx(confidence(probs) if len(probs) > 1 |
| else 1.0) |
| if ans["type"] == "score": |
| assert ans["legend"] == live["response"]["answers"][qid]["legend"] |
| assert ans["score"] == pytest.approx(sum(int(k) * v for k, v in |
| ans["probabilities"].items())) |
| assert body["model"] == MODEL_ID |
|
|
|
|
| @pytest.mark.parametrize("case", ["unknown_type", "extra_top", "bad_model", "opts256", |
| "eleven_levels", "no_model", "no_questions"]) |
| def test_systemone_live_errors(server, case): |
| live = JEV[case] |
| request = json.loads(json.dumps(live["request"])) |
| |
| |
| if case == "unknown_type": |
| request["questions"]["x"]["type"] = "classify" |
| if case == "extra_top": |
| request["foo"] = request.pop("options") |
| status, body = raw_post(server.url, "/v1/systemone", request) |
| assert status == live["status"] |
| if case == "bad_model": |
| assert body == live["response"] |
| elif status == 400: |
| assert body == live["response"] |
| else: |
| got = [{k: d[k] for k in ("type", "loc", "msg")} for d in body["detail"]] |
| want = [{k: d[k] for k in ("type", "loc", "msg")} for d in live["response"]["detail"]] |
| assert got == want |
|
|
|
|
| def test_systemone_noul_criteria_and_score_levels_in_schema(server): |
| raw_post(server.url, "/v1/systemone", JEV["ok_all"]["request"]) |
| _, schema, _ = server.fill.calls[-1] |
| props = schema["properties"] |
| assert props["urgent_crit"]["description"] == ( |
| "Is the customer asking for urgent action?\nYes means: Asks for action now\n" |
| "No means: No time pressure") |
| assert props["anger"] == {"type": "integer", "minimum": 0, "maximum": 3, |
| "description": "How angry is the customer?", |
| "x-levels": {"0": "calm", "1": "annoyed", "2": "angry", |
| "3": "furious"}} |
| assert props["department"]["oneOf"][2] == {"const": "shipping"} |
|
|
|
|
| def test_systemone_extensions(client): |
| got = client.systemone(EVENT_INPUT, { |
| "event": {"type": "choice", "instructions": "Which event?", |
| "criteria": {"refund": "money back", "replacement": None}}, |
| "order_id": {"type": "extract", "instructions": "The order id."}, |
| "note": {"type": "generate", "instructions": "Note for the agent.", "max_length": 4}}) |
| assert got.answers["event"].choice == "refund" |
| assert got.answers["order_id"] == {"type": "extract", "value": "AB-1234", "span": [10, 17], |
| "confidence": 0.9} |
| assert got.answers["note"] == {"type": "generate", "text": "note", "output_tokens": 2} |
| assert got.usage == {"input_tokens": got.usage.input_tokens, "output_tokens": 2} |
|
|
|
|
| def test_systemone_existing_jev_client_unchanged(server): |
| """The Clef/SystemOne client (rwkv_omarchy.rl.clef_server.ClefJudge, model |
| "clef-flash") works against deg as-is.""" |
| clef = pytest.importorskip("rwkv_omarchy.rl.clef_server") |
| QUESTIONS, ClefJudge, build_request = clef.QUESTIONS, clef.ClefJudge, clef.build_request |
|
|
| judge = ClefJudge(server.url) |
| out = judge.judge_request(build_request("fix dns", [{"command": "ls", "output": "x"}], |
| "Done, DNS fixed.")) |
| assert set(out) >= set(QUESTIONS) and out["input_tokens"] > 0 |
|
|
|
|
| |
|
|
|
|
| def test_fill_round_trip_and_x_generate_on_enum(client): |
| schema = {"type": "object", "properties": { |
| "event": {"enum": ["refund", "replacement"]}, |
| "order_id": {"type": "string"}, |
| "note": {"type": "string", "x-generate": True, "maxLength": 4}}, |
| "required": ["event", "order_id", "note"]} |
| got = client.fill(EVENT_INPUT, schema) |
| assert got.value == {"event": "refund", "order_id": "AB-1234", "note": "note"} |
| assert got.meta["/order_id"].span == [10, 17] |
| assert got.usage.generated_tokens == 2 |
| bad = json.loads(json.dumps(schema)) |
| bad["properties"]["event"]["x-generate"] = True |
| with pytest.raises(DegAPIError) as exc: |
| client.fill(EVENT_INPUT, bad) |
| assert exc.value.status == 422 |
| assert exc.value.body["error"]["type"] == "invalid_request" |
| assert any("x-generate" in d["msg"] for d in exc.value.body["error"]["detail"]) |
|
|
|
|
| def test_fill_unknown_option_is_422(client): |
| with pytest.raises(DegAPIError) as exc: |
| client.fill("x", {"type": "object", "properties": {"a": {"type": "boolean"}}}, |
| options={"beam": 4}) |
| assert exc.value.status == 422 |
|
|
|
|
| def test_malformed_json_bodies(server): |
| assert raw_post(server.url, "/v1/decisions", b"{")[0] == 400 |
| assert raw_post(server.url, "/v1/systemone", b"{")[0] == 422 |
| assert raw_post(server.url, "/v1/fill", b"{")[0] == 422 |
| assert raw_post(server.url, "/v1/nope", {})[0] == 404 |
|
|
|
|
| def test_models_and_health(client): |
| ids = [m.id for m in client.models().data] |
| assert {"deg-latest", MODEL_ID, "gpt-6-luna", "jev-latest"} <= set(ids) |
| health = client.health() |
| assert health.ok and health.model == MODEL_ID and health.batcher["jobs"] >= 0 |
|
|
|
|
| |
|
|
|
|
| def test_batcher_groups_concurrent_jobs_and_isolates_failures(): |
| seen: list[int] = [] |
| fake = FakeFill() |
|
|
| def fill(ctx, schema, opts): |
| if ctx == "bad": |
| raise ValueError("boom") |
| return fake(ctx, schema, opts) |
|
|
| def fill_many(jobs): |
| seen.append(len(jobs)) |
| time.sleep(0.05) |
| return [ValueError("boom") if ctx == "bad" else fake(ctx, schema, opts) |
| for ctx, schema, opts in jobs] |
|
|
| srv = Server(fill=fill, fill_many=fill_many, window_s=0.2) |
| try: |
| req = {"model": "deg-latest", "questions": [{"type": "predicate", |
| "instructions": "refund?"}]} |
| results: dict[int, tuple[int, Any]] = {} |
|
|
| def go(i: int) -> None: |
| results[i] = raw_post(srv.url, "/v1/decisions", |
| dict(req, input="bad" if i == 3 else f"refund {i}")) |
|
|
| threads = [threading.Thread(target=go, args=(i,)) for i in range(6)] |
| for t in threads: |
| t.start() |
| for t in threads: |
| t.join() |
| assert max(seen) > 1 |
| assert srv.batcher.stats()["max_seen_batch"] > 1 |
| assert results[3][0] == 500 and results[3][1]["error"]["type"] == "server_error" |
| assert all(results[i][0] == 200 for i in range(6) if i != 3) |
| finally: |
| srv.close() |
|
|
|
|
| def test_server_fill_many_equals_sequential_fill(): |
| """The real runtime bound like ``--ckpt`` (``bind_model``) on a tiny random Qwen3.5: |
| concurrent requests batched through ``fill_many`` answer exactly as one-at-a-time |
| ``fill`` (SPEC.md §10).""" |
| torch = pytest.importorskip("torch") |
| pytest.importorskip("transformers") |
| from test_model import SNAPSHOT, tiny_lm |
|
|
| from deg.model import load_model, load_tokenizer |
| from deg.server import bind_model |
|
|
| if SNAPSHOT is None: |
| pytest.skip("Qwen3.5-0.8B tokenizer not cached") |
| model = load_model(None, device="cpu", dtype=torch.float32, lm=tiny_lm(), |
| tokenizer=load_tokenizer(SNAPSHOT)) |
| torch.manual_seed(1) |
| with torch.no_grad(): |
| for p in model.heads.parameters(): |
| p.normal_(0.0, 0.5) |
| fill, fill_many = bind_model(model) |
| assert fill_many is not None |
| calls: list[int] = [] |
|
|
| def counted_many(jobs): |
| calls.append(len(jobs)) |
| return fill_many(jobs) |
|
|
| reqs = [ |
| ("/v1/decisions", {"model": "deg-latest", "input": EVENT_INPUT, "questions": [ |
| *EVENT_QUESTIONS[:4], |
| {"type": "score", "name": "anger", "instructions": "How angry?", |
| "levels": [{"label": "calm"}, {"label": "angry", "description": "upset"}]}, |
| {"type": "generate", "name": "n1", "instructions": "Short note.", "max_tokens": 3}, |
| {"type": "generate", "name": "n2", "instructions": "Reply.", "max_tokens": 5}]}), |
| ("/v1/systemone", {"model": "jev-latest", "state": {"msg": "late parcel, angry"}, |
| "questions": {"anger": {"type": "score", "instructions": "anger", |
| "criteria": ["calm", "angry", "furious"]}, |
| "late": {"type": "noul", "instructions": "late?"}}}), |
| ("/v1/decisions", {"model": "deg-latest", "input": "Order AB-9 for $80.", |
| "questions": [EVENT_QUESTIONS[0], EVENT_QUESTIONS[2]]}), |
| ] |
| seq = Server(fill=fill, fill_many=None) |
| try: |
| want = [raw_post(seq.url, path, body) for path, body in reqs] |
| finally: |
| seq.close() |
| assert all(status == 200 for status, _ in want), want |
| srv = Server(fill=fill, fill_many=counted_many, window_s=0.5) |
| try: |
| got: dict[int, Any] = {} |
| threads = [threading.Thread(target=lambda i=i: got.__setitem__(i, raw_post( |
| srv.url, *reqs[i]))) for i in range(len(reqs))] |
| for t in threads: |
| t.start() |
| for t in threads: |
| t.join() |
| finally: |
| srv.close() |
| assert calls and max(calls) > 1 |
| assert [got[i] for i in range(len(reqs))] == want |
|
|
|
|
| def test_batcher_rejects_when_full(): |
| gate = threading.Event() |
|
|
| def slow(ctx, schema, opts): |
| gate.wait(5) |
| return FakeFill()(ctx, schema, opts) |
|
|
| srv = Server(fill=slow, max_pending=1) |
| try: |
| req = {"model": "deg-latest", "input": "x", |
| "questions": [{"type": "predicate", "instructions": "y"}]} |
| out: list = [] |
| threads = [threading.Thread(target=lambda: out.append( |
| raw_post(srv.url, "/v1/decisions", req))) for _ in range(4)] |
| for t in threads: |
| t.start() |
| time.sleep(0.5) |
| gate.set() |
| for t in threads: |
| t.join() |
| codes = sorted(c for c, _ in out) |
| assert 429 in codes and 200 in codes |
| body = next(b for c, b in out if c == 429) |
| assert body["error"]["code"] == "rate_limit_exceeded" |
| finally: |
| srv.close() |
|
|
|
|
| |
|
|
|
|
| def test_parse_is_shared_between_apis(): |
| """predicate == noul, choice == choice, score == score: same field schemas.""" |
| o = parse_decisions({"model": "deg-latest", "input": "s", "questions": [ |
| {"type": "predicate", "name": "a", "instructions": "i"}, |
| {"type": "choice", "name": "b", "instructions": "i", |
| "choices": [{"value": "x", "description": "dx"}, {"value": "y"}]}, |
| {"type": "score", "name": "c", "instructions": "i", |
| "levels": [{"label": "lo"}, {"label": "hi"}]}]}) |
| j = parse_systemone({"model": "deg-latest", "state": "s", "questions": { |
| "a": {"type": "noul", "instructions": "i"}, |
| "b": {"type": "choice", "instructions": "i", "criteria": {"x": "dx", "y": None}}, |
| "c": {"type": "score", "instructions": "i", "criteria": ["lo", "hi"]}}}) |
| assert o.schema == j.schema |
|
|
