git:20260929.8d5c111 to git:20260929.449c56a

---

name: flat-test-persistence

description: Use when testing a flat-layered service's data-access package against the real datastore — its tables and the repository class that owns its transactions — pinning the generated constraint name where the translator branches on it, the catalogue exception a driver error is translated into on each write and read the class has, what each read returns, and each write's conditional contract where it has one — an upsert's update set from both sides, an ordering-stamp guard from both directions, an empty update set, a batch's chunk edge and same-key inputs, a paged read's tied edge, a multi-statement write's atomicity. Consumes the container and isolation fixtures rather than laying them (`flat-test-integration-setup`). Not the function a trigger calls that reaches this write path, which is `flat-test-run-function`, and not a hexagonal `IFooRepository` adapter, which is `hex-test-repository-contract`, in the `pyhouse-hex` plugin.

---

# Flat Test — Data-Access Contract

Consult `test-principles` for the testing constitution. Where this skill contradicts `test-principles`,

the constitution wins.

One integration-test file per repository class, under the distribution's own `tests/integration/`,

driven against the **real** datastore from `flat-test-integration-setup`. This is the only level that

can catch what the data-access package exists to guarantee: that a conflict updates the columns it

claims to, that a read returns what the store holds, and that a driver error arrives as the service's own

exception.

Where several distributions share one data-access library, the files sit with that library's own tests

instead — `myschema/tests/integration/` — and nothing else changes.

