airflow-pytest-plugin

View airflow-pytest-operator results in the Airflow 3 web UI.

Package

Badge	What it tells you
	Latest release on PyPI — pip install airflow-pytest-plugin
	Supported Python versions (3.10+)
	Targets Airflow 3.x (FastAPI plugin UI)
	Distributed under the Apache-2.0 licence

Quality & build

Badge	What it tells you
	Build & test suite (lint, types, unit, integration) on main
	Test coverage of the package
	Fully type-checked with mypy --strict
	Linted & formatted with Ruff
	OpenSSF supply-chain security score

Trigger two suites from Airflow, then open Pytest Reports in the same sidebar: the runs you just started are at the top of the list, next to their history — with per-test detail, flaky tests, a test×run heatmap and failures clustered by error.

The operator runs a pytest suite as an Airflow task and parses the JUnit report into a structured result. This plugin archives each of those reports — keyed by dag_id / run_id / task_id / try — and serves a small web UI to browse them: pass/fail counts per run, durations, and the per-test breakdown (with failure messages) for any run. On top of that it adds cross-run analytics — flaky-test detection, per-test history, run-to-run comparison, a duration histogram, and a searchable catalogue of unique tests.

It has two halves that share one on-disk layout:

Contents

• Screenshots
• Install
• Quickstart
• Do I need cleanup="never"?
• How it works
• HTTP API
• Access control (RBAC)
• Coverage
• Configuration
• Prometheus metrics
• Architecture (SOLID)
• Development
• License

Screenshots

Overview — the run list with the historical chart (per-status legend toggles, run numbers, a carousel beyond 30 runs, an optional pass-rate trend line with a success-threshold gridline, and tick runs in the list to filter the chart to just their trend) beside a flaky-tests panel (with its own search and a quarantined-only toggle), KPI cards (including a clickable unique tests count), and Airflow-matched colours and font. The run list is grouped by dag·task by default (a checkbox toggles the flat list) — collapsible groups with run count, pass-rate, average duration and last status, each sortable on its own; select a whole group to chart its trend. The ⚙ button in the header opens dashboard settings, where each main-board panel (Recent runs, Reliability, Flaky tests) can be switched off — it is then not rendered at all, and the choice is remembered in your browser across reloads. Everything is on by default and the run list is never affected:

Dashboard settings — switch any main-board panel off and it is not rendered at all; the choice is remembered in your browser across reloads:

A single run — a clickable success donut (pass-rate over the test count; click a slice to filter by status), a coverage card next to the duration (when the run was produced by airflow-pytest-operator >= 0.6 with coverage=True — see Coverage; omitted when the run carries none), a test-duration histogram (10-second buckets, scrollable), case search / group-by-module, and every test's captured output on expand:

Flaky tests & comparison — from a run, Flaky tests lists the tests that both pass and fail across recent runs, each with a recent-outcome strip, a flakiness score, a trend arrow, and a quarantine badge, over a configurable analysis window; Compare to previous diffs it against the prior run; expanding a case offers its full history:

Slow tests & duration regressions — the Slowdowns KPI opens a panel of tests whose execution time got slower (recent-half average vs the older half, over a configurable window) alongside the slowest tests by average duration. A test that speeds back up drops off the list. Inside a run, the case table sorts by execution time (slowest first) so the heaviest tests surface immediately.

