pytest-airflow-in-a-box

pytest-airflow-in-a-box is a pytest plugin for testing Apache Airflow DAGs without a live Airflow deployment. It targets Airflow 3 and provides the package and plugin foundation for a small, typed testing surface.

The package auto-registers with pytest, creates an isolated metadata database, and provides typed fixtures for persisted Dags, DagRuns, task instances, sessions, and Dag bags.

Requirements

• CPython 3.10 through 3.14
• pytest 8 or newer
• Apache Airflow 3.1 or newer, below 4
• Linux or macOS for Airflow-backed tests

Apache Airflow does not support native Windows installations. Windows development should use WSL2 or the included devcontainer; platform-independent package checks alone do not imply full Windows Airflow support.

The released compatibility matrix is exercised against Airflow 3.1.0, 3.1.1, 3.1.2, 3.1.3, 3.1.5, 3.1.6, 3.1.7, 3.1.8, 3.2.0, 3.2.1, 3.2.2, and 3.3.0 across CPython 3.10 through 3.14 using Airflow's published constraints files.

Installation

uv add --dev pytest-airflow-in-a-box

The pytest11 entry point loads the plugin automatically. Consumer projects do not need to add a pytest_plugins declaration.

The bundled pytest plugins are intentional runtime dependencies. pytest-xdist is part of the supported execution model: controller bootstrap state and worker-scoped artifacts are coordinated for parallel runs. pytest-timeout backs up Airflow's per-file Dag parse watchdog with a corpus-scaled deadline on every bundled smoke item, so whichever worker produces the shared corpus cannot wedge the test session outside the per-file parser boundary.

The plugin is inert on runs without Airflow-facing tests: session startup only prepares a disposable run directory and AIRFLOW__* environment variables. Airflow itself is imported and the metadata database migrated lazily, on the first test that carries a db_test/api_test marker or uses a database-backed plugin fixture. A pytest -k unrelated run in a shared venv never pays the Airflow import or migration cost. Tests that touch the metadata database directly (their own create_session calls, for example) without a plugin fixture must carry db_test to trigger initialization.

To disable the plugin entirely for a run:

pytest -p no:pytest_airflow_in_a_box

Database backends

The metadata database defaults to a tuned, WAL-mode SQLite file created per run -- fast and correct for single-writer test workloads, and the right default. SQLite serializes writers, so it cannot reproduce the concurrency semantics a real deployment runs on (row-level locking, SELECT ... FOR UPDATE, multiple concurrent writers). Opt into a disposable Postgres backend when you need that fidelity:

uv add --dev "pytest-airflow-in-a-box[postgres]"
pytest --airflow-db-backend=postgres

or persistently via the airflow_db_backend ini option (sqlite or postgres). The Postgres backend provisions one container per session with testcontainers and hands Airflow the resulting SQLAlchemy URL; every worker in an xdist run shares that one database, mirroring the production topology of one metadata database behind many workers. It requires a running Docker daemon and the postgres extra. When either is missing the plugin fails loudly with a usage error rather than silently skipping, so a misconfigured Postgres run can never be mistaken for a passing SQLite run.

SQLite-with-WAL and Postgres are not behaviorally equivalent; a suite green on one is not guaranteed green on the other. That divergence is the point -- run the Postgres backend to catch dialect- and concurrency-specific behavior before it ships.

Plugin contributors can install the optional dependencies with make install-postgres (or uv sync --extra postgres).

Development

uv sync
uv run prek install
make all

Run the GitHub Actions workflow locally on Linux with act:

act pull_request

act cannot reproduce native macOS or Windows behavior.

Task execution

from airflow.sdk import task
from airflow.utils.state import TaskInstanceState

from pytest_airflow_in_a_box.taskinstance import ordered_task_instances


def test_task(dag_maker):
    with dag_maker() as dag:

        @task
        def answer():
            return 42

        answer()

    dag_run = dag_maker.create_dagrun()
    ti = dag_maker.run_ti("answer", dag_run)

    assert ti.state == TaskInstanceState.SUCCESS
    assert ti.xcom_pull(task_ids="answer", session=dag_maker.session) == 42
    assert ordered_task_instances(dag_run, dag, session=dag_maker.session) == [ti]

Public task helpers live in pytest_airflow_in_a_box.taskinstance: run_task_instance, ordered_task_instances, run_trigger, TaskResolutionError, and TriggerExecutionError. run_task_instance resolves the executable task automatically for any dag_maker-persisted Dag, including task instances queried through a different session (e.g. the session fixture); pass task= only for Dags the plugin does not own. The DagMaker protocol additionally exposes create_dagrun, create_ti, and run_ti. Passing map_index expands a mapped task on demand; upstream-XCom mapping works after its producer has run in the same DagRun. Passing run_triggerer=True runs the persisted trigger event and resumes a deferred task inline, bounded by trigger_timeout seconds.

