loom.testing

Functions

__getattr__(name)	Resolve exports whose dependencies belong to optional extras on first access.
_http_test_harness()	Import HttpTestHarness, naming the rest extra when its dependencies are missing.
_repository_harness_export(name)	Import one repository_harness export, naming the sqlalchemy extra when missing.

class loom.testing.AgentHandleDouble(name)

Bases: object

In-memory double for AgentHandle.

Bound to a use case’s parameter through UseCaseTest.with_agent().

Stands in for the handle a real Agent() marker resolves to: no network, model or database call is ever made. Every run mode returns an answer scripted through on_run() / on_run_text(); every grant view is a further double reached through mcp() / sql() and cached, so scheduling its answers once configures every call the use case makes through it. Every call this handle itself receives is recorded on run_calls / run_text_calls, so a test can assert what the use case asked of it.

Parameters:

name (str) – Agent name as the Agent(name) marker declares it, named in this double’s refusal messages.

Example:

triage = AgentHandleDouble("incident-triage").on_run(SeverityAssessment(severity=5))
result = await (
    UseCaseTest(TriageIncidentUseCase())
    .with_caller(identity)
    .with_agent("incident-triage", triage)
    .with_params(incident_id="INC-1")
    .run()
)
assert triage.run_calls[0].prompt == "Assess INC-1."

on_run(output, *, usage=None, interaction_id=None)

Schedule the next run() call’s answer, declared or per-run shape alike.

Returns:

self for chaining.

Parameters:

• output (Any)

• usage (AgentUsage | None)

• interaction_id (str | None)

Return type:

AgentHandleDouble

on_run_text(text, *, usage=None, interaction_id=None)

Schedule the next run_text() call’s answer.

Returns:

self for chaining.

Parameters:

• text (str)

• usage (AgentUsage | None)

• interaction_id (str | None)

Return type:

AgentHandleDouble

async run(prompt, *, expect=None, conversation_id=None, state=None)

Record the call and return the answer scripted through on_run().

Raises:

AssertionError – If no answer was scheduled through on_run().

Parameters:

• prompt (str)

• expect (type[Any] | None)

• conversation_id (str | None)

• state (object | None)

Return type:

AgentAnswer[Any]

async run_text(prompt, *, conversation_id=None, state=None)

Record the call and return the answer scripted through on_run_text().

Raises:

AssertionError – If no answer was scheduled through on_run_text().

Parameters:

• prompt (str)

• conversation_id (str | None)

• state (object | None)

Return type:

AgentAnswer[str]

mcp(server)

Return this agent’s own double for the mcp grant named server.

Built once and cached: repeated calls with the same name return the same double.

Parameters:

server (str)

Return type:

McpHandleDouble

sql(connection)

Return this agent’s own double for the sql grant named connection.

Built once and cached: repeated calls with the same name return the same double.

Parameters:

connection (str)

Return type:

SqlGrantHandleDouble

with_grants(*names)

Declare the grant names this double reports, and refuse the rest.

Without it the double reports what has been reached, which is not what the real handle promises: there, the listing names what the artefact declares, whether or not anything used it. A test that asserts a grant is available would pass against the double and prove nothing about production, so declaring the set here makes the two agree.

Parameters:

names (str) – Every server and connection this agent declares.

Returns:

This double, for chaining.

Return type:

AgentHandleDouble

grants()

Return the declared grant names, matching what the real handle reports.

Falls back to what has been reached when nothing was declared, so a test that does not care keeps working.

Return type:

tuple[str, …]

class loom.testing.ContractScenario(*, expected_output=None, events=None, error_code=None)

Bases: LoomFrozenStruct

Engine behaviour one contract check requires.

An AgentPlan declares structure, not behaviour, while every contract check needs a scripted behaviour: a success run with its events, or a failure with its coded error. The scenario is therefore the right seam — the suite hands this neutral description and the adapter under test builds an engine exhibiting it: the fake maps it onto a script, and a real-engine adapter can map it onto a stubbed provider (FR-048).

Parameters:

• expected_output (object)

• events (tuple[TextDeltaEvent | ToolCallEvent | ToolResultEvent | ErrorEvent | FinalEvent, ...] | None)

• error_code (AgentRunErrorCode | None)

expected_output

Output run() and the terminal FinalEvent must produce in a success scenario.