Test×run heatmap — the Heatmap button on a dag·task group (or a run's toolbar) opens a matrix of tests (rows) × recent runs (columns), each cell coloured by that test's outcome (did not run is an empty dashed cell). Flaky tests read as alternating rows, a regression as a block of failures filling in on the right, and a run that broke the build as a red/error column — all at a glance. Rows sort most-broken first; click a cell to open that run, or a test name to open its history.

Unique tests & failures — the Unique tests KPI opens the searchable, paginated catalogue of every distinct test (each with its runs / pass-fail-error-skip counts / average time); the Failures KPI shows what's broken now — failures in each dag·task's latest run, so the count shrinks as tests are fixed — grouped into clusters by normalized error (biggest first) so common root causes surface instead of per-failure spam. Expand a cluster to its tests, or open the same clusters scoped to one run via the detail's Error clusters button:

Install

pip install airflow-pytest-plugin          # producer side (workers)
pip install 'airflow-pytest-plugin[web]'   # reader side (API server)

On Airflow 3 the API server already provides FastAPI, so the bare install is enough there too; the [web] extra only adds the standalone dev server.

Quickstart

1. Point your operator at the archiving parser — the only DAG change:

from airflow_pytest_operator import PytestOperator
from airflow_pytest_plugin import ArchivingResultParser

PytestOperator(
    task_id="run_tests",
    test_path="tests/",
    parser=ArchivingResultParser(),   # was JUnitResultParser()
)

After each run the task log gets a tracking link straight to the archived report, opened inside the Airflow UI (Pytest report archived — view it at http://…/plugin/pytest-reports?dag=…&run=…&task=…&try=1), provided [api] base_url (or [webserver] base_url) is set — without it the log lists the run's coordinates instead.

2. Tell both sides where reports live (one place, read by producer and reader alike):

export AIRFLOW_PYTEST_REPORTS_ROOT=/opt/airflow/pytest-reports

or in airflow.cfg:

[pytest_reports]
reports_root = /opt/airflow/pytest-reports

In a distributed deployment this should be a shared volume that both the workers (writing) and the API server (reading) can see.

3. Open the UI. The plugin registers itself via the airflow.plugins entry point — no config. The app mounts on the API server at /pytest-reports, with a Pytest Reports entry under Browse in the nav.

Preview locally, without Airflow

python -m airflow_pytest_plugin.web --root ./pytest-reports --port 8000
# open http://127.0.0.1:8000/

Do I need cleanup="never"?

No. In the operator, the parser owns the report location, and a parser-supplied directory is never deleted by the runner under any cleanup policy. ArchivingResultParser supplies its own directory, so reports always survive regardless of the runner's cleanup setting. cleanup="never" only matters when you let the runner use throwaway temp dirs — which is exactly the fragile path (random names, no dag/run/task association, not visible to other workers) this plugin replaces.

How it works

worker                              shared volume                 API server
──────                              ─────────────                 ──────────
PytestOperator                      {root}/{dag}/{run}/           FastAPI app
  └─ ArchivingResultParser ──▶   {task}/t{try}/        ◀──── FileSystemReportSource
       report_request() → path          ├─ junit.xml              └─ lists meta.json,
       parse()          → meta.json     └─ meta.json                 parses junit.xml

• report_request() reads the live Airflow context (get_current_context(), available because the parser runs inside the task's execute()), computes the archive directory, and hands it to the operator's JUnit parser.
• parse() reuses the operator's JUnit parsing, then drops a meta.json sidecar carrying the Airflow coordinates + the summary. That sidecar makes each report self-describing, so the reader needs no database access.
• The reader lists by scanning meta.json files (fast) and parses junit.xml on demand for the per-case detail.

The directory is a human-friendly container; the authoritative identity always lives in meta.json (and the API's opaque, reversible report token), so awkward run_id characters like : are sanitised in the path without losing anything.

HTTP API

The app is mountable under any prefix; the viewer derives its API base at runtime. Endpoints (relative to the mount):

The reads (GET) and the delete are gated by Airflow RBAC — see below.

Access control (RBAC)

Access is enforced the way Airflow 3 enforces it: every check goes through Airflow's auth manager (is_authorized_dag(...)) — the same call Airflow's own DAG-run endpoints make — keyed by the report's dag_id and the authenticated user. Two permissions are checked:

The report list is filtered to the DAGs you may read, opening a report you can't read returns 403, and deleting one requires permission to trigger its DAG. Every check fails closed: if the auth manager can't be consulted, access is denied.

Airflow 2 → 3 mapping. Airflow 2's FAB used (action, resource) pairs — can_read / can_edit / can_delete / can_create on a resource such as DAG:<id>. Airflow 3 replaced these with the auth manager's method: GET ↔ read, POST ↔ create, PUT ↔ edit, DELETE ↔ delete, MENU ↔ menu access. This plugin maps read → GET and delete → POST (trigger), so it inherits your existing per-DAG roles with no extra configuration.

Plugin visibility. The nav entry is an Airflow external_views item, which has no per-permission gate, so the menu link is visible to every signed-in user; access is enforced on the content (a user who may read no DAG sees an empty list and 403 on direct links).

When auth is unavailable the fallback depends on why. With no Airflow installed — the bundled dev server — everything is allowed, which is the point of that mode. With Airflow installed but its auth API unreachable (an upgrade moved it, say) the reader denies every report and logs the reason: it cannot verify DAG permissions, and serving every team's runs would be the worst possible reading of "auth unavailable".

GET /api/health reports which mode is live — worth asserting in a smoke check for a shared deployment:

Two things sit outside per-DAG RBAC, by design — worth knowing before a wide rollout:

• GET /api/metrics is gated by AIRFLOW_PYTEST_METRICS_TOKEN (constant-time compare), not by DAG permissions: any holder of that token sees {dag_id, task_id} series for every archived dag·task. It is disabled (404) until you set the token — treat the token as a read-everything credential and scope it to your Prometheus scraper.
• GET /api/health and GET /api/version need no auth. They expose no run data, but health does report the configured reports_root path.

Report tokens encode a run's coordinates (dag·run·task·try) and nothing else — they are identifiers, never capabilities. Guessing one gains nothing: every token-addressed route re-checks permission on the run's DAG before serving.

Allure / TestOps export

Opt in per task and install allure-pytest on the worker:

parser=ArchivingResultParser(allure=True)

The parser then adds --alluredir (pytest errors with unrecognized arguments if allure-pytest is missing), so the raw Allure results are archived next to the report, with an executor.json linking the launch back to the Airflow run. Download them from a report's detail view, or GET /api/reports/{id}/allure.zip — then upload to Allure TestOps (allurectl upload …). The JUnit viewer is unaffected; both artifacts coexist.

Coverage

The run detail shows a Coverage card next to Duration. It needs pytest-cov on the worker, and the fraction reaches the viewer by one of two routes.

Recommended: with the archive. One flag on the parser is enough — the operator does not need coverage=True:

PytestOperator(
    ...,
    result_parser=ArchivingResultParser(coverage=True),
)

The parser adds --cov plus --cov-report=json:<archive>/coverage.json, then reads the fraction while archiving and bakes it into the run's meta.json on the spot. Prefer this route, because it:

• survives a failed run. The operator raises on a red suite (fail_on_test_failure=True, the default) or a tripped cov_fail_under gate, so its return_value XCom is never pushed — but the parser has already run, so the coverage of exactly the runs you most want to inspect is preserved;
• needs no metadata-DB query. The value is served from the report, so the api-server never round-trips to Airflow's shared metadata DB to render the card;
• coexists with the operator. Setting coverage=True on both is fine: the runner appends the parser's flags after the operator has built its own, so the operator still splices its flags, still parses its terminal TOTAL row for the XCom value, and its cov_fail_under gate is unaffected. A duplicate --cov measures the same thing once.

Scope. A bare --cov measures everything. If your project already narrows coverage — say addopts = "--cov=src" in pyproject.toml — pass the same scope, because pytest-cov unions scopes and a bare --cov on top would silently widen the number (usually by pulling the tests themselves in):

ArchivingResultParser(coverage=True, coverage_source="src")   # -> --cov=src

Fallback: from XCom. Without the parser flag, and with coverage=True on the operator, the viewer reads the fraction from the operator's return_value XCom on first view and bakes it in from there. This still works, but only for runs that finished successfully (see above), and a run opened seconds after it finishes may show the card a moment late while the XCom lands.

Either way the card is simply omitted when a run carries no coverage.

The bar it is read against comes from one of two places — the per-task one wins:

# 1. Global default for every run this viewer shows.
export AIRFLOW_PYTEST_SUCCESS_COVERAGE=0.7

# 2. Pinned to one suite; travels in the run's meta.json and OUTRANKS the env var.
ArchivingResultParser(coverage=True, coverage_threshold=0.5)   # legacy suite, gentler bar
ArchivingResultParser(coverage=True, coverage_threshold=0.95)  # core library, strict bar

A single reader-side variable cannot say that a core library should sit at 95% while a legacy smoke suite is fine at 50% — so the suite may state its own standard, and the viewer honours it. Leave coverage_threshold unset and the env var (then 0.85) applies. A value outside 0–1 is rejected with a warning and falls back to the default, rather than clamped — a task that meant 0.9 but wrote 90 should not silently paint every run red.

At or above the bar the card is green and reads meets target 70%; below it the card is red and reads below target 70%. The verdict is spelled out in words next to the colour, so it does not depend on seeing the tint.

Coverage never fails a run. Falling short only paints the card red — the run's own pass/fail is decided by its tests (and AIRFLOW_PYTEST_SUCCESS_THRESHOLD). If you want a shortfall to actually fail the task, that is the operator's job: set cov_fail_under on PytestOperator. The two are independent on purpose — you can watch coverage in the UI long before you are ready to enforce it in CI.

Configuration

Enable / disable the reader. Set AIRFLOW_PYTEST_PLUGIN_ENABLE to a falsey value (0, false, no, off) to stop the plugin registering its UI and API with Airflow; any other value, or leaving it unset, keeps it on (the default). This is a kill switch for the reader only — the producer-side ArchivingResultParser still archives reports regardless. It is read once at plugin discovery, so toggling it takes effect on the next API-server restart.

export AIRFLOW_PYTEST_PLUGIN_ENABLE=false   # hide the Pytest Reports UI + API

Scan cache. Loading the home page hits several summary-driven endpoints (the run list, the flaky panel, the unique-tests count), and the filter box queries as you type. To avoid walking the report tree once per call, the filesystem source reuses a single scan for AIRFLOW_PYTEST_SCAN_CACHE_TTL seconds (default 2.0; deletes invalidate it immediately). New runs therefore appear within a couple of seconds (or on Refresh); set it to 0 for no caching, or higher on a very large tree.

Prometheus metrics

GET /api/metrics exposes per-dag·task gauges (from each dag·task's latest run) in the Prometheus text format — airflow_pytest_latest_{passed,failed,errors,skipped, tests,pass_ratio,duration_seconds,success,run_timestamp_seconds}{dag_id,task_id} and airflow_pytest_dagtask_runs{dag_id,task_id}, plus globals airflow_pytest_{up,runs,dagtasks,latest_failures,series_truncated,build_info} (all gauges).

It's disabled by default and turns on only when you set a scrape token; requests must then present it as a bearer token (constant-time compared). The scrape is cheap and bounded — one cached directory scan, summary-derived (no per-run reads), capped at 2000 series — so it's safe to poll frequently.

export AIRFLOW_PYTEST_METRICS_TOKEN="$(openssl rand -hex 16)"

# prometheus.yml
scrape_configs:
  - job_name: airflow-pytest
    metrics_path: /pytest-reports/api/metrics   # the plugin's mount prefix
    authorization:
      credentials: "<AIRFLOW_PYTEST_METRICS_TOKEN>"
    static_configs:
      - targets: ["airflow-apiserver:8080"]

Retention (auto-cleanup)

Reports accumulate forever unless you prune them. Set any of the AIRFLOW_PYTEST_RETENTION_* knobs above (all opt-in; unset = keep everything) and schedule prune_reports from a maintenance DAG:

from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow_pytest_plugin import prune_reports

with DAG("pytest_reports_retention", schedule="@daily", catchup=False, ...):
    PythonOperator(task_id="prune", python_callable=prune_reports)

Each knob is a dimension and they combine as a union — a run is deleted if any applies:

• age — older than …_MAX_AGE_DAYS;
• count — beyond the newest …_MAX_RUNS of its dag·task;
• size — oldest-first until the tree fits …_MAX_TOTAL_MB.

The newest run of each dag·task is always kept, so a task's latest result never disappears. prune_reports(dry_run=True) reports what would go without deleting (its RetentionResult carries deleted, freed_bytes, scanned). Cleanup is scheduler-driven — the plugin never deletes on its own. For a custom policy, build a RetentionPolicy(...) and pass it (prune_reports(policy)).

Email alerts

Opt-in email notifications with an adaptive, styled HTML body — green for a clean pass, amber for flaky, red for a failure — listing the failed / flaky tests and linking back to the run:

Automatic notifications are per task, via email=True. The ArchivingResultParser(email=) flag is the switch: with email=True a "run finished" mail is sent after every run (styled by outcome — green pass / amber flaky / red fail), so a team gets notified without watching the run; the default email=False sends nothing, so a noisy ping / smoke suite can't flood the mailbox:

from airflow_pytest_plugin import ArchivingResultParser

parser = ArchivingResultParser(email=True)            # email me when each run finishes
parser = ArchivingResultParser(email_only_fail=True)  # email ONLY on a failed / flaky run
parser = ArchivingResultParser()                      # default: never auto-emails (ping-safe)

email_only_fail=True is for teams that don't want success mail: nothing arrives while runs are green, a red/amber notification arrives the moment a run fails or turns flaky (it wins over email=True when both are set).

Recipients are validated (RFC-bounded addresses; invalid configured entries are dropped with a warning) and deduplicated case-insensitively — listing the same mailbox twice, in the env or in the UI dialog, sends one email. The UI dialog also validates as you submit and names the bad address before anything is sent; the server re-validates and answers a plain-language 400.

A run is "failing" below AIRFLOW_PYTEST_SUCCESS_THRESHOLD (the same 0–1 bar the Passing runs KPI uses; default 0.85) — which is what colours the email. Recipients + transport are configured once, one of two ways:

Airflow mode — mail rides Airflow's own SMTP. Set the recipients here, then configure Airflow's SMTP the standard way (per the Airflow email guide):

export AIRFLOW_PYTEST_ALERTS_EMAIL_TO="team@example.com, oncall@example.com"
export AIRFLOW_PYTEST_SUCCESS_THRESHOLD=0.85   # optional

On Airflow 3 the default backend (airflow.utils.email.send_email) takes the SMTP host / port / STARTTLS from the [smtp] config but the login / password from the smtp_default connection — so you usually need both, present on the service that runs the task (the worker), not only the scheduler / API server. A minimal Gmail setup:

• [smtp] (env on every Airflow service): AIRFLOW__SMTP__SMTP_HOST=smtp.gmail.com, AIRFLOW__SMTP__SMTP_PORT=587, AIRFLOW__SMTP__SMTP_STARTTLS=True, AIRFLOW__SMTP__SMTP_SSL=False, AIRFLOW__SMTP__SMTP_MAIL_FROM=you@gmail.com.
• smtp_default connection: type SMTP, host smtp.gmail.com, port 587, login you@gmail.com, password = a Gmail App Password (2FA required; strip the spaces).

Standalone mode — no Airflow SMTP available (the bundled dev server, or a worker without mail configured), so also point the built-in SMTP client at your server:

export AIRFLOW_PYTEST_ALERTS_EMAIL_TO="team@example.com"
export AIRFLOW_PYTEST_SMTP_HOST=smtp.example.com
export AIRFLOW_PYTEST_SMTP_PORT=587
export AIRFLOW_PYTEST_SMTP_STARTTLS=true                 # default true; set false for plain SMTP
export AIRFLOW_PYTEST_SMTP_USER=apikey                   # omit user+password for an open relay
export AIRFLOW_PYTEST_SMTP_PASSWORD="$SMTP_PASSWORD"     # keep the secret out of shell history
export AIRFLOW_PYTEST_SMTP_FROM="pytest-reports@example.com"

Send one run by hand (from the UI). Open a run and click the ✉ Email button in its toolbar to email that run's summary. Recipients are optional (leave the field empty to use the configured AIRFLOW_PYTEST_ALERTS_EMAIL_TO); any you type are validated as email addresses and capped. The button appears only when a mail transport is configured. The action is RBAC-gated — it needs permission to read the run's DAG — and backed by POST /api/reports/{id}/email.

Every send is logged on the run. Each attempt (automatic or manual) lands in the run's meta.json; the run's toolbar then shows an ✉ Emails N bench that opens the send log — who was mailed, when, and a ✓ delivered / ✗ failed mark per send (newest 50 kept). When the run has raw Allure results, the notification email carries them as an allure-results.zip attachment (skipped above 10 MB so mail servers accept the message).

Transport precedence: if you set AIRFLOW_PYTEST_SMTP_HOST, that standalone SMTP client is used directly — even inside Airflow — so your explicit config always wins. Otherwise mail goes through Airflow's configured SMTP (airflow.utils.email.send_email; set it up per the Airflow email/SMTP guide). Alerting is best-effort — a mail or config failure is logged (the real reason lands in the reader's log; the UI/endpoint only says "failed to send") and never fails the task that archived the run. The decision (evaluate_alerts) and orchestrator (notify_for_run, which takes dry_run= and a custom mailer=) are importable and side-effect-free for testing.

Architecture (SOLID)

Mirrors the operator's layering — each piece has one reason to change:

Adding a different backing store (e.g. an XComReportSource reading the metadata DB) is a new ReportSource, not an edit of the web app (Open/Closed).

Development

pip install -e '.[dev,web]'
pytest -q
ruff check src tests && ruff format --check src tests
mypy src

License

Apache-2.0. See LICENSE.