Adapter conformance🔗
pydapper.testing.adapter_conformance is a reusable, typed conformance suite for adapter command classes. It
ships inside the installed pydapper distribution, so first-party and third-party adapter authors run the same
contract without copying tests out of the pydapper repository. The helper uses only the Python standard library
and pydapper core: it does not require pytest or any other test framework, it never imports an optional database
driver, and importing it starts no service, container, or emulator. It is not exported from the pydapper
package root; import it explicitly:
from pydapper.testing.adapter_conformance import run_core_sync
Mandatory profiles: core-sync and core-async🔗
core-sync and core-async are the stable identifiers of the two mandatory profiles.
- Every registration that provides a sync command class promises
core-sync. - Every registration that provides an async command class promises
core-async.
The two profiles are independent. An adapter registered for both modes must pass both; passing sync conformance
says nothing about async conformance, and a sync-only harness is never asked for async fields (or vice versa).
Mode support is represented by which command classes a registration supplies — these mandatory profiles are
not AdapterCapability flags.
Each profile covers registration and selection, the #541 cursor lifecycle (cleanup on success and active
failure, error precedence, truthy-exit non-suppression, plain and context-manager cursors, unbuffered
exhaustion and explicit close()/aclose()), parameter handling and execution, row mapping and RawRow
mappers, zero/one/many cardinality, scalar semantics (no row vs SQL NULL vs falsey values), command options,
preparation-hook ordering, and capability honesty. Cases are either live (they run through your real
database resources) or instrumented (they wrap your real concrete command class around framework-owned
recording and fault-injection connections, so cursor counts, call order, exception precedence, and
"fails before driver work" guarantees are observed directly rather than inferred from return values).
Optional capability profiles🔗
Optional behaviors are independent profiles keyed to pydapper.AdapterCapability members — there is no linear
"better adapter" tier, and no adapter is required to support any optional capability. The production catalog is
returned by capability_profiles() and currently ships one profile: transactions, covering the sync
transaction APIs. The runners automatically append the profile cases for every capability the
command class under test declares, so a declaring adapter is exercised against its capability profiles in the
same run_core_sync() / run_core_async() call as the core inventory — there is no separate capability runner
to invoke. The framework enforces honesty in both directions:
- A command class that declares a valid capability with no populated profile for its mode fails conformance
(
capabilities.declared-profile-populated). An unimplemented capability can never appear covered; zero-case profiles are rejected outright. - A declaration that is not a
frozensetofAdapterCapabilitymembers fails conformance (capabilities.declaration-valid). supports()must exactly match the class declaration (capabilities.supports-matches-declaration), and valid non-defaultCommandOptionsfor undeclared capabilities must raiseUnsupportedFeatureErrorbefore cursor acquisition or driver work (options.unsupported-raises,options.unsupported-before-driver-work).
Native database support alone is never enough to declare a capability: a capability describes behavior the command class actually implements, tests through its conformance profile, and documents. Until then, the feature must fail loudly rather than silently execute as if supported.
The transactions profile🔗
Nine sync cases pin the sync commit() / rollback() / transaction() contract for every declaring adapter.
Every live case first commits the harness baseline (create_commands() then commit()), so scenarios start
from a durable, transaction-free state even when a harness seeds inside an open transaction. The profile has no
async cases yet — the async transaction APIs are a separate feature — so an async command class may not declare
TRANSACTIONS until they land.
| Case id | Kind | Pins |
|---|---|---|
transactions.commit-persists |
live | commit() makes a write durable; a later rollback() cannot remove it |
transactions.rollback-discards |
live | rollback() discards uncommitted work, committed rows stay intact |
transactions.context-commits-on-exit |
live | a transaction() block commits on clean exit |
transactions.context-rolls-back-on-error |
live | a raising block rolls back and re-raises the same exception object |
transactions.commit-delegates-once |
instrumented | one connection commit, no cursor acquired, returns None, failures propagate unchanged |
transactions.rollback-delegates-once |
instrumented | one connection rollback, no cursor acquired, returns None, failures propagate unchanged |
transactions.context-lifecycle-order |
instrumented | the block's commit comes after the command's full cursor lifecycle |
transactions.context-error-precedence |
instrumented | rollback on error, and the block's error wins over a rollback failure |
transactions.commit-failure-propagates |
instrumented | a failed exit commit propagates unchanged with no rollback attempt |
The harness API🔗
A harness supplies factories, isolation, and narrow dialect knobs. It never owns assertions: every behavioral check belongs to the framework runner, and harness- or adapter-provided code has no API to declare a case passed.
Sync adapters subclass SyncAdapterHarness; async adapters subclass AsyncAdapterHarness.
Every configuration field below may be declared on the subclass or assigned per instance — none of them is a
ClassVar, so assigning to self in __init__ type checks without a type: ignore. That matters for the
common case of a value that does not exist until a fixture has run, such as a DSN that is only known once a test
container has started:
class MyHarness(SyncAdapterHarness):
adapter_name = "acmedb"
command_class = AcmeDbCommands
def __init__(self, container):
self.connect_dsn = f"acmedb://{container.host}:{container.port}/conformance"
A field that is genuinely static (table_name, column_case, strict_rowcounts, …) is still clearest as a
class attribute; both forms are supported and mix freely on one harness.
| Field | Required | Meaning |
|---|---|---|
adapter_name |
yes | The registered adapter name (exact, case-sensitive). |
command_class |
yes | The declared concrete Commands / CommandsAsync subclass for the mode. |
connect_dsn |
yes | A DSN that safely reaches an isolated test database through connect() / connect_async(). |
create_commands() |
yes | Return a fresh command instance over a fresh connection with the canonical dataset freshly seeded. Called once per case. |
teardown_commands(commands) |
yes | Release everything create_commands() built. Runs after success and failure. |
connect_kwargs |
no | Extra kwargs for the connect() path (for example BigQuery's client=). |
table_name |
no | Name (optionally schema/dataset qualified) of the conformance table. Default pydapper_conformance. |
column_case |
no | "lower" (default) or "upper" for dialects that fold unquoted identifiers (Oracle). |
supports_empty_strings |
no | False only when the database cannot store '' distinctly from NULL (Oracle); the empty-string dataset value then seeds as the documented fallback "blank". |
strict_rowcounts |
no | False only when DML rowcounts are documented as unreliable (BigQuery emulator). Rowcount cases still run and must return an int; only exact equality is relaxed. |
sql_overrides |
no | Statement-id → SQL overrides for the narrow cases where portable SQL is insufficient (still pydapper ?name? placeholders). |
recover_after_error(commands) |
no | Backend recovery needed after an intentionally failed command (default no-op). |
cursor_factory_style |
async only | "awaitable" (default: connection.cursor() returns an awaitable, the CommandsAsync base contract) or "synchronous" (the psycopg3-async shape: cursor() returns the cursor directly and the command class normalizes it). Instrumented cases use this to exercise your real cursor() override. |
The framework owns the canonical dataset and every query it runs. CONFORMANCE_COLUMNS is
("id", "label", "score", "note") and seed_rows(supports_empty_strings) returns the three canonical rows —
including a SQL NULL note, a falsey 0 score, and an empty-string label — that create_commands() must
seed. All query and DML statements are written in pydapper's portable ?name? placeholder syntax, so they run
unchanged on every adapter; the harness owns only DDL, seeding, and connection construction.
Every value the framework itself binds as a parameter is non-NULL. Some drivers — BigQuery's DBAPI, for
example — derive a parameter's type from its Python value and reject an untyped None, so binding NULL is
not part of the mandatory core profiles. The canonical dataset still contains a real SQL NULL note and
scalar.null-returns-none still asserts that a stored NULL reads back as None; producing that NULL is
the harness's job, and a harness whose driver cannot bind None may seed it any way its dialect allows (the
BigQuery harness binds a sentinel and then clears it with a literal UPDATE ... SET note = NULL).
A harness that does not supply a field a case needs fails that case with a structured
HarnessDefinitionError carrying the profile id, the case id, and the missing field name — a missing harness
field never silently skips or weakens a core case. There is deliberately no "skip this core case" option;
driver limitations are expressed only through the narrow, documented knobs above.
Where to seed: under the adapter or through it🔗
create_commands() must return a freshly seeded dataset, and there are two reasonable ways to produce it. Both
are used in this repository, so it is worth choosing deliberately:
- Driver-level (recommended default). Insert the rows with the raw driver, underneath the adapter — the
connection.executemany(...)in both examples below, and what pydapper's own sqlite3 harness does. Setup then shares no code path with the thing under test, so a broken seed cannot be confused with broken adapter behavior, and reading the harness tells you exactly what is in the table. - Through the adapter. Insert the rows with
commands.execute(...)and pydapper's portable?name?placeholders, as pydapper's service-backed harnesses do through the sharedtests/conformance_support.py::seed_through_adapterhelper. One seeding statement then works unchanged across every dialect, which is why harnesses for many backends share it; the cost is that setup runs through the adapter it is about to judge.
Prefer driver-level seeding unless you are maintaining harnesses for several dialects at once and want a single
portable seeding path. This is a readability and blast-radius preference, not a correctness one: failures inside
create_commands() are attributed to the harness either way (see below), so seeding through the adapter no
longer risks a broken seed being reported as adapter misbehavior.
Harness setup failures are attributed to the harness🔗
If the harness's own create_commands() raises — a connection factory that cannot connect, a CREATE TABLE
against the wrong dataset, a seeding statement the driver rejects — the resulting CaseResult is failed with
harness_setup_failed=True and a message that names create_commands() as the culprit, while cause remains
the original exception object. ConformanceReport.harness_setup_failed is True if any case in the run has it,
and ConformanceFailureError's summary gains a [HARNESS SETUP FAILED: n of m failure(s) …] segment, so the
traceback CI prints already says the harness never got off the ground.
This exists because one broken setup call fails every case that needs a fixture: without attribution, a single
bad seeding statement reads as dozens of identical behavioral failures against a perfectly good adapter. A run
with harness_setup_failed set is not a verdict on the adapter — fix the harness and run again.
The boundaries are narrow and deliberate:
- Only the harness's own
create_commands()call is wrapped, never the framework's validation or bookkeeping around it, and never the adapter behavior a case exercises afterwards. Genuine behavioral failures keep their existing shape and reportharness_setup_failed=False. KeyboardInterrupt,SystemExit, andasyncio.CancelledErrorstill propagate unconverted; cancellation and shutdown are never turned into case results.- A harness that omits a required field or override is still a
HarnessDefinitionErrorwith itsmissing_field— a missing override is a definition error, not a setup failure, and reporting for it is unchanged. - The internal exception type used to carry this across the runner boundary is private. Branch on the two
structured attributes (
CaseResult.harness_setup_failed,ConformanceReport.harness_setup_failed), not on an exception class and not on message text.
Case isolation and cleanup🔗
Every case gets a fresh command instance, connection, and dataset through create_commands(); case ordering
and earlier failures cannot make later cases pass. teardown_commands() runs after every case, on success and
on failure. If both a case and its teardown fail, the case failure is preserved and the teardown failure is
retained as structured secondary information on the result (CaseResult.cleanup_error); a teardown failure
after an otherwise passing case fails that case.
Deterministic, structured results🔗
run_core_sync() / run_core_async() return a ConformanceReport whose results follow the declared
profile/case order exactly, run after run. Each CaseResult carries the profile id, case id, pass/fail, a
human-readable message, the original cause exception when one exists, the missing_field for harness
validation failures, any cleanup_error, and harness_setup_failed. report.raise_for_failures() raises a
ConformanceFailureError that exposes the same structured failures — nothing requires parsing exception
prose.
Two flags that make a run mean less than it looks🔗
report.passed answers "did every case that ran pass?" and nothing more. Two separate booleans say whether the
run was a conformance run at all, and both are structured attributes so no caller has to read messages:
| Flag | True means |
What it invalidates |
|---|---|---|
report.covers_full_inventory |
every planned case ran (the default; False only after a case_ids filter) |
a passing run that covered a subset is not conformance |
report.harness_setup_failed |
at least one case failed inside the harness's own create_commands() |
the failures describe your harness, not the adapter |
Anything that claims conformance — a CI gate, a published matrix row, a release note — must therefore assert
report.passed and report.covers_full_inventory. raise_for_failures() deliberately ignores completeness: it
is about failures only, so a filtered run that passes everything it ran raises nothing.
report = run_core_sync(MyAdapterHarness())
assert report.passed and report.covers_full_inventory
Running one adapter🔗
from pydapper.testing.adapter_conformance import run_core_sync
report = run_core_sync(MyAdapterHarness())
for failure in report.failures:
print(failure.profile_id, failure.case_id, failure.message)
report.raise_for_failures()
run_core_async is a plain awaitable and is awaited inside your own event loop — the framework never calls
asyncio.run() itself:
report = await run_core_async(MyAsyncAdapterHarness())
Debugging one case with case_ids🔗
Both runners accept an optional case_ids collection that narrows the run to the named cases. It exists purely
to shorten the debug loop: against a container-backed backend, a full profile costs minutes per iteration —
mostly the client's connection retry backoff — which is a miserable way to chase one failing case.
report = run_core_sync(MyAdapterHarness(), case_ids=["rows.query-buffered", "scalar.null-returns-none"])
assert report.covers_full_inventory is False # by construction: this is a debug run, not a conformance run
The rules are chosen so a filtered run can never be mistaken for a real one:
- A filtered run is never a conformance result.
covers_full_inventoryisFalsewhenever a filter narrowed the run, andConformanceFailureError's summary gains a[PARTIAL RUN: …]segment. Naming every planned case is still full coverage, andcase_ids=None(the default) runs the full inventory exactly as before. - Unknown ids fail loudly, before anything runs. A typo raises
CaseSelectionError— carryingprofile_id,requested_case_ids, andunknown_case_ids— rather than quietly running fewer cases. An empty selection raises the same error with an emptyunknown_case_ids, because "no cases ran and none failed" is not a pass. - A bare string is a
TypeError.case_ids="rows.query-buffered"would otherwise be a collection of characters; pass a list. - Ids come from
core_sync_profile().sync_cases/core_async_profile().async_cases(plus any declared-capability profile cases, which are planned before filtering and so can also be selected). Duplicates are de-duplicated, and results still follow declared case order, not the order you requested.
Complete third-party sync example🔗
This one computes connect_dsn per instance in __init__ — the shape a harness needs when the DSN only exists
after a fixture has run — and seeds at the driver level.
"""A complete third-party sync-only adapter running the reusable conformance suite.
The "acmedb" driver below is stdlib sqlite3 wearing a third-party hat: the point is
the shape of the harness, not the database. No pytest is required.
"""
import sqlite3
import tempfile
import pydapper
from pydapper.commands import BaseSqlParamHandler
from pydapper.commands import Commands
from pydapper.testing.adapter_conformance import SyncAdapterHarness
from pydapper.testing.adapter_conformance import run_core_sync
from pydapper.testing.adapter_conformance import seed_rows
class AcmeDbCommands(Commands):
class SqlParamHandler(BaseSqlParamHandler):
def get_param_placeholder(self, param_name: str) -> str:
return "?"
@classmethod
def connect(cls, parsed_dsn, **connect_kwargs) -> "AcmeDbCommands":
return cls(sqlite3.connect(parsed_dsn.database, **connect_kwargs))
pydapper.register_adapter(
"acmedb",
commands=AcmeDbCommands,
using_connection_predicate=lambda connection: isinstance(connection, sqlite3.Connection),
)
class AcmeDbConformanceHarness(SyncAdapterHarness):
adapter_name = "acmedb"
command_class = AcmeDbCommands
def __init__(self) -> None:
# no harness field is a ClassVar, so configuration a real harness only learns at
# runtime -- a container's host and port, a temp directory -- is assigned on self
workdir = tempfile.mkdtemp()
# four slashes make the sqlite path absolute; the +acmedb suffix selects this adapter
self.connect_dsn = f"sqlite+acmedb:///{workdir}/example.db"
def create_commands(self) -> Commands:
# a fresh, isolated database and dataset for every conformance case
connection = sqlite3.connect(":memory:")
connection.execute(
"CREATE TABLE pydapper_conformance (id INTEGER PRIMARY KEY, label TEXT, score INTEGER, note TEXT)"
)
connection.executemany(
"INSERT INTO pydapper_conformance (id, label, score, note) VALUES (?, ?, ?, ?)",
seed_rows(self.supports_empty_strings),
)
connection.commit()
return AcmeDbCommands(connection)
def teardown_commands(self, commands: Commands) -> None:
commands.connection.close()
report = run_core_sync(AcmeDbConformanceHarness())
for failure in report.failures:
print(f"{failure.profile_id}/{failure.case_id}: {failure.message}")
print(f"{report.profile_id} for {report.adapter_name!r}: {len(report.results)} cases, passed={report.passed}")
report.raise_for_failures()
# claiming conformance takes both: every case passed *and* every planned case ran
assert report.passed and report.covers_full_inventory
# core-sync for 'acmedb': 54 cases, passed=True
Complete third-party async example🔗
This one declares connect_dsn on the class instead, which is the simpler form when the value is known up
front. Both forms are supported.
"""A complete third-party async-only adapter running the reusable conformance suite.
The "acmeaio" driver below is a tiny async facade over stdlib sqlite3, standing in
for a real async driver. The conformance API itself never calls ``asyncio.run()`` —
``run_core_async`` is awaited inside whatever event loop the caller owns, which here
is the one this script starts at the bottom.
"""
import asyncio
import sqlite3
import tempfile
import pydapper
from pydapper.commands import BaseSqlParamHandler
from pydapper.commands import CommandsAsync
from pydapper.testing.adapter_conformance import AsyncAdapterHarness
from pydapper.testing.adapter_conformance import run_core_async
from pydapper.testing.adapter_conformance import seed_rows
WORKDIR = tempfile.mkdtemp()
class AcmeAioCursor:
"""An async cursor facade over a sqlite3 cursor."""
def __init__(self, cursor: sqlite3.Cursor) -> None:
self._cursor = cursor
@property
def description(self):
return self._cursor.description
@property
def rowcount(self) -> int:
return self._cursor.rowcount
async def execute(self, sql, parameters=None) -> None:
if parameters is None:
self._cursor.execute(sql)
else:
self._cursor.execute(sql, parameters)
async def executemany(self, sql, seq_of_parameters=None) -> None:
self._cursor.executemany(sql, seq_of_parameters or [])
async def fetchone(self):
return self._cursor.fetchone()
async def fetchall(self):
return self._cursor.fetchall()
async def fetchmany(self, size=1):
return self._cursor.fetchmany(size)
async def close(self) -> None:
self._cursor.close()
class AcmeAioConnection:
"""An async connection facade whose cursor() returns an awaitable."""
def __init__(self, connection: sqlite3.Connection) -> None:
self._connection = connection
async def cursor(self, *args, **kwargs) -> AcmeAioCursor:
return AcmeAioCursor(self._connection.cursor())
def close(self) -> None:
self._connection.close()
class AcmeAioCommands(CommandsAsync):
class SqlParamHandler(BaseSqlParamHandler):
def get_param_placeholder(self, param_name: str) -> str:
return "?"
@classmethod
async def connect_async(cls, parsed_dsn, **connect_kwargs) -> "AcmeAioCommands":
return cls(AcmeAioConnection(sqlite3.connect(parsed_dsn.database, **connect_kwargs)))
pydapper.register_adapter(
"acmeaio",
async_commands=AcmeAioCommands,
using_connection_predicate=lambda connection: isinstance(connection, AcmeAioConnection),
)
class AcmeAioConformanceHarness(AsyncAdapterHarness):
adapter_name = "acmeaio"
command_class = AcmeAioCommands
# four slashes make the sqlite path absolute; the +acmeaio suffix selects this adapter
connect_dsn = f"sqlite+acmeaio:///{WORKDIR}/example.db"
# our cursor() returns an awaitable, the CommandsAsync base contract; a driver whose
# cursor() returns the cursor synchronously (like psycopg3 async) declares "synchronous"
cursor_factory_style = "awaitable"
async def create_commands(self) -> CommandsAsync:
connection = sqlite3.connect(":memory:")
connection.execute(
"CREATE TABLE pydapper_conformance (id INTEGER PRIMARY KEY, label TEXT, score INTEGER, note TEXT)"
)
connection.executemany(
"INSERT INTO pydapper_conformance (id, label, score, note) VALUES (?, ?, ?, ?)",
seed_rows(self.supports_empty_strings),
)
connection.commit()
return AcmeAioCommands(AcmeAioConnection(connection))
async def teardown_commands(self, commands: CommandsAsync) -> None:
commands.connection.close()
async def main() -> None:
report = await run_core_async(AcmeAioConformanceHarness())
for failure in report.failures:
print(f"{failure.profile_id}/{failure.case_id}: {failure.message}")
print(f"{report.profile_id} for {report.adapter_name!r}: {len(report.results)} cases, passed={report.passed}")
report.raise_for_failures()
asyncio.run(main())
# core-async for 'acmeaio': 54 cases, passed=True
Reporting service-backed runs you could not execute🔗
Service-free evidence (mock harnesses and instrumented concrete-class cases) and live service-backed evidence are different things, and reports must keep them distinct. A live suite that was skipped, or merely collected, is a test-run status — never a pass. When you cannot run a live suite, report:
- whether the optional database driver was installed;
- whether the service or emulator was reachable;
- whether credentials were required and available; and
- which service-free conformance cases cover the same shared behavior.
First-party driver conformance matrix🔗
One row per first-party command class. "Verification" describes what is actually executed where — service-free coverage runs everywhere, live suites run only under their backend marker when their Docker/testcontainers service or emulator is available.
| Adapter name | Command class | Mode | Declared capabilities | Conformance entry | Verification | Driver limits |
|---|---|---|---|---|---|---|
sqlite3 |
Sqlite3Commands |
sync | transactions |
tests/test_sqlite/test_conformance.py |
live, service-free (runs in the core and sqlite suites) | SQLite |
psycopg2 |
Psycopg2Commands |
sync | transactions |
tests/test_postgresql/test_conformance.py |
service-free instrumented + mock coverage; live suite under the postgresql marker when Docker is available |
PostgreSQL |
psycopg |
Psycopg3Commands |
sync | transactions |
tests/test_postgresql/test_conformance.py |
service-free instrumented + mock coverage; live suite under the postgresql marker when Docker is available |
PostgreSQL |
psycopg |
Psycopg3CommandsAsync |
async | (none) | tests/test_postgresql/test_conformance.py |
service-free instrumented + mock coverage (including its synchronous cursor-factory normalization); live suite under the postgresql marker when Docker is available |
PostgreSQL |
aiopg |
AiopgCommands |
async | (none) | tests/test_postgresql/test_conformance.py |
service-free instrumented + mock coverage; live suite under the postgresql marker when Docker is available |
PostgreSQL; aiopg is autocommit-only and emulates executemany by looping execute; its connection-level commit/rollback raise, so transactions is not declared (the async transaction APIs are a separate feature) |
mysql |
MySqlConnectorPythonCommands |
sync | transactions |
tests/test_mysql/test_conformance.py |
service-free instrumented + mock coverage (including its query_first unread-result drain); live suite under the mysql marker when Docker is available |
MySQL; unread results must be drained before a cursor closes |
pymssql |
PymssqlCommands |
sync | transactions |
tests/test_mssql/test_conformance.py |
service-free instrumented + mock coverage; live suite under the mssql marker when Docker is available |
Microsoft SQL Server |
oracledb |
OracledbCommands |
sync | transactions |
tests/test_oracle/test_conformance.py |
service-free instrumented + mock coverage; live suite under the oracle marker when Docker is available |
Oracle; unquoted identifiers fold to uppercase (column_case="upper"), and '' is stored as NULL (supports_empty_strings=False) |
google |
GoogleBigqueryClientCommands |
sync | (none) | tests/test_bigquery/test_conformance.py |
service-free instrumented + mock coverage; live suite under the bigquery marker when the emulator is available |
Google BigQuery; the emulator's DML rowcounts are unreliable (strict_rowcounts=False), and the DBAPI cannot bind an untyped None, so the harness seeds the canonical SQL NULL note through a sentinel and a literal UPDATE; the DBAPI's commit() is a no-op and there is no rollback(), so transactions is not declared |
Extending the suite in future capability work🔗
Every PR that adds or changes an optional capability must, in the same change:
- update the
AdapterCapabilityenum/declaration when necessary; - add a real, reusable capability profile to the production catalog (zero-case profiles are rejected, so an empty placeholder cannot ship);
- test at least one supported and one unsupported adapter against the new profile;
- update the affected first-party capability declaration;
- update driver limitations and documentation, including the matrix above; and
- update the type tests when public typing changes.
This keeps supports(), the declarations, the profiles, and the documentation honest together: a capability is
declared only when its implementation, its profile, and its documentation all exist.