Wiring Great Expectations Checkpoints into pytest

9 min read

The obvious way to run a Great Expectations checkpoint from pytest is one test that calls checkpoint.run() and asserts result.success. It works, and it throws away almost everything the checkpoint produced: a suite of forty expectations collapses to a single boolean, and the failure message names the checkpoint rather than the expectation that failed. This guide sits within Great Expectations spatial expectations and covers wiring the two tools together so the checkpoint’s detail survives into pytest’s output.

The technique is to run the checkpoint once per session and parametrise pytest over its individual results. Each expectation becomes a named test, and the failure that reaches CI says which expectation failed, on how many rows, and which rows they were.

Root cause: one assertion discards a structured report

A validation result is a tree — checkpoint, then validation, then one result per expectation, each carrying a success flag, an unexpected count, and a sample of offending rows. Asserting on the root collapses that tree to a leaf.

One assertion versus one test per expectation Two ways of surfacing a Great Expectations checkpoint result inside pytest are compared. In the first, a single test asserts on the checkpoint's overall success flag; its failure message names only the checkpoint, does not say how many of the expectations failed, provides no offending row identifiers, and stops at the first problem so the rest of the report is never seen. In the second, the checkpoint runs once and pytest is parametrised over the individual expectation results, so each expectation becomes a separately named test whose failure names that expectation, reports its unexpected row count, carries a sample of example row identifiers, and leaves every unrelated expectation free to report its own verdict independently. assert result.success 1 test for 40 expectations failure names the checkpoint only no unexpected count no offending row identifiers one failure hides the other 39 the report was produced, then discarded parametrise over results 40 named tests, checkpoint runs once failure names the expectation unexpected count in the message example row ids in the message every expectation reports independently the report becomes the test output

Step-by-step implementation

Step 1 — Run the checkpoint once, in a session fixture

The checkpoint is expensive — it reads the layer and evaluates every expectation — so it must not run once per test. A session-scoped fixture runs it exactly once and hands the results to everything that follows:

import pytest
import great_expectations as gx


@pytest.fixture(scope="session")
def validation_results(gpkg_path):
    """Run the spatial checkpoint once; return its per-expectation results."""
    context = gx.get_context()
    checkpoint = context.get_checkpoint("parcels_checkpoint")
    result = checkpoint.run(
        batch_request={"path": str(gpkg_path), "layer": "parcels"},
    )
    # A checkpoint may validate several batches; flatten to one list.
    flattened = []
    for validation in result.run_results.values():
        flattened.extend(validation["validation_result"].results)
    return flattened

Returning the flattened list rather than the raw result object is what makes the next step possible, and it keeps the checkpoint’s internal structure — which changes between Great Expectations releases — confined to one function.

Step 2 — Turn each result into a named test

pytest’s parametrisation happens at collection time, before fixtures run, so the results cannot be parametrised over directly. pytest_generate_tests bridges the gap by running the checkpoint at collection:

# conftest.py
import great_expectations as gx

_RESULTS_CACHE = {}


def _results(gpkg_path):
    if gpkg_path not in _RESULTS_CACHE:
        context = gx.get_context()
        checkpoint = context.get_checkpoint("parcels_checkpoint")
        out = checkpoint.run(batch_request={"path": str(gpkg_path),
                                            "layer": "parcels"})
        flat = []
        for validation in out.run_results.values():
            flat.extend(validation["validation_result"].results)
        _RESULTS_CACHE[gpkg_path] = flat
    return _RESULTS_CACHE[gpkg_path]


def pytest_generate_tests(metafunc):
    if "expectation_result" not in metafunc.fixturenames:
        return
    path = metafunc.config.getoption("--gpkg")
    results = _results(path)
    metafunc.parametrize(
        "expectation_result",
        results,
        ids=[_test_id(r) for r in results],
    )


def _test_id(result) -> str:
    cfg = result.expectation_config
    column = cfg.kwargs.get("column")
    name = cfg.type.removeprefix("expect_")
    return f"{column}-{name}" if column else name