Deferrable operators

run_trigger drives one trigger's async run() to its first TriggerEvent on a private event loop, with no triggerer job, DagRun, or metadata database. cleanup() always runs, and a trigger that never fires raises TriggerExecutionError instead of hanging the suite.

from pytest_airflow_in_a_box.taskinstance import run_trigger


def test_trigger_fires():
    event = run_trigger(MyTrigger(target=42), timeout=5.0)

    assert event.payload == {"value": 42}

Compose the two halves to cover defer -> fire -> resume in one test:

def test_operator_resumes(dag_maker):
    with dag_maker():
        MyDeferrableOperator(task_id="wait")

    ti = dag_maker.run_ti("wait", run_triggerer=True, trigger_timeout=5.0)

    assert ti.state == TaskInstanceState.SUCCESS

DB-free task execution

run_task executes one operator through the Task SDK in process, with no metadata database. XCom, Variable, and Connection traffic is answered from seeded dictionaries; unseeded lookups fail exactly like a live deployment. Task callbacks and listeners stay silent unless the call passes run_callbacks=True. try_number selects the synthetic attempt; operator retry configuration determines whether a failure reaches UP_FOR_RETRY and its retry callback. Asset inlet/outlet validation is accepted as active in this deployment-free path.

def test_operator(run_task):
    result = run_task(
        my_operator,
        variables={"answer": "42"},
        connections={"db": {"conn_type": "postgres", "host": "example.com"}},
    )

    assert result.state == TaskInstanceState.SUCCESS
    assert result.xcoms["return_value"] == "expected"

Seeding Variables and Connections

airflow_variables and airflow_connections are the metastore counterparts to run_task's variables=/connections= keywords. Rows are committed, so hooks, operators, and Airflow's metastore secrets backend resolve them exactly as they would in a deployment, and every row the fixture inserted is deleted on teardown -- including after a failing test:

def test_hook(airflow_connections, airflow_variables, dag_maker):
    airflow_variables({"answer": "42"})
    airflow_connections(
        {"db": {"conn_type": "postgres", "host": "example.com", "password": "s3cret"}}
    )

    with dag_maker():
        MyOperator(task_id="read", conn_id="db")

    assert dag_maker.run_ti("read").state == TaskInstanceState.SUCCESS

Connection fields are the same flat shape run_task(connections=...) takes, so conn_type defaults to generic and extra is a JSON object string, not a dict. A uri is not accepted -- pass the fields. Repeated calls accumulate rather than replace, and neither fixture overwrites a row it did not insert, so an existing key, conn_id, or one of Airflow's default connections fails loudly instead of being clobbered.

Environment variables outrank these rows. Airflow's default secrets search path is the environment backend and then the metastore backend, so AIRFLOW_VAR_ANSWER or AIRFLOW_CONN_DB -- however it got set, including through airflow_config(env=...) -- wins over anything seeded here. Rather than leave a silently shadowed row to debug, both fixtures refuse to seed an identifier whose AIRFLOW_VAR_*/AIRFLOW_CONN_* name is already set. Seed through the environment when you want the environment backend exercised, and through these fixtures when you want the metastore one.

Seeded names are database-global, not test-local. Every xdist worker shares one metadata database and the conn_id in the test's operator cannot be renamed per worker, so two tests seeding the same name concurrently collide. Give each test unique identifiers, or group colliding tests onto one worker with @pytest.mark.xdist_group.

Structlog capture

Airflow 3 logs through structlog, where pytest's builtin caplog cannot see records. The cap_structlog fixture records every event emitted during the test:

def test_logging(cap_structlog, dag_maker):
    ...
    assert "task_event" in cap_structlog
    assert {"answer": 42, "log_level": "warning"} in cap_structlog

Dag-file collection

Point the collector at a directory of real Dag files and every *.py file below it is collected as a dag-import test item that fails on import errors or a Dag-free file. Off unless configured:

pytest --collect-dag-folder=dags/

or persistently via the airflow_collect_dags_folder ini option. Collected items are auto-marked db_test; files also matching test_*.py naming are deduplicated against pytest's default Python collector.

A Dag file may pin param cases through a module-level literal, read without importing the file:

PYTEST_DAG_CASES = {
    "dev": {"environment": "dev"},
    "prod": {"environment": "prod"},
}

Each case collects as a sibling dag-params[...] item that validates the pinned values against every Dag the file declares -- undeclared keys and schema violations fail the case.

Airflow configuration