Type:

object

events

Events the engine may replay in a success scenario, ending in a FinalEvent; the suite checks stream structure only, never that these exact events come back. None lets the engine choose its own events.

Type:

tuple[loom.ai.abc.TextDeltaEvent | loom.ai.abc.ToolCallEvent | loom.ai.abc.ToolResultEvent | loom.ai.abc.ErrorEvent | loom.ai.abc.FinalEvent, …] | None

error_code

When set, the scenario is a failure: the stream must end in an ErrorEvent with this code, and events is ignored.

Type:

loom.ai.errors.AgentRunErrorCode | None

class loom.testing.FakeAgentEngine(*, script=None, output=None)

Bases: object

Deterministic, offline AgentEngine test double.

Replays a fixed event script: no network, no credentials, no clocks and no randomness, so two instances built from the same arguments produce byte-for-byte identical results and streams.

Parameters:

• script (Sequence[AgentEvent] | None) – Event sequence to replay. Must end in exactly one terminal event (FinalEvent or ErrorEvent), with no terminal event before the last position. When omitted, a fixed default script ending in a FinalEvent carrying output is replayed.

• output (object | None) – Output of the default script’s FinalEvent. Ignored when script is provided.

Raises:

ValueError – If script is empty, does not end in a terminal event, or contains a terminal event before the last position.

Example:

engine = FakeAgentEngine(output={"answer": 42})
result = await engine.run("question", identity=identity)

async run(prompt, *, identity, conversation=None, state=None)

Replay the script to completion.

Parameters:

• prompt (str) – Caller prompt; ignored, the script is fixed.

• identity (Identity) – Verified caller; ignored, the script is fixed.

• conversation (Conversation | None) – Conversation to continue; ignored, the script is fixed.

• state (object | None) – This run’s state; ignored, the script is fixed.

Returns:

The terminal FinalEvent’s output, usage and messages.

Raises:

FakeAgentRunError – If the script ends in an ErrorEvent.

Return type:

AgentResult

run_stream(prompt, *, identity, conversation=None, state=None)

Replay the script as an event stream.

The returned context manager closes the iterator on exit via aclose() — deterministically, never left to the garbage collector — mirroring how a real engine must release its provider connection.

Parameters:

• prompt (str) – Caller prompt; ignored, the script is fixed.

• identity (Identity) – Verified caller; ignored, the script is fixed.

• conversation (Conversation | None) – Conversation to continue; ignored, the script is fixed.

• state (object | None) – This run’s state; ignored, the script is fixed.

Returns:

An async context manager yielding the scripted event stream.

Return type:

AbstractAsyncContextManager[AsyncIterator[TextDeltaEvent | ToolCallEvent | ToolResultEvent | ErrorEvent | FinalEvent]]

async health()

Report a fixed "ok" status without any I/O.

Returns:

the fake has no dependency that could degrade.

Return type:

Always "ok"

class loom.testing.GoldenHarness

Bases: object

Executes use cases in isolation with fake repositories.

Allows injecting fake repo instances, forcing errors on specific methods, and asserting performance baselines — all without a real database.

Example:

harness = GoldenHarness()
harness.inject_repo(IProductRepo, FakeProductRepo(), model=Product)
harness.force_error(IProductRepo, "create", Conflict("duplicate"))
result = await harness.run(CreateProductUseCase, payload={"name": "X"})

inject_repo(interface, fake_instance, *, model=None)

Register a fake repository instance for an interface type.

Parameters:

• interface (type) – Repository interface used as the DI resolution key.

• fake_instance (Any) – Fake repository instance to inject.

• model (type | None) – Optional domain model to register a RepoFor mapping, enabling auto-injection for use cases that inherit the default UseCase.__init__.

Return type:

None

force_error(interface, method, error)

Force a specific repository method to raise an error.

Parameters:

• interface (type) – Repository interface type.

• method (str) – Method name to intercept.

• error (Exception) – Exception instance to raise on call.

Return type:

None

simulate_system_error(interface, method)

Simulate a SystemError on a specific repository method.

Parameters:

• interface (type) – Repository interface type.

• method (str) – Method name to intercept.

Return type:

None

async run(use_case_type, *, params=None, payload=None, identity=None)

Execute a use case with injected fake repositories.

Parameters:

