Testing

The contract suites an adapter must pass (pip install 'rhizodat[testing]').

rhizodat.testing.contract

The contract every store adapter must pass, run against the real store.

rhizodat's guarantees hold only if an adapter's Branch and Participant keep their promises: nothing is visible before the decision, finish is idempotent, abort really discards, and a crash at any step of the commit protocol ends all-or-nothing after one recovery run. An adapter author writes a Harness for the store and subclasses AdapterContract::

from rhizodat.testing.contract import AdapterContract

class TestMySqlContract(AdapterContract):
    @pytest.fixture
    def harness(self) -> Iterator[MySqlHarness]:
        h = MySqlHarness(...)
        yield h
        h.cleanup()

The crash tests commit one meta-transaction over the adapter's store and an in-memory two-phase store, crash the coordinator at each step, simulate the process dying (Harness.abandon) and run Recovery once.

Harness

Bases: Protocol

What the contract needs to know about one store kind.

open_store

open_store(name: str) -> Store

A store named name in a place of its own (schema, database, graph, namespace), installed and ready for write. The harness cleans it up.

write

write(branch: Branch, key: str) -> None

Write one record identified by key in the branch's open transaction.

count

count(store: Store, key: str) -> int

Committed records with key, read through native access.

Crash

Bases: BaseException

The coordinator's process dies here (not an Exception: nothing catches it).

MemoryStore

A two-phase store in memory: rows by key, prepared transactions by gid.

AdapterContract

The contract as pytest tests: subclass it and provide the harness fixture.

abandon

abandon(harness: Harness, branch: _Probe) -> None

What a dying process leaves: open transactions die with their connections, prepared ones and durable staged data stay. A harness may define abandon(branch) for a store that needs something else.

rhizodat.testing.parts

The contract a store must keep to hold entity parts, run against the real store.

A store that implements PartStore (and whose branches implement PartBranch) can hold a share of an entity's fields. Subclass PartStoreContract and provide a parts_store fixture: a fresh store of your kind, installed, cleaned up afterwards::

from rhizodat.testing.parts import PartStoreContract

class TestMySqlParts(PartStoreContract):
    @pytest.fixture
    def parts_store(self) -> Iterator[MySqlStore]:
        store = MySqlStore(...)
        yield store
        store.close()

Every write is committed the way the coordinator would: stage, then prepare and finish for a two-phase branch, or apply after the decision for a one-phase one (from the journal payload too, the way recovery does).

PartStoreContract

The parts contract as pytest tests: subclass it and provide parts_store.

part_values

part_values(faker: Faker) -> dict[str, Any]

One version's values, of every type an entity field may have.

commit

commit(
    store: Any,
    write: Callable[[Any], None],
    ts: int,
    *,
    via_journal: bool = False,
) -> uuid.UUID

Run write(branch) and commit it at ts as the coordinator would.

via_journal: a one-phase branch is applied by the participant from its payload, as recovery does after the process that staged it died.