loom.testing

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)

Execute a use case with injected fake repositories.

Parameters:

• use_case_type (type[UseCase[Any, 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.

Returns:

Result produced by the use case.

Return type:

Any

async run_with_baseline(use_case_type, *, params=None, payload=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]]) – 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.

• 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.HttpTestHarness

Bases: object

Builds a FastAPI TestClient with injected fake repositories.

Combines bootstrap_app() and create_fastapi_app() into a single test-oriented entry point.

Repositories are keyed by domain model type (the TModel generic parameter of the use case) rather than by interface. The harness derives the correct binding automatically by scanning the use cases declared in the REST interfaces.

Parameters:

None

Example:

harness = HttpTestHarness()
harness.inject_repo(Product, InMemoryRepository(Product))
harness.force_error(Product, "get_by_id", NotFound("not found"))
client = harness.build_app(interfaces=[ProductRestInterface])

resp = client.get("/products/99")
assert resp.status_code == 404

inject_repo(model, fake)

Register a fake repository for a domain model type.

The harness resolves the model from the use-case generic parameter (UseCase[Model, …]) at bootstrap time and injects fake as its repository.

Parameters:

• model (type[Any]) – msgspec.Struct subclass used as the resolution key. Must match the TModel parameter of the use cases that need this repository.

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

Return type:

None

Example:

harness.inject_repo(Product, InMemoryRepository(Product))
harness.inject_repo(Order, FakeOrderRepo())

force_error(model, method, error)

Force a repository method to raise a specific error on every call.

Errors that are LoomError subclasses are mapped to the corresponding HTTP status code by the router.

Parameters:

• model (type[Any]) – Domain model type identifying the repository.

• method (str) – Method name to intercept (e.g. "get_by_id").

• error (Exception) – Exception instance to raise.

Return type:

None

Example:

from loom.core.errors import NotFound
harness.force_error(Product, "get_by_id", NotFound("not found"))

simulate_system_error(model, method)

Simulate an unhandled SystemError on a repository method.

The resulting HTTP response will have status 500.

Parameters:

• model (type[Any]) – Domain model type identifying the repository.

• method (str) – Method name to intercept.

Return type:

None

build_app(interfaces, **fastapi_kwargs)

Build a TestClient from REST interface declarations.

Extracts use cases from the interface route declarations, bootstraps the application with the injected fake repositories, and returns a TestClient ready for HTTP assertions.

The client is configured with raise_server_exceptions=False so that unhandled server errors are returned as 500 responses instead of being re-raised in the test.

Parameters:

• interfaces (Sequence[type[RestInterface[Any]]]) – RestInterface subclasses whose routes should be exposed. Use cases are derived automatically from their routes declarations.

• **fastapi_kwargs (Any) – Forwarded to the FastAPI constructor (e.g. title, version).

Returns:

TestClient wrapping the test app.

Return type:

fastapi.testclient.TestClient

Example:

client = harness.build_app(interfaces=[ProductRestInterface])
resp = client.post("/products/", json={"name": "Widget"})
assert resp.status_code == 201

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.RepositoryIntegrationHarness(*, session_manager, entities, load_order=())

Bases: object

Test harness for integration tests over repository implementations.

Parameters:

• session_manager (SessionManager)

• entities (Mapping[str, EntityHarness])

• load_order (tuple[str, ...])

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_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.

• 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.build_repository_harness(*, session_manager, models, repositories=None, load_order=())

Build a repository integration harness with generic/default repositories.

Parameters:

• session_manager (SessionManager) – Shared SQLAlchemy session manager.

• models (Mapping[str, type[Any]]) – Mapping entity_key -> model class.

• repositories (Mapping[str, Any] | None) – Optional mapping entity_key -> repository instance.

• load_order (tuple[str, ...]) – Optional seed load order.

Returns:

Configured repository integration harness.

Return type:

RepositoryIntegrationHarness

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"