• use_case_type (type[UseCase[Any, Any, RepoFor[Any]]]) – UseCase subclass to execute.

• params (dict[str, Any] | None) – Primitive parameter values keyed by name.

• payload (dict[str, Any] | None) – Raw dict for Input() command construction.

• identity (Identity | None) – Caller the execution runs as. Required by use cases declaring Caller(); pass ANONYMOUS to pin the unauthenticated path.

Returns:

Result produced by the use case.

Raises:

loom.core.errors.Unauthenticated – If the use case declares Caller() and no identity is given.

Return type:

Any

async run_with_baseline(use_case_type, *, params=None, payload=None, identity=None, name, max_ms, baseline_dir)

Execute a use case and assert it completes within a time baseline.

Writes a <name>.json file to baseline_dir recording the measured duration. Raises AssertionError if execution exceeds max_ms.

Parameters:

• use_case_type (type[UseCase[Any, Any, RepoFor[Any]]]) – UseCase subclass to execute.

• params (dict[str, Any] | None) – Primitive parameter values keyed by name.

• payload (dict[str, Any] | None) – Raw dict for Input() command construction.

• identity (Identity | None) – Caller the execution runs as.

• name (str) – Baseline identifier used as the filename.

• max_ms (float) – Maximum allowed execution duration in milliseconds.

• baseline_dir (Path) – Directory where baseline JSON files are written.

Returns:

Result produced by the use case.

Raises:

AssertionError – If elapsed time exceeds max_ms.

Return type:

Any

class loom.testing.InMemoryRepository(entity_type, *, id_field='id', creator=None)

Bases: Generic[T]

Generic in-memory repository for testing any msgspec.Struct model.

Stores entities in a plain dict keyed by their id field value. Provides the standard repository surface (get_by_id, create, update, delete, list_paginated) without any database dependency.

The create method derives entity fields from the command automatically when no creator callable is provided: fields present on the command are copied to the entity, and the id field is assigned from an internal auto-increment counter.

Parameters:

• entity_type (type[T]) – The msgspec.Struct subclass this repository stores.

• id_field (str) – Name of the identity field on the entity. Defaults to "id".

• creator (Callable[[Any, int], T] | None) – Optional (cmd, next_id) -> T callable used by create(). When provided, the automatic field-mapping is bypassed entirely.

Example:

repo = InMemoryRepository(Product, id_field="id")
repo.seed(Product(id=1, name="Widget"), Product(id=2, name="Gadget"))

harness = HttpTestHarness()
harness.inject_repo(Product, repo)
client = harness.build_app(interfaces=[ProductRestInterface])

seed(*entities)

Pre-load entities into the store.

The internal id counter is advanced past the highest integer id seen so that subsequent create() calls do not collide.

Parameters:

*entities (T) – Entity instances to load.

Return type:

None

Example:

repo.seed(Product(id=1, name="A"), Product(id=2, name="B"))

async get_by_id(obj_id, profile='default')

Return the entity with obj_id, or None if not found.

Parameters:

• obj_id (Any) – The identity value to look up.

• profile (str) – Ignored; present for repository interface compatibility.

Returns:

Entity instance, or None if no entity has that id.

Return type:

T | None

async create(cmd)

Create and store a new entity from cmd.

If a creator callable was provided at construction it is called as creator(cmd, next_id). Otherwise, command attributes whose names match entity fields are copied automatically, and the id field is set from the internal auto-increment counter.

Parameters:

cmd (Any) – Command or payload object carrying the new entity’s data.

Returns:

The created and stored entity.

Return type:

T

async update(obj_id, data)

Update the entity at obj_id with fields from data.

Only non-None fields present on both data and the entity are overwritten; the id field is never changed.

Parameters:

• obj_id (Any) – Identity value of the entity to update.

• data (Any) – Object or dict with updated field values.

Returns:

The updated entity, or None if no entity has that id.

Return type:

T | None

async delete(obj_id)

Delete the entity at obj_id.

Parameters:

obj_id (Any) – Identity value to delete.

Returns:

True if the entity existed and was removed, False if not found.

Return type:

bool

async list_paginated(*args, **kwargs)

Return all stored entities.

Parameters:

• *args (Any) – Ignored; present for repository interface compatibility.

• **kwargs (Any) – Ignored; present for repository interface compatibility.

Returns:

List of all entities in insertion order.

