Testing & conformance kit¶

dna.testing ships the SDK's port-adapter compliance suites — in the spirit of the Python DB-API compliance suite. An adapter author hands a factory for their adapter and gets back the battery of cases every conforming implementation must pass. See Running the conformance kit for a walkthrough.

dna.testing ¶

Public testing kit for SDK port adapters (s-dna-source-conformance-kit).

Ship-with-the-SDK compliance suites, in the spirit of the DB-API compliance suite: an adapter author hands us a factory for their adapter and gets back the battery of cases every conforming implementation must pass.

from dna.testing import source_conformance_suite

async def my_factory():
    src = MySource(...)
    async def cleanup():
        await src.close()
    return src, cleanup

# pytest consumption — one test per case:
import pytest
CASES = source_conformance_suite(my_factory)

@pytest.mark.asyncio
@pytest.mark.parametrize("case", CASES, ids=lambda c: c.name)
async def test_source_conformance(case):
    await case.run()

Division of labor (keep this straight): - kernel.source() boot gate → NAMES ONLY (runtime_checkable Protocols can't check behavior). First line of defense. - THIS kit → BEHAVIOR, capability-aware: it reads the adapter's declared SourceCapabilities and fails when a declared capability isn't honored. The real safety net.

CaseNotApplicable ¶

Bases: SkipTest

Raised by a case when the adapter doesn't declare the capability the case exercises. pytest/unittest report it as a skip.

ConformanceCase `dataclass` ¶

ConformanceCase(
    name: str,
    requires: str,
    factory: SourceFactory,
    capabilities: SourceCapabilities | None,
    _fn: Callable[[_Ctx], Any],
    _predicate: Callable[[_Ctx], bool],
)

One runnable conformance case bound to an adapter factory.

run `async` ¶

run() -> None

Build a fresh source, check applicability, run, always cleanup.

Raises :class:CaseNotApplicable (skip) when the adapter doesn't declare the capability this case exercises; AssertionError when a declared capability isn't honored.

ConformanceReport `dataclass` ¶

ConformanceReport(
    passed: list[str],
    failed: list[tuple[str, str]],
    skipped: list[tuple[str, str]],
)

Outcome of :func:run_source_conformance (non-pytest consumption).

RWConformanceCase `dataclass` ¶

RWConformanceCase(
    name: str, kind: str | None, _fn: Callable[[], None]
)

One runnable reader/writer conformance case.

run ¶

run() -> None

Run the case. Raises AssertionError on a broken invariant, :class:CaseNotApplicable (a skip) when the pair can't be exercised for an honest, documented reason.

RecordSearchCase `dataclass` ¶

RecordSearchCase(
    name: str,
    requires: str,
    factory: ProviderFactory,
    _fn: Callable[[Any], Any],
)

One runnable conformance case bound to a provider factory.

SearchCaseNotApplicable ¶

Bases: SkipTest

Raised when a provider doesn't support an optional surface a case needs (e.g. no delete). pytest/unittest report it as a skip.

CoreSourceStub ¶

Minimal test double that satisfies the kernel.source() boot gate.

Carries the CORE SourcePort surface as harmless no-ops — subclass it and override only what your test exercises:

class _FakeSource(CoreSourceStub):
    async def load_layer(self, scope, layer_id, layer_value, readers=None):
        return my_docs

The boot gate (validate_source_port) requires the core members BY NAME on anything handed to kernel.source(); ad-hoc fakes with a single method no longer pass. Capability-mediated members (list_doc_refs/load_one/query/count) stay absent on purpose: the kernel serves those via fallbacks, and their absence is how a stub declares "not granular, no pushdown".

fixture_docs ¶

fixture_docs() -> list[dict[str, Any]]

The canonical fixture: 1 Genome + 3 Story records.

Writable adapters get these written through their own API; read-only factories must install them in their native storage (YAML files on disk, JSON objects in a bucket, rows in a table, ...).

run_source_conformance `async` ¶

run_source_conformance(
    factory: SourceFactory,
    *,
    capabilities: SourceCapabilities | None = None
) -> ConformanceReport

Run the whole suite programmatically (scripts, CI without pytest).

source_conformance_suite ¶

source_conformance_suite(
    factory: SourceFactory,
    *,
    capabilities: SourceCapabilities | None = None
) -> list[ConformanceCase]

THE public conformance suite for Source adapters.

Args: factory: async zero-arg callable returning (source, cleanup) (cleanup async zero-arg or None). Called once PER CASE — each case runs against a fresh adapter. The factory owns environment setup (temp dirs, DB schemas, kernel wiring) and, for read-only sources, pre-seeding :func:fixture_docs under :data:FIXTURE_SCOPE. capabilities: explicit SourceCapabilities override. Default: read from the built source (declared capabilities(), with the deprecated reflection fallback for legacy adapters).

Returns: list of :class:ConformanceCase — parametrize them in pytest (ids=lambda c: c.name) and await case.run().

default_fixture ¶

default_fixture(kind_port: Any) -> dict[str, Any]

Minimal synthetic raw doc for a Kind, derived from its StorageDescriptor (body_field when declared, content otherwise). Override per-kind via the fixtures argument when a hand-rolled writer needs richer spec.

reader_writer_conformance_suite ¶

reader_writer_conformance_suite(
    kernel_factory: KernelFactory,
    *,
    fixtures: dict[str, dict[str, Any]] | None = None,
    real_bundle_roots: list[Path | str] | None = None
) -> list[RWConformanceCase]

THE public round-trip conformance suite for Reader/Writer pairs.

Args: kernel_factory: zero-arg callable returning a LOADED kernel (Kernel.auto(), or a hand-wired kernel with your extension). Called once — readers/writers are stateless. fixtures: per-kind raw-document overrides for Kinds whose hand-rolled writer needs a richer spec than the synthetic default (see :func:default_fixture). real_bundle_roots: scope directories holding REAL bundles (e.g. scopes/market-integration/.dna/market-demo); every bundle a registered reader detects gains a real_roundtrip case.

Returns: list of :class:RWConformanceCase — parametrize in pytest (ids=lambda c: c.name) and call case.run().

fixture_records ¶

fixture_records() -> list[dict[str, Any]]

Canonical fixture: three Stories with disjoint vocabularies + a Genome.

Each doc's text is a distinct token cloud so the fake hash embedder gives a clean relevance signal: a query drawn from one doc's vocabulary ranks that doc above the others (which share no tokens with the query).

record_search_conformance_suite ¶

record_search_conformance_suite(
    factory: ProviderFactory,
) -> list[RecordSearchCase]

THE public conformance suite for RecordSearchProvider adapters.

Args: factory: async zero-arg callable returning (provider, cleanup). Called once PER CASE (fresh provider each time). The factory wires a kernel with the deterministic FakeEmbeddingProvider so the suite runs fully offline.

Returns: list of :class:RecordSearchCase — parametrize in pytest and await case.run().

run_record_search_conformance `async` ¶

run_record_search_conformance(
    factory: ProviderFactory,
) -> RecordSearchReport