loom.testing

Functions

__getattr__(name)

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

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

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

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