airflow_config overrides Airflow configuration options and plain environment variables through one code path, as a context manager or a decorator. Options are applied as AIRFLOW__SECTION__KEY environment variables -- the same pre-import-safe mechanism bootstrap uses -- so they reach every Airflow configuration parser in the process, including the Task SDK parser added in Airflow 3.2:

from pytest_airflow_in_a_box.config import airflow_config


def test_with_overrides():
    with airflow_config({("core", "unit_test_mode"): "False"}, env={"MY_FLAG": "1"}):
        ...


@airflow_config({("core", "dagbag_import_timeout"): "120"})
def test_decorated(): ...

Every name is restored exactly on exit, and a name that was absent beforehand is deleted rather than emptied. Nesting restores last-in-first-out. A None value makes a name absent for the duration of the context, so Airflow falls back to airflow.cfg and then to its own default:

with airflow_config({("core", "dagbag_import_timeout"): None}):
    ...  # conf.get returns Airflow's default

Both mappings are validated before anything is assigned, so a malformed argument cannot leave the environment partly modified. Validation runs on context entry, so a bad argument to the decorator form surfaces as a test failure rather than a collection error. env names may not start with AIRFLOW__ -- pass configuration options through overrides instead -- but AIRFLOW_HOME and other single-underscore names are fine.

Airflow resolves SQL_ALCHEMY_CONN, DAGS_FOLDER, and PLUGINS_FOLDER into airflow.settings globals once at import, and those do not follow an environment assignment. Pass refresh_settings=True for options read through settings rather than through the config parser:

with airflow_config({("core", "plugins_folder"): str(tmp_path)}, refresh_settings=True):
    ...  # airflow.settings.PLUGINS_FOLDER now agrees

It defaults to off because it imports Airflow and rewrites process-global state bootstrap owns, and it is a partial remedy: a module that re-exported a settings value by value froze that binding at import and no refresh can update it.

Three things worth knowing:

• Values are expanded when Airflow reads them. conf.get runs the raw variable through expandvars then expanduser, so a value containing ~ or $ does not round-trip -- os.environ holds the literal while conf.get returns the expansion.
• A None override does not hide a _CMD/_SECRET sibling. Setting a plain value always wins, but None means "fall back to whatever Airflow would otherwise do", and an already-set sibling variable is one of those things.
• Do not wrap the first use of api_client/api_server_url. Those fixtures launch a session-scoped subprocess that inherits the environment live at startup, so an override would outlive the context inside that server.

conf_vars ships as a deprecated alias under the name public Airflow docs teach. It emits a DeprecationWarning and carries this plugin's semantics, so it does not recompute the settings globals the way upstream's does -- use airflow_config(..., refresh_settings=True) for that.

Smoke tests

A bundled catalog of zero-boilerplate checks against the configured Dag folder, synthesized with no files written. Off unless configured:

pytest --airflow-smoke --dag-folder=dags/

or persistently via the airflow_smoke ini option. Every item carries smoke, so -m smoke / -m "not smoke" select exactly the bundled catalog. Explicit selection is honored: pointing pytest at a file or node ID (pytest tests/test_x.py, pytest tests/test_x.py::test_one) runs only that selection and drops the catalog, while directory positionals (pytest tests/), bare runs, and testpaths-driven runs keep it. -k, -m, and --deselect ::smoke::<name> apply to the items as usual:

Under pytest-xdist, bundled items remain independently schedulable across workers. The first item to need the corpus parses it once and publishes a serialized artifact below the isolated run root; the other workers reuse that artifact instead of reparsing every Dag. The smoke marker itself has no scheduling effect, so user-authored smoke tests remain fully parallel too.

• test_dag_bag_integrity -- fails on import errors and per-file parse timeouts (airflow_dag_parse_timeout, default 30 seconds, exported as AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT so Airflow hard-kills runaway files); warns with SlowDagParseWarning on files above airflow_dag_parse_slowpoke_ratio (default 0.75) of the timeout without failing the run; logs a slowest-first parse-timing table
• test_dag_serialization_roundtrip -- every parsed Dag survives Airflow's scheduler serialization round trip; logs a slowest-first per-Dag timing table and carries a corpus-scaled pytest-timeout deadline (floored at 30 seconds, so a tuned-down parse timeout cannot starve the serialization pass) so a pathological Dag is named before an outer CI timeout
• test_no_duplicate_dag_ids -- no two Dag files declare the same dag_id
• test_schedule_sanity -- every scheduled Dag computes its next run without raising
• test_pool_references_exist -- every task's pool exists in the metadata database (db_test)