**The isolation fixture follows the declared transaction owner** (`test-principles` reliability

fixture handed to the repository class's constructor, with `truncate_all` cleaning up, for one that

opens its own — asserting through a fresh connection only where its own rollback is under test

(rule 10).

## When to use vs. neighbours

- A pure function in the data-access package — a connection-string builder, the row mapper → not this

skill; it is a unit test with no fixtures at all.

- The container, migration and isolation fixtures themselves → `flat-test-integration-setup`; this skill

- Writing the tables, repository class and migrations under test → `flat-persistence`.

- The work a trigger calls reaching this write path as part of a run → `flat-test-run-function`; this skill tests

the write path, that one tests the wiring above it.

- The client feeding these rows → `flat-test-service-client`.

- A static "no package outside the data-access package constructs a statement" rule →

`test-architecture-rule`.

- The shared groundwork — fixture placement, assertion strength, reliability → `test-principles`.

- The adapter sits behind a domain repository protocol → `hex-test-repository-contract`, in the

`pyhouse-hex` plugin.

## Template — the repository class's contract (pytest, SQLAlchemy async over Postgres)

`tests/integration/test_foo_repository.py`:

```python

from datetime import UTC, datetime

import pytest

from sqlalchemy import select

from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine

from myapp.exceptions import StorageWriteRejectedError

from myapp.postgres import FooRepository

from myapp.postgres.foo_table import foo_table

from myapp.schemas import Foo, FooExternalId

_EARLIER = datetime(2024, 1, 1, tzinfo=UTC)

_LATER = datetime(2024, 1, 2, tzinfo=UTC)

def _a_foo(external_id: str = "alpha", name: str = "first", as_of: datetime = _EARLIER) -> Foo:

return Foo(external_id=FooExternalId(external_id), name=name, as_of=as_of)

# only where a write updates on conflict

async def test_a_second_write_of_one_external_id_updates_the_set_and_keeps_the_rest(

engine: AsyncEngine,

) -> None:

repository = FooRepository(engine)

await repository.upsert(_a_foo(name="first"))

await repository.upsert(_a_foo(name="second", as_of=_LATER))

assert [tuple(row) for row in rows] == [(first_id, "second", _LATER)]

# only where writes for one key can arrive out of order

repository = FooRepository(engine)

await repository.upsert(_a_foo(name="newer", as_of=_LATER))

await repository.upsert(_a_foo(name="older", as_of=_EARLIER))

assert [tuple(row) for row in rows] == [("newer", _LATER)]

async def test_a_value_the_store_refuses_arrives_as_the_catalogue_error(engine: AsyncEngine) -> None:

with pytest.raises(StorageWriteRejectedError) as exc_info:

await FooRepository(engine).upsert(_a_foo(name="nul\x00"))

assert exc_info.value.context == {"sqlstate": "22021", "constraint": None}

```

The first test pins the declared update set from both sides in one comparison: one row, not two; the

name and stamp the set covers changed; the id it does not cover kept the value the first write minted.

The second write carries the later stamp, so the test holds with an ordering guard on the write and

without one.

The translation test forces the write through the public method a caller uses, refused by the store

itself — a NUL character, which a Postgres text value cannot hold, is SQLSTATE `22021` in the

data-exception class — so no constraint had to be invented for the test.

Where a write resolves a conflict by doing nothing, the file gains the no-op test of rule 5; where it

spans statements, the atomicity test of rule 10; where a method takes a batch, the two tests of rule 6; and a test of what

each read returns, rule 14, wherever the class reads.

## Other bindings

- **The ORM in place of Core.** Mapped classes replace the table objects and the session replaces the

connection; every rule below is unchanged, because each pins a property of the stored data rather than

of the statement that wrote it. Assertions still read the store back (rule 1) — through a fresh query,

not through the session's identity map, which can hand back the object just written without it ever

having reached the datastore.

- **Another engine or dialect.** The upsert spelling, the generated constraint names and the driver

errors a violation raises change together; the contract questions — idempotence, the update set from

both sides, the translated exception — do not. Where the dialect has no single-statement upsert, the

repository's *contract* is still the subject and the test pins whatever its write does instead.

- **How the datastore is provisioned** — a disposable container, a pre-provisioned throwaway database, an

in-process engine. That choice belongs to `flat-test-integration-setup`, which names its trade-offs;

nothing in these files changes with it except that an in-process engine of a different dialect

invalidates rules 2 and 3.

## Rules

1. **This level runs against the real store, substitutes nothing below the repository class, and

asserts through a query against the store** — `test-principles`, *Datastore contract* rule 1.

2. **Where the translator branches on a constraint's name, pin that name** — `test-principles`,

*Datastore contract* rule 2.

3. **A unique constraint the translator branches on is tested on the plain insert *and*, where the

class has one, on the conflict-resolution path** — `test-principles`, *Datastore contract* rule 3.

4. **Where a write updates on conflict, test its declared update set from both sides.** The second write must change what the set names

*and* leave every column outside it alone — that is the whole contract of an upsert, and both are

pinned, whether in one comparison or two.

5. **An empty update set means "do nothing on conflict".** Where a write declares one, test it as a

no-op that raises nothing and changes nothing; a write that meets a key already recorded and must

leave it alone relies on exactly that.

6. **Where a method takes a batch, it carries the two tests `test-principles`, *Datastore contract*

rule 7, states** — a chunk edge crossed and two inputs sharing one key in one call.

7. **A forced driver error asserts the catalogue exception the data-access package produces, on each

write and read the class has** — `test-principles`, *Datastore contract* rule 4. A read

with no input the store can refuse takes reliability rule 6's one exception.

8. **Where the store assigns a timestamp, assert it as `test-principles` reliability rule 4 states.**

9. **Ordering is asserted only where the query guarantees it.** Add an explicit order clause to any

query whose result is compared to a list; a relational store promises no insertion order, and a test

that passes on two rows fails on two hundred.

10. **Where one write spans statements, it has an atomicity test** that forces its last statement to fail

through a constraint the schema declares, expects the narrowest catalogue exception that constraint

produces, and asserts through a fresh connection that the earlier statements left nothing behind —

a refactor splitting the statements into two transactions passes every test.

11. **Where a run walks a table in pages, a test crosses a page edge where the ordering column ties.**

12. **A test arranges every row it asserts on as `test-principles` *Datastore contract* rule 5 states** —

the service's declared type written through the repository class, or through the schema's own table

object where the class has no write for it — and, where the subject opens its own connection,

committed before it runs (`flat-test-integration-setup`).

13. **Where writes for one key can arrive out of order, the newer record is written and then the older**

— `test-principles`, *Datastore contract* rule 6; the builder takes the stamp, so each write carries

its own.

14. **Each read the class has is tested on what it returns** — `test-principles`, *Datastore contract*

rule 9 — its rows arranged as rule 12 states.

## Hard stops

- The constraint under test does not exist in a migration yet → stop, add the revision first

(`flat-persistence`); a test asserting a constraint the schema never had passes for the wrong reason.

- The adapter sits behind a domain repository protocol → stop, use `hex-test-repository-contract`, in

the `pyhouse-hex` plugin.