Testing without a model

Tests that call a real model are slow, cost money, and give a different answer each time. FakeModel replaces the model and nothing else: the engine, your tools, the schema checks, approvals, storage and memory all run for real. Use it for unit tests and CI; keep real models for a few manual checks and for evals.

Script the replies#

Each entry of the script is the next model reply: a string is a text answer, FakeReply.tool_call(name, args) (FakeReply.toolCall) is a tool call, FakeReply.tool_calls([...]) (toolCalls) is several at once. After the script runs out, the model echoes the last user message.

def test_the_agent_looks_the_order_up() -> None:
    model = FakeModel([
        FakeReply.tool_call("get_order", {"order_id": "A-1001"}),  # 1st model call
        "Your order has shipped.",                                  # 2nd model call
    ])
    agent = Agent(model, tools=[get_order])
    result = agent.run_sync("Where is A-1001?")

    assert result.text == "Your order has shipped."
    assert (result.model_calls, result.tool_calls) == (2, 1)
    tool_message = result.messages[2]  # user, assistant (the call), tool, assistant
    assert tool_message["tool"] == "get_order" and tool_message["status"] == "success"

test("the agent looks the order up", async () => {
  const model = new FakeModel([
    FakeReply.toolCall("get_order", { orderId: "A-1001" }), // 1st model call
    "Your order has shipped.", // 2nd model call
  ]);
  const result = await new Agent({ model, tools: [getOrder] }).run("Where is A-1001?");

  assert.equal(result.text, "Your order has shipped.");
  assert.deepEqual([result.modelCalls, result.toolCalls], [2, 1]);
  const toolMessage = result.messages[2]; // user, assistant (the call), tool, assistant
  assert.equal(toolMessage.tool, "get_order");
  assert.equal(toolMessage.status, "success");
});

What to assert:

result.text, result.output, result.stop_reason;
result.model_calls, result.tool_calls;
the transcript, result.messages: each tool message has tool, status (success, error, rejected) and content; an error has error["code"] (invalid_arguments, tool_error, tool_failed, ...).

The arguments in a scripted tool call go through the same schema check as a real model's, so a test can also show that a bad call reaches the model as invalid_arguments and never your function.

Many runs at once#

By default the script is used up in order, across every run of the model. With by_turn=True (byTurn: true), each run starts from the top of the script: the first reply for a run's first model call, the second for its second, and so on. One fake model then serves many runs, also concurrently:

def test_many_conversations_at_once() -> None:
    # by_turn: the reply depends on the model call's place in its run, not on
    # how many calls came before; one model serves any number of runs.
    model = FakeModel([FakeReply.tool_call("get_order", {"order_id": "A-1"}), "Shipped."], by_turn=True)
    agent = Agent(model, tools=[get_order], storage=Storage())

    async def scenario() -> None:
        replies = await asyncio.gather(*(agent.thread(f"t-{i}").send("Where?") for i in range(20)))
        assert all(r.text == "Shipped." for r in replies)

    asyncio.run(scenario())

test("many conversations at once", async () => {
  // byTurn: the reply depends on the model call's place in its run, not on
  // how many calls came before; one model serves any number of runs.
  const model = new FakeModel([FakeReply.toolCall("get_order", { orderId: "A-1" }), "Shipped."], { byTurn: true });
  const agent = new Agent({ model, tools: [getOrder], storage: new Storage() });
  const replies = await Promise.all(Array.from({ length: 20 }, (_, i) => agent.thread(`t-${i}`).send("Where?")));
  assert.ok(replies.every((r) => r.text === "Shipped."));
});

Failures, retries and fallbacks#

fail_first=n (failFirst) makes the first n calls fail with rate_limited, a transient error, so you can test retries and fallbacks. name sets the model name in results, to tell fake models apart:

def test_a_rate_limit_is_retried() -> None:
    # The first two calls fail with rate_limited (transient), the third answers.
    model = FakeModel(["Hello."], fail_first=2)
    agent = Agent(model, model_retry=Retry(max_attempts=3, initial_backoff=0.01))
    assert agent.run_sync("Hi").text == "Hello."


def test_the_fallback_answers_when_the_first_model_is_down() -> None:
    down = FakeModel(["never"], fail_first=100, name="primary")
    backup = FakeModel(["From the backup."], name="backup")
    agent = Agent(down, fallbacks=[backup], model_retry=Retry(max_attempts=1))
    result = agent.run_sync("Hi")
    assert result.text == "From the backup."
    assert result.answered_by == {"provider": "fake", "model": "backup"}

test("a rate limit is retried", async () => {
  // The first two calls fail with rate_limited (transient), the third answers.
  const model = new FakeModel(["Hello."], { failFirst: 2 });
  const agent = new Agent({ model, modelRetry: { maxAttempts: 3, initialBackoffSecs: 0.01 } });
  assert.equal((await agent.run("Hi")).text, "Hello.");
});

test("the fallback answers when the first model is down", async () => {
  const down = new FakeModel(["never"], { failFirst: 100, name: "primary" });
  const backup = new FakeModel(["From the backup."], { name: "backup" });
  const agent = new Agent({ model: down, fallbacks: [backup], modelRetry: { maxAttempts: 1 } });
  const result = await agent.run("Hi");
  assert.equal(result.text, "From the backup.");
  assert.deepEqual(result.answeredBy, { provider: "fake", model: "backup" });
});

Keep retry backoffs small in tests (initial_backoff=0.01), or the test waits for real. delay (delayMs) makes each reply take time, to test cancellation, timeouts and busy threads.

Approvals, threads and memory#

Use a fresh in-memory Storage() per test; nothing is left behind. The tutorial has a complete test of an approval: send, check needs_approval, approve, check the answer. Long-term memory works with a fake model too; for memory by meaning, Embedder.fake() gives vectors without a network call (texts that share words are close; synonyms adds more), and Embedder.fake(failing=True) tests the fall back to keywords.

Workflows#

Workflow nodes are your own functions: call them directly in unit tests, and run the compiled workflow for the wiring. Agent nodes take a FakeModel like any agent.

Errors in tests#

Test the failure paths as well as the happy path:

To test	Do
A tool's `ToolError`	Script a call with arguments that make the tool raise; assert the tool message's `status` is `error`
A crashing tool and `on_tool_exception="fail"`	Expect `ToolExecutionError`, with your exception as `__cause__` / `cause`
A limit	Set `limits={"model_calls": 2}` and script more tool calls; expect `LimitExceededError`
A model outage	`fail_first` larger than the retries; expect `ModelError`
A busy thread	`Storage(on_busy="reject")`, a `delay`, two concurrent `send`s; expect `ThreadBusyError`

In Rust#

Rust's FakeModel also fails on purpose per call (failing_on, flaky_first):

use criogenai_agent::Agent;
use criogenai_providers::{FakeModel, FakeReply};
use serde_json::json;

// Scripted replies: no network, no cost, the same answer every time.
let model = FakeModel::new()
    .then(FakeReply::tool_call(
        "get_order",
        json!({"order_id": "A-1001"}),
    ))
    .then(FakeReply::text("It ships tomorrow."));
let agent = Agent::builder(model).build()?; // add your tools as usual

Next#

Evals and quality gates: measure real models on your cases.
Tracing: look at a run step by step.
API: Python models (FakeModel), TypeScript models.