The ids argument is doing the real work here. Without it, pytest labels the tests expectation_result0 through expectation_result39, which is no better than the single assertion. With it, a failing run reports test_expectation[geometry-column_geometries_to_be_valid], which names the problem in the test identifier itself — visible in CI summaries, in test-history dashboards, and in the flake-tracking described in detecting flaky spatial tests.

Step 3 — Write one assertion with a message worth reading

def test_expectation(expectation_result, layer):
    r = expectation_result
    if r.success:
        return

    cfg = r.expectation_config
    unexpected = r.result.get("unexpected_count", "?")
    total = r.result.get("element_count", "?")
    sample_idx = (r.result.get("partial_unexpected_index_list") or [])[:5]
    sample_ids = list(layer.loc[sample_idx, "id"]) if sample_idx else []

    pytest.fail(
        f"{cfg.type} failed on column {cfg.kwargs.get('column')!r}: "
        f"{unexpected}/{total} rows unexpected"
        + (f"; example ids: {sample_ids}" if sample_ids else "")
        + (f"; kwargs: {cfg.kwargs}" if cfg.kwargs else ""),
        pytrace=False,
    )

pytrace=False suppresses the Python traceback, which for a data failure is pure noise — nobody investigating an invalid-geometry report needs to see the frames inside the assertion helper. What they need is the expectation name, the ratio, and five identifiers, and that is exactly what remains.

The five parts of a useful validation failure message Each component of a well-formed data-validation failure message is listed with what it contributes to the reader. The expectation name identifies which rule was broken. The column name says where in the layer the problem is. The unexpected count relative to the total element count says how severe the problem is and whether it is systemic or isolated. A short list of example row identifiers gives the reader a concrete place to begin investigating. The expectation's keyword arguments record the threshold that was actually applied, letting a reader judge whether the data is wrong or the rule has become too strict. A Python traceback contributes nothing to any of these questions and is deliberately suppressed. MESSAGE PART QUESTION IT ANSWERS expect_column_geometries_to_be_valid which rule was broken column 'geometry' where in the layer 3412/184092 rows unexpected how bad — systemic or isolated example ids: [8812, 8814, 9001, ...] where to start looking kwargs: {'mostly': 0.99} was the rule or the data wrong a traceback answers none of these — pass pytrace=False

Step 4 — Keep the validation report as an artefact

pytest output is ephemeral; the checkpoint’s own Data Docs are not, and they are the thing a data owner will actually open. Write them from a session-scoped finaliser and attach the path:

@pytest.fixture(scope="session", autouse=True)
def build_data_docs(request):
    yield
    context = gx.get_context()
    context.build_data_docs()
    print("\nvalidation report:",
          context.get_docs_sites_urls()[0]["site_url"])

Uploading that directory as a CI artefact turns a red build into something a non-engineer can investigate, which is usually the point of running Great Expectations rather than plain assertions in the first place.

Making the results reviewable over time

A single validation run answers whether the data is acceptable today. What a team usually needs is whether it is getting better or worse, and neither pytest nor a checkpoint answers that on its own — both report a verdict and discard the number behind it.

The missing piece is small: record the unexpected fraction for every expectation on every run, keyed by expectation name and run timestamp, and append it to a file or a metrics backend. A validity rate that has been sitting at 0.4% for six months is a known characteristic of the source; the same 0.4% after three months at 0.05% is an incident, and only the history distinguishes them.

def record_rates(results, sink):
    for r in results:
        total = r.result.get("element_count") or 0
        unexpected = r.result.get("unexpected_count") or 0
        if total:
            sink.write(f"{r.expectation_config.type}\t{unexpected / total:.6f}\n")

Writing the rate even when the expectation passed is the part that matters, and it is the part usually skipped. A number recorded only on failure produces a series with gaps exactly where the good news was, which makes the trend unreadable at precisely the moment somebody asks whether things are improving.

Where the boundary between the two tools falls