Return type:

list[T]

class loom.testing.McpHandleDouble(server)

Bases: object

In-memory double for McpHandle.

No network call is ever made. call() and call_untyped() return a result scripted per tool name through on_call() / on_call_untyped(); every invocation is recorded on calls so a test can assert what the use case asked of it.

Parameters:

server (str) – Server name this double stands in for, named in its refusal messages.

with_tools(*tools)

Script the tool names tools() returns.

Returns:

self for chaining.

Parameters:

tools (str)

Return type:

McpHandleDouble

on_call(tool, result)

Script the decoded result the next typed call() for tool returns.

Returns:

self for chaining.

Parameters:

• tool (str)

• result (Any)

Return type:

McpHandleDouble

on_call_untyped(tool, result)

Script the raw result the next call_untyped() for tool returns.

Returns:

self for chaining.

Parameters:

• tool (str)

• result (Mapping[str, Any])

Return type:

McpHandleDouble

tools()

Return the tool names scripted through with_tools().

Return type:

tuple[str, …]

async call(tool, arguments, *, expect)

Record the call and return the result scripted for tool.

Raises:

AssertionError – If no result was scheduled for tool through on_call().

Parameters:

• tool (str)

• arguments (Mapping[str, Any])

• expect (type[Any])

Return type:

Any

async call_untyped(tool, arguments)

Record the call and return the raw result scripted for tool.

Raises:

AssertionError – If no result was scheduled for tool through on_call_untyped().

Parameters:

• tool (str)

• arguments (Mapping[str, Any])

Return type:

Mapping[str, Any]

class loom.testing.SqlGrantHandleDouble(connection)

Bases: object

In-memory double for SqlGrantHandle.

No connection is ever opened. query() returns rows scripted through on_query(), and every call is recorded on calls.

Parameters:

connection (str) – Connection name this double stands in for, named in its refusal message.

on_query(rows)

Script the rows every query() call returns.

Returns:

self for chaining.

Parameters:

rows (Sequence[Mapping[str, Any]])

Return type:

SqlGrantHandleDouble

async query(statement, *, parameters=None)

Record the call and return the rows scripted through on_query().

Raises:

AssertionError – If no rows were scheduled through on_query().

Parameters:

• statement (str)

• parameters (Mapping[str, Any] | None)

Return type:

Sequence[Mapping[str, Any]]

class loom.testing.UseCaseTest(use_case)

Bases: Generic[ResultT]

Fluent test harness for executing UseCases without HTTP or framework overhead.

Builds and runs the real ExecutionPlan — no shortcuts or mocking of the pipeline. Designed for unit and integration tests that must exercise computes, rules, and load steps in full.

Parameters:

use_case (UseCase[Any, ResultT]) – Constructed UseCase instance to test.

Example:

result = await (
    UseCaseTest(UpdateUserUseCase(repo=fake_repo))
    .with_params(user_id=1)
    .with_input(email="new@example.com")
    .run()
)

with_params(**kwargs)

Set primitive parameter values bound by name.

Parameters:

**kwargs (Any) – Parameter names and values matching the UseCase’s non-Input, non-Load parameters.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

with_input(**kwargs)

Set raw payload fields for command construction.

The payload is passed to the Command’s from_payload method. Use with_command if you have a pre-built Command instance.

Parameters:

**kwargs (Any) – Payload fields matching the Command struct.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

with_command(cmd)

Set a pre-built Command instance as the execution payload.

Serializes the command via msgspec.to_builtins so it is compatible with the standard from_payload pipeline.

Parameters:

cmd (Any) – A Command (msgspec.Struct) instance.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

with_loaded(entity_type, entity)

Pre-load an entity, bypassing repository calls for this type.

Parameters:

• entity_type (type[Any]) – The entity class used in the LoadById() marker.

• entity (Any) – The pre-loaded entity instance.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

with_deps(entity_type, repo)

Register a repository for a given entity type.

Used when the UseCase has LoadById() steps that require a repo. with_loaded takes precedence over with_deps for the same type.

Parameters:

• entity_type (type[Any]) – The entity class used in the LoadById() marker.

• repo (Any) – Repository implementing get_by_id.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

with_caller(identity)

Run the use case as identity, filling its Caller() parameter.

Without this call a use case declaring Caller() fails closed, which is the point: an authorization test must state whose request it is.

