Skip to content

testing.md

Latest commit

 

History

History
163 lines (103 loc) · 11.2 KB

testing.md

File metadata and controls

163 lines (103 loc) · 11.2 KB
Raw

Testing

composeai.testing is the dev/test kit: FakeModel scripts an agent with no network or provider SDK, cassettes record real model traffic once and replay it offline forever, @agent(cache=True) speeds up local iteration against a real provider, and reset_registries() clears the @agent/@flow/@task name registries between tests. None of it requires a provider SDK to be installed — it only ever sees whatever registry.resolve() would have returned, real or FakeModel.

FakeModel

FakeModel(script, *, usage=None) implements the Model protocol with a fixed, ordered list of scripted responses — one item consumed per complete() (or stream()) call:

from pydantic import BaseModel, Field

import composeai as compose
from composeai.testing import FakeModel


class FactSheet(BaseModel):
    topic: str
    key_facts: list[str] = Field(description="Three crisp, verifiable facts")


@compose.tool
def count_words(text: str) -> int:
    """Count the words in a piece of text.

    Args:
        text: The text whose words should be counted.
    """
    return len(text.split())


model = FakeModel([
    {"tool_calls": [{"name": "count_words", "arguments": {"text": "a b c"}}]},  # turn 1
    {"json": {"topic": "quantum computing", "key_facts": ["fact one"]}},        # turn 2
])


@compose.agent(model=model, tools=[count_words])
def researcher(topic: str) -> FactSheet:
    """You are a meticulous researcher."""
    return compose.prompt(f"Build a fact sheet about: {topic}")


facts = researcher("quantum computing")

Each script item is one of five forms:

  • str — a plain text response, stop reason END_TURN.
  • {"tool_calls": [{"name", "arguments", "id": optional}, ...], "text": optional} — an assistant message with one ToolCallPart per entry (plus a leading TextPart if "text" is given), stop reason TOOL_USE. A missing "id" is autogenerated.
  • {"json": {...}} — a structured response: parsed is set to the dict, the message's text is its JSON encoding, stop reason END_TURN.
  • a ModelResponse — returned as-is, letting a test control every field directly (stop_reason, usage, a ThinkingPart, ...).
  • a callable — invoked with the ModelRequest; its return value is interpreted using the rules above (any form except another callable).

usage= (a Usage instance) applies to every non-ModelResponse item — pass it to make token/cost math exact and deterministic in a test:

from composeai.messages import Usage

FakeModel(["done"], usage=Usage(input_tokens=10, output_tokens=10, cost_usd=0.05))

Left unset, each response reports Usage(input_tokens=10, output_tokens=20, cost_complete=False) — nonzero token counts at an unknown price (cost_usd=None), the same convention a real adapter uses for a call it can't price; that keeps a Budget(usd=...) test from accidentally seeing scripted spend it never asked for. See budgets.

.stream() derives word-level text_delta/thinking_delta events (and tool_call_started/tool_args_delta/tool_call_finished for tool calls) from the same script item .complete() would have used, ending in a response_done event carrying the full response — consuming only that last event is equivalent to having called .complete(). Every request the agent made is recorded, in order, in model.requests for assertions. Once the script is exhausted, a further call raises ComposeError naming how many requests were made.

Cassettes

Cassettes record a resolved model's real responses to a JSON file and replay them later — offline, deterministically, with no provider adapter ever constructed on replay.

