Testing Pipelines¶

PyCharter's test harness substitutes configured batch DAG sources, load destinations, and quarantine destinations with fixture/capture implementations. It reads declared contract files. Retained Python still executes with the process's permissions; connector substitution is not a sandbox for arbitrary Python. Run untrusted code in an independently restricted process or container.

Overview¶

The testing module is at pycharter.pipeline_generator.testing and provides:

Component	Purpose
`MockExtractor`	Yields pre-configured fixture data instead of calling a real source
`MockLoader`	Captures loaded records for inspection instead of writing to a real destination
`PipelineTestHarness`	Wires mock extractor + loader into a pipeline and runs it
`load_fixture`	Loads fixture records from a YAML or JSON file
`assert_records_match`	Asserts loaded records equal expected records
`assert_record_count`	Asserts the number of loaded rows
`assert_fields_present`	Asserts every record contains required fields
`assert_no_field`	Asserts a field was removed by the pipeline
`assert_field_values`	Asserts a field's values across all records
`assert_schema_shape`	Asserts field values have the correct Python types
`validate_pipeline_config`	Validates a pipeline YAML config without running it

Quick start¶

import pytest
from pycharter import Pipeline
from pycharter.pipeline_generator.testing import (
    PipelineTestHarness,
    assert_record_count,
    assert_records_match,
    assert_fields_present,
)

FIXTURE = [
    {"userId": "1", "price": "19.99"},
    {"userId": "2", "price": "5.00"},
]


@pytest.mark.asyncio
async def test_transform_pipeline():
    pipeline = Pipeline.from_dict({
        "name": "orders",
        "steps": [
            {"id": "source", "type": "extract", "source": {
                "type": "file", "path": "unused-input.json", "format": "json",
            }},
            {"id": "normalize", "type": "transform", "input": "source",
             "operations": [
                 {"rename": {"userId": "user_id"}},
                 {"convert": {"price": "float"}},
             ]},
            {"id": "sink", "type": "load", "input": "normalize", "target": {
                "type": "file", "path": "unused-output.json", "format": "json",
            }},
        ],
    })

    harness = PipelineTestHarness(pipeline, fixture_data=FIXTURE)
    result = await harness.run()

    assert result.success, result.errors

    assert_record_count(result, expected=2)
    assert_fields_present(harness.loaded_records, ["user_id", "price"])
    assert_records_match(
        harness.loaded_records,
        [
            {"user_id": "1", "price": 19.99},
            {"user_id": "2", "price": 5.00},
        ],
    )

MockExtractor¶

Implements the Extractor protocol — yields fixture data as async batches.

from pycharter.pipeline_generator.testing import MockExtractor

# Flat list — auto-batched by batch_size (default 1000)
extractor = MockExtractor(data=[{"id": 1}, {"id": 2}])

# Pre-batched (two batches of two)
extractor = MockExtractor(data=[[{"id": 1}, {"id": 2}], [{"id": 3}, {"id": 4}]])

# Small batch size for testing batching logic
extractor = MockExtractor(data=list(range(10)), batch_size=3)

To supply your own implementations, pass a run-local PipelineIO mapping. Every configured extract/load/quarantine connector needs an entry; missing entries fail without falling back to configured I/O. For a DAG with extract step source and load step sink, without a quarantine target:

from pycharter.pipeline_generator.io import PipelineIO
from pycharter.pipeline_generator.testing import MockExtractor, MockLoader

loader = MockLoader()
result = await pipeline.arun(io=PipelineIO(
    extractors={"source": MockExtractor(fixture_data)},
    loaders={"sink": loader},
))

MockLoader¶

Implements the Loader protocol — captures records instead of writing to a destination.

from pycharter.pipeline_generator.testing import MockLoader

loader = MockLoader()

# After running:
print(loader.loaded_records)  # All records across all batches (flat)
print(loader.load_calls)  # List of per-batch record lists

loader.reset()  # Clear for the next run

Simulating a load failure¶

loader = MockLoader(simulate_failure=True, failure_error="DB connection refused")

PipelineTestHarness¶

Supplies independent connectors to one run without changing the pipeline or patching factories. Separate harnesses can run concurrently on the same pipeline; a later ordinary pipeline.arun() uses the configured connectors.

from pycharter.pipeline_generator.testing import PipelineTestHarness

harness = PipelineTestHarness(
    pipeline,
    fixture_data=[{"id": 1, "name": "Alice"}],
    batch_size=500,  # optional, default 1000
)

result = await harness.run()

# Convenient properties
harness.loaded_records  # Flat list of all loaded records
harness.load_calls  # List of per-batch record lists
harness.loaded_records_by_step  # Load step ID -> captured records
harness.quarantined_records_by_step  # Step ID -> diagnostic envelopes

For multiple sources, pass fixture_data={"source_a": records_a, "source_b": records_b} with exactly one entry for every extract step. A flat or pre-batched fixture remains supported for one source. loaded_records and load_calls aggregate sinks in DAG order; use the per-step views when sink identity matters. Captures accumulate over repeated runs of a harness. Fixture batches and captured loads/envelopes are copied to keep later Python mutations from changing the supplied fixtures or recorded writes.