Check Belongs to Why
Incoming layer’s schema and value ranges Great Expectations The audience is the data owner
Geometry validity rate in source data Great Expectations A report, not a stack trace
Feature count against the previous run Great Expectations Trend, not boolean
Round-trip parity across formats pytest Your code’s behaviour
CRS transformation accuracy pytest Your code’s behaviour
Topology rules your pipeline produces pytest Your code’s behaviour
The expectations themselves work pytest Custom expectations are code

The last row is easy to overlook and matters: a custom expectation is code with a metric implementation behind it, and it needs unit tests of its own. An expectation that silently returns success for every input passes every checkpoint it appears in, which is the quietest possible failure. Writing custom geometry expectations covers the tests those need.

Running the checkpoint against more than one batch

Most suites start against a single file and eventually need to run against several — one GeoPackage per region, one per delivery date, or one per upstream supplier. The parametrisation above extends to that case naturally, and the extension is worth making early because retrofitting it later means changing every test identifier.

The change is to make the parametrised unit a pair of batch and expectation rather than an expectation alone. The identifier then reads north-geometry-column_geometries_to_be_valid, which distinguishes a supplier whose data has always been poor from one that has just regressed — a distinction the aggregated view actively hides.

Two practical constraints follow. First, the collection-time run now costs the sum of every batch’s validation, so a marker or command-line option gating the whole thing stops being a nicety. Second, the number of generated tests is the product of batches and expectations, which reaches the hundreds quickly; that is fine for pytest, and it will overwhelm a CI summary view unless the identifiers are ordered batch-first so the grouping is readable.

Where batches genuinely differ in contract — one supplier legitimately lacks a column another has — resist the urge to add conditionals inside a shared suite. Separate suites per contract, each with its own name, keeps every expectation meaningful for the data it describes and keeps the failure report free of expectations that were never going to apply.

The checkpoint runs at collection, not at execution pytest's lifecycle is drawn as two phases. During collection, before any fixture has run, the checkpoint executes and its per-expectation results become the parametrisation, so by the time execution begins each expectation already exists as a separately named test. During execution, each of those tests simply reads its own result and either passes or fails with a message naming the expectation. The consequence noted alongside is that even an invocation that only collects tests, or one filtered to unrelated tests, will still execute the checkpoint, which is why it must be gated behind a marker or a command-line option rather than left to run unconditionally. COLLECTION pytest_generate_tests no fixtures have run yet checkpoint runs once, cached by path results become parameters ids name column and rule EXECUTION one test per expectation reads its own result, reports Because the work happens at collection, even pytest --collect-only or a run filtered to unrelated tests will execute the checkpoint. Gate it behind a marker or a command-line option, or every pytest invocation pays for the validation.

Failure modes and edge cases

Collection-time execution slows every pytest invocation. pytest_generate_tests runs during collection, so even pytest --collect-only or a run filtered to unrelated tests will execute the checkpoint. Guard it on a marker or a command-line option so the spatial validation only runs when it is being asked for.

The checkpoint result structure changes between releases. Great Expectations reorganised its result objects substantially at 0.16 and again at 0.18. Confine access to one adapter function, as _results does, so a version bump is one edit rather than forty.

Skipping is not the same as passing. A checkpoint that finds no batch — wrong path, missing layer — can return success with zero validations. Assert that the flattened result list is non-empty before parametrising, or an entire suite silently evaporates.

Parametrised tests share the fixture, so failures are not independent in cost. If the checkpoint itself raises, every parametrised test errors identically and the report is forty copies of one message. Catch the checkpoint failure at collection and emit a single explicit error instead.

Data Docs write to disk on every run. Under pytest-xdist, several workers will build docs concurrently into the same directory. Build them from the controlling process only — gate on PYTEST_XDIST_WORKER being unset — or the last writer wins non-deterministically.

Conclusion

Run the checkpoint once at collection, parametrise pytest over the individual expectation results with identifiers that name column and rule, and make the failure message carry the unexpected ratio, example identifiers, and the kwargs that were applied. Keep the Data Docs as a build artefact for the audience who will not read CI logs, and keep the two tools’ responsibilities separate: data validation reports on what arrived, pytest asserts on what your code did.