airflow-pytest-plugin 0.1.0

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

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

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.

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

Side	Where it runs	What it is
Producer	the worker	ArchivingJUnitResultParser, a drop-in parser= for PytestOperator
Reader	the API server	a FastAPI app + single-page viewer, registered as an Airflow plugin

---

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 ArchivingJUnitResultParser

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

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. ArchivingJUnitResultParser 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
  └─ ArchivingJUnitResultParser ──▶   {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):

Method & path	Returns
GET /	the single-page viewer (HTML)
GET /api/reports?dag_id=&run_id=	summaries, newest first
GET /api/reports/{report_id}	one report with per-case rows
GET /api/health	{"status": "ok"}
GET /api/docs	OpenAPI docs

Configuration

Setting	Default	Purpose
AIRFLOW_PYTEST_REPORTS_ROOT (env)	—	report root (highest precedence)
[pytest_reports] reports_root (cfg)	—	report root
built-in default	/opt/airflow/pytest-reports	fallback

Architecture (SOLID)

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

Module	Responsibility
layout.ReportLayout	the single ReportRef → directory mapping, shared by both sides
producer.ArchivingJUnitResultParser	write JUnit XML + meta.json (extends the operator's parser)
sources.ReportSource / FileSystemReportSource	read/index reports behind an interface (Dependency Inversion)
web.create_app	map HTTP onto a ReportSource — knows nothing about the filesystem
plugin.PytestReportsPlugin	register the app with Airflow
compat	the only module that imports Airflow; version differences resolved once
models	JSON-serializable view types; the web layer never sees operator types

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.