The isolation boundary covers configured extract/load connectors and step contract quarantine on extract, transform, join, union, and load steps. Contract validation, coercion, privacy handling, and quality evaluation still run. The harness rejects configured contract stores, enabled lineage, and progress/extraction/grounded-emitter callbacks before execution; use file contract refs. It cannot intercept direct file, database, network, subprocess, or global-state access in retained Python, imported modules, custom validators, or caller-supplied implementations. It does not verify a real destination's transactions, constraints, permissions, or retry behavior.

Extraction preview versus offline checks¶

These commands have different I/O boundaries:

pycharter pipeline validate pipeline.yaml
pycharter pipeline inspect pipeline.yaml
pycharter pipeline run pipeline.yaml --dry-run

validate parses the manifest and DAG and resolves/checks contracts; inspect displays the configured DAG. Neither constructs source connectors or reads source records. For offline validation, use local contract files without settings.contract_store; a configured registry can require network reads. These commands do not establish a locked execution plan or prove delivery.

run --dry-run (Python: await pipeline.arun(dry_run=True)) is an extraction preview: it reads the configured source, runs transformations, and applies step contracts, including validation and inline contract quality checks. With FMP this can make billable/live requests; it is not an offline operation. Normal load and quarantine connectors are not constructed, so their configured destinations remain untouched. Invalid records follow the contract's policy: fail produces a failed preview, while quarantine removes them from the preview output without persisting quarantine envelopes. Its record_outcomes.would_quarantine counts candidates, while quarantined and total_quarantined remain zero. A quarantine policy still requires a declared DLQ destination. rows_loaded is zero; a successful preview does not prove that a later load will commit or that post-load checks will pass. Existing lineage/progress hooks may still emit events. Retained Python has the same permissions as during a normal run. The inspector's step previews also suppress load/quarantine writes while reading the upstream sources needed for the selected step.

Loading fixtures from files¶

Keep test data in version-controlled YAML or JSON files:

fixtures/orders.yamlfixtures/orders_batched.yaml

# Flat list format
- {order_id: "ORD-001", customer_id: "C1", amount: "99.50"}
- {order_id: "ORD-002", customer_id: "C2", amount: "14.99"}

name: "orders batched fixture"
batches:
  - [{order_id: "ORD-001", amount: "99.50"}]
  - [{order_id: "ORD-002", amount: "14.99"}]

from pycharter.pipeline_generator.testing import load_fixture

records = load_fixture("tests/fixtures/orders.yaml")
harness = PipelineTestHarness(pipeline, fixture_data=records)
result = await harness.run()

Assertion helpers¶

`assert_records_match`¶

from pycharter.pipeline_generator.testing import assert_records_match

# Ordered comparison (default)
assert_records_match(
    actual=harness.loaded_records,
    expected=[{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}],
)

# Order-insensitive
assert_records_match(
    actual=harness.loaded_records,
    expected=[{"id": 2}, {"id": 1}],
    order_matters=False,
)

# Subset check (actual must contain expected)
assert_records_match(
    actual=harness.loaded_records,
    expected=[{"id": 1}],
    subset=True,
)

`assert_fields_present` and `assert_no_field`¶

from pycharter.pipeline_generator.testing import assert_fields_present, assert_no_field

assert_fields_present(harness.loaded_records, ["user_id", "email", "created_at"])
assert_no_field(harness.loaded_records, "internal_debug_id")

`assert_field_values`¶

from pycharter.pipeline_generator.testing import assert_field_values

assert_field_values(
    harness.loaded_records,
    field_name="status",
    expected_values=["pending", "pending"],  # default filled in
)

`assert_schema_shape`¶

from pycharter.pipeline_generator.testing import assert_schema_shape

assert_schema_shape(
    harness.loaded_records,
    schema={"user_id": str, "price": float, "quantity": int},
)

`assert_record_count`¶

from pycharter.pipeline_generator.testing import assert_record_count

assert_record_count(result, expected=100)

Validating pipeline config¶

Check that a YAML config file is syntactically and semantically valid without running the pipeline:

from pycharter.pipeline_generator.testing import validate_pipeline_config

is_valid, errors = validate_pipeline_config("pipelines/orders/")
if not is_valid:
    for err in errors:
        print(f"[{err['section']}] {err['message']}")

Also accepts a config dict or a path to a single YAML file.

Patterns¶

pytest-asyncio setup¶

Add to conftest.py or pyproject.toml:

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

Or use @pytest.mark.asyncio on each async test.

Parametrized tests across fixtures¶

import pytest
from pathlib import Path
from pycharter.pipeline_generator.testing import load_fixture, PipelineTestHarness

FIXTURES = list(Path("tests/fixtures").glob("*.yaml"))


@pytest.mark.parametrize("fixture_path", FIXTURES, ids=[f.stem for f in FIXTURES])
async def test_pipeline_on_all_fixtures(pipeline, fixture_path):
    records = load_fixture(fixture_path)
    harness = PipelineTestHarness(pipeline, fixture_data=records)
    result = await harness.run()
    assert result.success, result.errors

Testing incremental logic¶

async def test_incremental_skips_old_records(pipeline):
    all_records = [
        {"id": 1, "updated_at": "2024-01-01T00:00:00"},
        {"id": 2, "updated_at": "2024-06-01T00:00:00"},
    ]
    # Supply only the recent record (simulating watermark-filtered extraction)
    harness = PipelineTestHarness(pipeline, fixture_data=[all_records[1]])
    result = await harness.run()
    assert_record_count(result, expected=1)
    assert harness.loaded_records[0]["id"] == 2