Recording wraps every model registry.resolve() returns for the duration of a with block, appending one entry per complete() call (or per stream() call's final response):

from composeai.testing import record_cassette

with record_cassette("tests/cassettes/researcher.json"):
    facts = researcher("quantum computing")

On exit it writes {"version": 2, "entries": [...]}, even if the block raised — whatever was recorded before the failure is kept. record_cassette/replay_cassette can't be nested; entering one while another is already active raises ConfigError.

Replaying intercepts registry.resolve() to return one shared model built from the file's entries, without ever calling the real resolve() — no provider adapter (or its SDK) is constructed:

from composeai.testing import replay_cassette

with replay_cassette("tests/cassettes/researcher.json"):
    facts = researcher("quantum computing")

A cassette whose "version" isn't the current format raises ConfigError telling you to re-record — a v1 file predates full_hash covering the provider label and each tool's full spec, so it can never produce a meaningful match under the current hashing.

Matching: each entry carries two hashes, both sha256 of canonical JSON. full_hash covers (provider, model, system, messages, tools, output_schema, max_tokens, temperature) — the full request; two agents that happen to share a tool name with different schemas or a different requires_approval gate never share an entry. message_hash covers only (system, messages) — the one thing a cassette built by compose export from a persisted trace span always has, since that path never recorded provider/tools/output_schema/max_tokens/temperature in the first place. Replay tries full_hash first and falls back to message_hash only when nothing (still-available) matches on full_hash; entries are consumed in order within whichever hash matched, so the same request made twice against two matching entries gets the first, then the second.

The cassette pytest fixture picks the right mode automatically — re-export it from your conftest.py so it's requestable by name in any test module:

from composeai.testing import cassette  # noqa: F401
def test_researcher(cassette):
    with cassette("tests/cassettes/researcher.json"):
        facts = researcher("quantum computing")
  • COMPOSE_RECORD=1 set in the environment → records (always re-records, even if the file already exists).
  • Else, the file already exists → replays it.
  • Else → live passthrough: neither wraps nor intercepts anything, and no file is written.

Record a fresh cassette with COMPOSE_RECORD=1 pytest tests/test_researcher.py against the real provider, then commit the resulting JSON file — every later run replays it with no key, no network, and no SDK constructed. compose export RUN_ID --cassette path.json builds the same file format directly from an already-persisted run, without re-running anything; see observability.

@agent(cache=True)

cache=True wraps the resolved model (and fallback, if any) in CachingModel: a filesystem-backed cache of complete() responses under {COMPOSE_DIR}/cache/, keyed by the same full-request hash cassettes use. Meant for iterating locally against a real, billed provider — repeated runs during development don't re-pay for identical calls:

@compose.agent(model="anthropic/claude-sonnet-5", cache=True)
def researcher(topic: str) -> FactSheet:
    """You are a meticulous researcher."""
    return compose.prompt(f"Build a fact sheet about: {topic}")


researcher("quantum computing")   # real call, written to the cache
researcher("quantum computing")   # identical request -> cache hit, no call made

A hit skips the real call entirely and reports usage=Usage() (zeroed) on the llm span — a cache hit is never re-billed for the tokens it originally cost — and tags the span attributes["cached"] = True. Applies to complete() only: when the wrapped model also supports .stream(), CachingModel exposes it unwrapped, straight passthrough — a streaming call always runs for real and is never cached, since replaying stored deltas isn't implemented.

For deterministic, offline test fixtures, prefer a cassette instead — cache=True is a dev-loop convenience against a real provider, not a substitute for a committed, replayable fixture.

New in this release: CachingModel.last_was_hit is thread-local. A single agent's model slot (and therefore its CachingModel) is shared across a pipe.map/aggregate fan-out's worker threads; a plain instance attribute would let one thread's hit/miss overwrite another's between the complete() call and the read. Each thread now observes only the outcome of its own most recent call, regardless of how concurrently the calls overlap — the guarantee composeai.agentfn._call_llm relies on when it reads last_was_hit right after invoking the model, to decide whether to tag that call's own span cached=True.

reset_registries()

@agent/@flow/@task names are unique per process, since resume() uses the name to route a paused or crashed durable run back to its definition. That collides with a test suite that redefines the same function name across test modules or parametrized cases — the second decoration raises ConfigError.

composeai.testing.reset_registries() clears all three registries (the supported replacement for reaching into the private _AGENT_REGISTRY/_FLOW_REGISTRY/_TASK_REGISTRY dicts by hand):

import pytest
from composeai.testing import reset_registries


@pytest.fixture(autouse=True)
def _reset():
    yield
    reset_registries()

After a reset, resume() can't route a run back to a definition until its module re-registers it (re-import or re-decoration) — harmless in a test suite, where each test function re-decorates its own agents/flows/tasks fresh.

replace=True vs. name=

Both exist to avoid the "duplicate name" ConfigError, for different situations:

  • name= gives an agent/flow/task an explicit, unique registered name instead of relying on the function's __name__ — reach for it when two definitions legitimately need to coexist (a runtime-bound factory that builds several agents from one template, or two test modules that both happen to define a function called greeter).
  • replace=True re-binds an existing name instead of raising — reach for it for a single, deliberate rebind (a fixture that redefines the same agent under the same name every test run) rather than a blanket per-test reset.

For a whole test suite that redefines many agents/flows/tasks across many modules, reset_registries() (above) as an autouse fixture is usually simpler than threading name=/replace=True through every definition — pick whichever matches how much of the suite needs it. One caveat specific to @agent(replace=True): standalone-agent resume has no fingerprint/staleness check (unlike @flow, which hashes its source at decoration time), so a paused run resumed after a replace=True rebind continues silently against the new definition.

See also

agents covers @agent(model=...) accepting a FakeModel (or any Model) directly, and name=/replace= in depth; flows covers the fingerprint check @flow has and @task/@flow's own name=/replace=; observability covers compose export, the CLI command that builds a cassette from an already-persisted run; budgets covers Usage/Budget interaction with a scripted FakeModel's usage=.