The serialization-backed checks (test_dag_serialization_roundtrip, test_schedule_sanity, test_dag_serialization_snapshot) share the producer's serialized-Dag cache across workers, so the corpus is parsed and the selected Dags are serialized once per run. Two ini options bound the cost on large generated corpora:

• airflow_serialization_sample_size (default 0, meaning every Dag) -- serialize only a deterministic sample of N Dags, selected by hashing each dag_id with airflow_serialization_sample_seed (default 0); the same corpus and seed always select the same sample, and test_schedule_sanity skips Dags outside it. Incompatible with --airflow-smoke-update, which must regenerate every snapshot
• run with --log-cli-level=INFO to stream per-Dag serialization progress live; captured-only logs do not survive a hard outer kill

Four additional policy checks appear only when their ini is configured, so defaults stay zero-config:

• airflow_dag_id_pattern -- every dag_id matches the given regex
• airflow_required_dag_tags -- every Dag carries the listed tags
• airflow_forbid_default_owner -- no task is owned by the stock airflow owner
• airflow_dag_snapshot_dir -- every Dag's serialized structure (topology, schedule, params, task attrs) matches its committed snapshot in the configured directory; regenerate with --airflow-smoke-update

Database cleanup

clear_db is a registry-driven whole-database reset for serial setup and teardown contexts:

from pytest_airflow_in_a_box.db import TableGroup, clear_db

clear_db()  # every group
clear_db(tables={TableGroup.VARIABLES})  # one group

Requesting a group also clears the groups whose rows reference it (RUNS clears task instances and XCom rows), and clearing CONNECTIONS recreates Airflow's default connections.

Live REST API

api_client lazily starts one isolated airflow api-server per test process on a loopback ephemeral port and returns a typed client authenticated through SimpleAuthManager:

import pytest


@pytest.mark.api_test
def test_api(api_client, dag_maker):
    with dag_maker(dag_id="visible"):
        ...

    response = api_client.get("/api/v2/dags/visible")

    assert response.status == 200
    assert response.body["dag_id"] == "visible"

The api_test marker alone also starts the server, and every activated test -- marked or requesting api_client/api_server_url -- gets the selected URL published as AIRFLOW__API__BASE_URL for its duration, so application code can discover the endpoint through active Airflow configuration:

import pytest
from airflow.configuration import conf


@pytest.mark.api_test
def test_application_client():
    base_url = conf.get("api", "base_url")
    assert base_url.startswith("http://127.0.0.1:")

Markers

• db_test: requires the isolated metadata database (triggers its lazy initialization)
• api_test: starts the isolated REST API server lazily and publishes its URL as AIRFLOW__API__BASE_URL for the test's duration (triggers lazy database initialization)
• postgres: requires a provisioned Postgres metadata database (the postgres extra plus Docker)
• compat: end-user tests exercised across the version matrix
• need_serialized_dag([enabled]): request serialized Dag behavior from dag_maker
• environment(name): run only when the named environment's sentinel path exists, configured via the airflow_environments ini line list (lab = /opt/lab/sentinel)
• smoke: a bundled zero-boilerplate check, opt in with airflow_smoke

Compatibility suite

The repository's tests/enduser/ suite is a sanitized consumer-style catalog run on every certified matrix leg. It covers custom operators, TaskFlow and mapping, hooks and connections, SQLite provider SQL, sensors, deferral, callbacks and retries, assets, provider-shaped packages, DagBag/collection, logging, xdist, and REST API CRUD. The provider-shaped corpus verifies user package composition and execution; registering a real provider distribution entry point remains out of scope because that is Airflow's packaging surface rather than this plugin's test surface.

Defaults

The plugin needs zero ini configuration. It applies --tb=short, -ra, --durations=20, and failed-only tmp_path retention, but only where the user has not chosen a value -- explicit flags and ini settings always win. Warning filters silence traced third-party deprecation noise (flask_appbuilder, flask_sqlalchemy, starlette) while keeping Airflow's own deprecation warnings visible, and promote pytest's collection and unraisable warnings to errors. User-supplied filterwarnings lines take precedence.

Diagnostics

--airflow-doctor prints a one-shot, copy-pasteable report and exits without collecting or running tests -- useful for bug reports and "why is it slow/failing here" triage:

pytest --airflow-doctor

The report covers the storage ladder decision and its reason, the resolved AIRFLOW_HOME, database URL scheme, and backend tier, plugin/pytest/Python/Airflow versions plus the resolved capability table, and API server state. The API server section always reads "not started": the api_server_url fixture is a lazy, per-process, session-scoped subprocess with no state before a test requests it or an api_test-marked test runs, and a standalone --airflow-doctor invocation never does either.

License

Apache License 2.0. See LICENSE, NOTICE, and PROVENANCE.md.