Parameters:

identity (Identity) – Caller the execution runs as. Pass ANONYMOUS to exercise the unauthenticated path explicitly.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

with_agent(name, double)

Bind double to the Agent(name) marker parameter named name.

Without this call a use case declaring Agent(name) fails closed when run, naming the use case and the agent — the same fail-closed design with_caller() applies to Caller(), extended to the whole handle: every run mode and every grant view, not only the execution itself, so nothing that reaches the agent can run without a network, a model or a database standing in.

Parameters:

• name (str) – Agent name exactly as the Agent(name) marker declares it in the use case under test.

• double (AgentHandleDouble) – Pre-built AgentHandleDouble — script its answers with on_run() / on_run_text() and its grant views with mcp() / sql() before passing it here.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

with_mcp(server, double)

Bind double to the Mcp(server, include=...) marker parameter named server.

Without this call a use case declaring Mcp(server, ...) fails closed when run, naming the use case and the server — the same fail-closed design with_agent() applies to Agent(name).

The double is narrowed to the resolving binding’s own include before the use case ever sees it: a scripted tool outside include, or inside it but never scripted through with_tools(), is refused with the same AgentRunError(TOOL_UNKNOWN) production raises before any network call.

Parameters:

• server (str) – Server name exactly as the Mcp(server, ...) marker declares it in the use case under test.

• double (McpHandleDouble) – Pre-built McpHandleDouble — script its tools and results with with_tools() / on_call() / on_call_untyped() before passing it here.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

with_main_repo(repo)

Inject the main repository dependency into the UseCase instance.

This is useful for unit tests of UseCase[TModel, TResult] where the core logic reads from self.main_repo.

Parameters:

repo (RepoFor[Any]) – Repository instance compatible with the UseCase’s main model.

Returns:

self for chaining.

Return type:

UseCaseTest[ResultT]

async run()

Compile and execute the UseCase through the full pipeline.

Returns:

The result produced by the UseCase.

Raises:

• loom.core.errors.RuleViolations – If one or more rule steps fail.

• NotFound – If a Load step finds no entity.

• loom.core.errors.Unauthenticated – If the UseCase declares Caller(), Agent(name) or Mcp(server, ...) and no with_caller() was set.

• RuntimeError – If the UseCase declares Agent(name) and no matching with_agent() call registered a double for name, or declares Mcp(server, ...) and no matching with_mcp() call registered a double for server.

• loom.core.engine.compiler.CompilationError – If the UseCase fails structural validation.

Return type:

ResultT

property plan: ExecutionPlan

Compile and return the ExecutionPlan for the UseCase.

Useful for asserting plan structure in advanced test scenarios.

Returns:

The compiled ExecutionPlan.

loom.testing.agent_engine_contract_suite(engine_factory)

Run the shared AgentEngine contract suite (FR-048).

Every check exercises only the protocol surface and the loom.ai value types, never an engine’s internals, so the same suite validates the fake and any real engine adapter. Checks: run() result and usage shape, exactly-one-terminal streams for success and failure, usage carried only by FinalEvent, the full run-time error-code taxonomy with FR-028 retriability, deterministic stream closure on early exit, and a known health() status.

The function is synchronous by design: each check runs in its own fresh event loop via asyncio.run, so streams closed by one check can never leak into the next. Call it from a synchronous test; calling it from an async test with a running loop raises RuntimeError.

Parameters:

engine_factory (Callable[[ContractScenario], AgentEngine]) – Builds one engine exhibiting the behaviour a ContractScenario describes; called once per check invocation.

Raises:

AssertionError – If a check fails; the message names the check.

Return type:

None

loom.testing.serialize_plan(plan)

Produce a deterministic, JSON-serialisable snapshot of an ExecutionPlan.

All keys are sorted alphabetically so the output is stable across runs regardless of insertion order. Types and callables are encoded as their fully qualified module.qualname string.

Parameters:

plan (ExecutionPlan) – Compiled execution plan to serialise.

Returns:

Dictionary containing only JSON-primitive values (str, int, list, dict, None). Suitable for json.dumps comparison.

Return type:

dict[str, Any]

Example:

snapshot = serialize_plan(compiler.get_plan(MyUseCase))
assert snapshot["use_case"] == "my_app.use_cases.MyUseCase"