judgevet.testing

Offline fakes of the System One ports for caller tests.

FakeSystemOnePort satisfies judgevet.ports.SystemOnePort and AsyncFakeSystemOnePort satisfies judgevet.ports.AsyncSystemOnePort. Both make no network call. A scripted answer is returned for its question name. Every other typed question receives a seeded answer that passes the domain answer validators. The same seed, state and question give the same answer. A scripted Choice answer whose option is outside its question's criteria raises JevResponseError with status 200, as the HTTP adapters do.

Seeded values are synthetic. They exercise caller code paths; they do not predict what the service would answer. The returned model echoes the model argument.

Each fake takes six keyword-only arguments. Two script the whole call and four match the HTTP adapter arguments of the same name:

Parameters:

Examples:

from judgevet.domain.answers import NoulAnswer
from judgevet.domain.questions import Choice, Noul
from judgevet.testing import FakeSystemOnePort

fake = FakeSystemOnePort(seed=3, answers={"billing": NoulAnswer(noul=0.9)})
response = fake.system_one(
    state="I was charged twice.",
    questions={
        "billing": Noul(instructions="Is this about billing?"),
        "queue": Choice(criteria={"billing": "Money", "technical": "Bugs"}),
    },
    model="jev-1.13.0",
)
assert response.nouls["billing"].noul == 0.9
assert response.choices["queue"].choice in {"billing", "technical"}
assert len(fake.calls) == 1

from judgevet.domain.errors import JevRateLimitError
from judgevet.domain.usage import Usage

metered = FakeSystemOnePort(usage=Usage(input_tokens=12, output_tokens=3))
assert metered.system_one("text", {"q": Noul()}, "m").usage.input_tokens == 12
limited = FakeSystemOnePort(error=JevRateLimitError("slow down", 429))
try:
    limited.system_one("text", {"q": Noul()}, "m")
except JevRateLimitError as error:
    assert error.status_code == 429
else:
    raise AssertionError("the scripted error was not raised")
assert len(limited.calls) == 1

from judgevet.domain.errors import JevBudgetExceededError
from judgevet.domain.spend import SpendCap


class ListSink:
    def __init__(self) -> None:
        self.records = []

    def record(self, record) -> None:
        self.records.append(record)


sink = ListSink()
capped = FakeSystemOnePort(spend_cap=SpendCap(max_attempts=1), audit=sink)
capped.system_one("text", {"q": Noul()}, "m")
try:
    capped.system_one("text", {"q": Noul()}, "m")
except JevBudgetExceededError as error:
    assert error.limit == "attempts"
else:
    raise AssertionError("the spend cap did not refuse the second call")
assert [r.outcome for r in sink.records] == ["success", "error"]
assert len(capped.calls) == 1

Attributes:

import anyio
from judgevet.domain.questions import Noul


async def main() -> float:
    fake = AsyncFakeSystemOnePort(seed=1)
    response = await fake.system_one("text", {"q": Noul()}, "jev-1.13.0")
    return response.nouls["q"].noul


assert 0.0 <= anyio.run(main) <= 1.0

from judgevet.domain.questions import Score
from judgevet.ports import SystemOnePort

port: SystemOnePort = FakeSystemOnePort(seed=1)
response = port.system_one(
    "text", {"tone": Score(criteria=["cold", "warm"])}, "jev-1.13.0"
)
assert response.scores["tone"].legend == {0: "cold", 1: "warm"}