Testing
Lesson 14 — Python
Learning objectives
- Write and run tests with pytest
- Use fixtures for shared setup
- Parametrise tests to cover many cases
- Test exceptions, warnings and floating-point results
- Test DataFrames correctly
- Measure coverage and run tests in CI
pytest
pip install pytest pytest-covproject/
├── src/
│ └── study/
│ ├── __init__.py
│ ├── derivations.py
│ └── validation.py
└── tests/
├── conftest.py
├── test_derivations.py
└── test_validation.py# tests/test_derivations.py
from datetime import date
from study.derivations import study_day
def test_reference_date_is_day_one():
assert study_day(date(2026, 3, 15), date(2026, 3, 15)) == 1
def test_there_is_no_day_zero():
assert study_day(date(2026, 3, 14), date(2026, 3, 15)) == -1
def test_days_after_reference():
assert study_day(date(2026, 3, 20), date(2026, 3, 15)) == 6pytest # everything
pytest tests/test_derivations.py
pytest -k "day_zero" # by name pattern
pytest -v # verbose
pytest -x # stop at the first failure
pytest --lf # only the tests that failed last time
pytest -q --tb=short # concise output
pytest -n auto # parallel (needs pytest-xdist)Discovery conventions: files named test_*.py, functions named test_*, classes named Test*. No registration, no boilerplate — a plain assert is a test.
Assertions
assert result == expected
assert result != wrong
assert result is None
assert isinstance(result, pd.DataFrame)
assert "USUBJID" in df.columns
assert len(result) == 3
assert all(x > 0 for x in values)
assert any(x is None for x in values)pytest rewrites the assertion so the failure message shows the values:
E assert 183 == 182
E + where 183 = study_day(datetime.date(2026, 9, 12), ...)Add context with a message:
assert len(result) == 306, f"Expected 306 subjects, got {len(result)}"Floating point
import pytest
assert result == pytest.approx(22.857, rel=1e-3)
assert result == pytest.approx(22.857142857, abs=1e-6)
assert [1.0, 2.0] == pytest.approx([1.0000001, 2.0])
assert {"a": 1.0} == pytest.approx({"a": 1.0000001})Never == on floats. The same rule as R’s dplyr::near().
Exceptions
def test_negative_height_is_rejected():
with pytest.raises(ValueError, match="must be positive"):
derive_bmi(70, -175)
def test_error_carries_the_offending_value():
with pytest.raises(ValueError) as exc_info:
derive_bmi(70, 0)
assert "0" in str(exc_info.value)
def test_type_error_on_string_input():
with pytest.raises(TypeError):
derive_bmi("70", 175)match= is a regex against the message. Asserting on the message text keeps error quality from eroding — the same argument as snapshot-testing error messages in R’s testthat lesson.
Warnings
def test_warns_on_truncation():
with pytest.warns(UserWarning, match="truncated"):
result = write_xpt_checked(df, spec, path, strict=False)
def test_no_warning_on_valid_input():
with warnings.catch_warnings():
warnings.simplefilter("error") # any warning becomes an error
derive_bmi(70, 175)Fixtures
Shared setup, requested by naming them as parameters.
# tests/conftest.py
import pytest
import pandas as pd
from datetime import date
@pytest.fixture
def adsl():
"""A minimal ADSL for testing."""
return pd.DataFrame({
"USUBJID": ["001", "002", "003", "004"],
"TRT01A": ["Placebo", "Placebo", "Drug A", "Drug A"],
"AGE": [45, 72, 38, 66],
"SEX": ["F", "M", "F", "M"],
"SAFFL": ["Y", "Y", "N", "Y"],
"TRTSDT": pd.to_datetime(["2026-03-15", "2026-03-18",
None, "2026-03-20"]),
})
@pytest.fixture
def adae(adsl): # fixtures can use other fixtures
return pd.DataFrame({
"USUBJID": ["001", "001", "002", "004"],
"AEDECOD": ["Headache", "Nausea", "Rash", "Headache"],
"ASTDT": pd.to_datetime(["2026-04-01", "2026-04-15",
"2026-03-10", "2026-05-01"]),
})def test_safety_population_excludes_untreated(adsl):
result = filter_safety(adsl)
assert len(result) == 3
assert "003" not in result["USUBJID"].valuesconftest.py fixtures are available to every test in that directory and below, with no import.
Scope
@pytest.fixture(scope="function") # default — fresh for every test
@pytest.fixture(scope="class")
@pytest.fixture(scope="module") # once per test file
@pytest.fixture(scope="session") # once for the whole runUse session scope only for genuinely read-only expensive setup:
@pytest.fixture(scope="session")
def large_dataset():
return pd.read_parquet("tests/data/large.parquet")A mutable session-scoped fixture is a source of order-dependent test failures — one test modifies it and a later one fails for no visible reason.
Setup and teardown
@pytest.fixture
def temp_output(tmp_path):
"""A temporary directory, cleaned up automatically."""
d = tmp_path / "output"
d.mkdir()
yield d # the test runs here
# anything after yield is teardown
@pytest.fixture
def database():
conn = create_connection()
yield conn
conn.close() # always runs, even on failureBuilt-in fixtures worth knowing:
def test_writes_a_file(tmp_path): # a temp directory per test
path = tmp_path / "out.csv"
write_data(df, path)
assert path.exists()
def test_prints_summary(capsys): # capture stdout/stderr
print_summary(df)
captured = capsys.readouterr()
assert "306 subjects" in captured.out
def test_env_var(monkeypatch): # patch anything, undone after
monkeypatch.setenv("STUDY_ENV", "test")
monkeypatch.setattr(module, "CONSTANT", 42)
monkeypatch.chdir(tmp_path)monkeypatch is the equivalent of withr::local_*() — it reverses everything after the test.
Parametrisation
One test, many cases:
@pytest.mark.parametrize("age,expected", [
(10, "<18"),
(17, "<18"),
(18, "18-64"), # boundary
(30, "18-64"),
(64, "18-64"), # boundary
(65, ">=65"), # boundary
(90, ">=65"),
(None, "Missing"),
])
def test_age_group(age, expected):
assert derive_agegr1(age) == expectedEight tests, reported individually:
tests/test_derivations.py::test_age_group[10-<18] PASSED
tests/test_derivations.py::test_age_group[18-18-64] PASSED
tests/test_derivations.py::test_age_group[65->=65] PASSEDMultiple parameters produce the cross product:
@pytest.mark.parametrize("weight", [50, 70, 100])
@pytest.mark.parametrize("height", [150, 175, 200])
def test_bmi_is_positive(weight, height):
assert derive_bmi(weight, height) > 0 # 9 testsWith IDs for readable output:
@pytest.mark.parametrize(
"dtc,expected",
[
("2026-03-15", date(2026, 3, 15)),
("2026-03", date(2026, 3, 1)),
("2026", date(2026, 1, 1)),
],
ids=["complete", "missing-day", "year-only"],
)
def test_impute_dtc(dtc, expected):
assert impute_dtc(dtc, "first") == expectedTesting DataFrames
import pandas as pd
from pandas.testing import assert_frame_equal, assert_series_equal
def test_derivation_produces_expected_output(adsl):
result = derive_age_groups(adsl)
expected = pd.DataFrame({
"USUBJID": ["001", "002", "003", "004"],
"AGE": [45, 72, 38, 66],
"AGEGR1": ["18-64", ">=65", "18-64", ">=65"],
})
assert_frame_equal(
result[["USUBJID", "AGE", "AGEGR1"]],
expected,
check_dtype=False, # allow int64 vs Int64
check_like=True, # ignore column and row order
)Often better to assert on properties rather than the whole frame:
def test_derivation_preserves_structure(adsl):
result = derive_age_groups(adsl)
assert len(result) == len(adsl)
assert set(adsl.columns) <= set(result.columns) # nothing lost
assert "AGEGR1" in result.columns
assert result["AGEGR1"].notna().all()
assert set(result["AGEGR1"]) <= {"<18", "18-64", ">=65", "Missing"}
assert result["USUBJID"].is_uniqueProperty assertions survive an added column; a full-frame comparison does not. Use assert_frame_equal when the exact output is the contract, and property assertions otherwise.
Marks
@pytest.mark.slow
def test_full_pipeline():
...
@pytest.mark.skip(reason="Blocked on a data fix")
def test_something():
...
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX paths")
def test_paths():
...
@pytest.mark.xfail(reason="Known issue #142")
def test_known_bug():
...# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
markers = [
"slow: long-running tests",
"integration: requires a database",
]
addopts = "-q --strict-markers"pytest -m "not slow"
pytest -m "integration"--strict-markers makes a typo in a mark name an error rather than silently doing nothing.
Coverage
pytest --cov=study --cov-report=term-missing --cov-report=htmlName Stmts Miss Cover Missing
---------------------------------------------------------
src/study/derivations.py 84 6 93% 45-48, 91, 103
src/study/validation.py 62 14 77% 28-35, 55-60
---------------------------------------------------------
TOTAL 146 20 86%--cov-report=term-missing names the uncovered lines, which is the useful output — those are the branches no test exercises.
Coverage measures which lines ran, not whether the assertions were meaningful. Use it to find untested code, not as a target.
[tool.coverage.report]
fail_under = 85
exclude_lines = ["pragma: no cover", "if TYPE_CHECKING:", "raise NotImplementedError"]Mocking
from unittest.mock import Mock, patch
def test_handles_api_failure():
with patch("study.fetch.requests.get") as mock_get:
mock_get.side_effect = ConnectionError("Network down")
result = fetch_terminology("2026-03")
assert result is None
def test_calls_the_right_endpoint():
with patch("study.fetch.requests.get") as mock_get:
mock_get.return_value.json.return_value = {"terms": []}
mock_get.return_value.status_code = 200
fetch_terminology("2026-03")
mock_get.assert_called_once()
assert "2026-03" in mock_get.call_args[0][0]Mock what you do not control — network, time, filesystem, randomness. Do not mock your own functions; if a test needs to, the design is probably wrong.
CI
# .github/workflows/test.yml
name: Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.11", "3.12"]
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: pip
- run: pip install -e ".[dev]"
- run: ruff check .
- run: ruff format --check .
- run: mypy src/
- run: pytest --cov=study --cov-report=xml -m "not slow"
- uses: codecov/codecov-action@v4
with:
files: coverage.xmlR and Python side by side
| testthat | pytest |
|---|---|
test_that("desc", {...}) |
def test_desc(): |
expect_equal(a, b) |
assert a == b |
expect_equal(a, b, tolerance=) |
assert a == pytest.approx(b) |
expect_true(x) |
assert x |
expect_error(f(), "msg") |
with pytest.raises(E, match="msg"): |
expect_warning(f()) |
with pytest.warns(W): |
expect_type(x, "double") |
assert isinstance(x, float) |
expect_length(x, 3) |
assert len(x) == 3 |
helper-*.R |
conftest.py |
withr::local_*() |
monkeypatch |
expect_snapshot() |
pytest --snapshot-update (syrupy) |
covr::report() |
pytest --cov |
devtools::test() |
pytest |
skip_if() |
@pytest.mark.skipif |
Common mistakes
| Mistake | Consequence | Fix |
|---|---|---|
== on floats |
Flaky failures | pytest.approx() |
| Tests depending on each other | Order-dependent failures | Fixtures, self-contained tests |
| Mutable session-scoped fixture | One test breaks another | Function scope, or return a copy |
| Testing only the happy path | Breaks on real data | Empty, None, boundaries |
| Full-frame comparison for everything | Breaks on any added column | Assert properties |
| Mocking your own code | Tests the mock, not the code | Refactor instead |
| Chasing 100% coverage | Meaningless tests | Target uncovered logic |
| Unseeded randomness | Flaky suite | Fix the seed |
Exercise 9.1 — Test a derivation thoroughly
Write a complete test suite for derive_trtemfl(ae_start, trt_start, trt_end, window=30) returning "Y"/"N", using parametrisation for the boundary cases.
Show solution
# src/study/derivations.py
from datetime import date, timedelta
def derive_trtemfl(
ae_start: date | None,
trt_start: date | None,
trt_end: date | None,
window: int = 30,
) -> str:
"""Derive the treatment-emergent adverse event flag.
An event is treatment emergent if it starts on or after the first dose
and no later than `window` days after the last dose.
Args:
ae_start: Adverse event start date. `None` is treated conservatively
as treatment emergent when the subject was treated.
trt_start: First dose date. `None` means untreated.
trt_end: Last dose date. `None` means treatment is ongoing.
window: Days after last dose to include. Must be non-negative.
Returns:
"Y" or "N".
Raises:
ValueError: If `window` is negative, or `trt_end` precedes `trt_start`.
TypeError: If a date argument is not a date or None.
"""
if window < 0:
raise ValueError(f"window must be non-negative, got {window}")
for name, value in (("ae_start", ae_start), ("trt_start", trt_start),
("trt_end", trt_end)):
if value is not None and not isinstance(value, date):
raise TypeError(f"{name} must be a date or None, got {type(value).__name__}")
if trt_start is not None and trt_end is not None and trt_end < trt_start:
raise ValueError(f"trt_end ({trt_end}) precedes trt_start ({trt_start})")
if trt_start is None:
return "N" # untreated subject
if ae_start is None:
return "Y" # conservative: unknown date, treated
if ae_start < trt_start:
return "N"
if trt_end is not None and ae_start > trt_end + timedelta(days=window):
return "N"
return "Y"# tests/test_trtemfl.py
from datetime import date
import pytest
from study.derivations import derive_trtemfl
TRT_START = date(2026, 3, 15)
TRT_END = date(2026, 6, 15)
# ---- Core behaviour ------------------------------------------------------
@pytest.mark.parametrize(
"ae_start,expected",
[
(date(2026, 1, 1), "N"), # long before
(date(2026, 3, 14), "N"), # day before first dose
(date(2026, 3, 15), "Y"), # ON first dose — boundary
(date(2026, 3, 16), "Y"), # day after first dose
(date(2026, 5, 1), "Y"), # during treatment
(date(2026, 6, 15), "Y"), # ON last dose
(date(2026, 7, 15), "Y"), # last dose + 30 — boundary, inclusive
(date(2026, 7, 16), "N"), # last dose + 31 — outside
(date(2027, 1, 1), "N"), # long after
],
ids=["long-before", "day-before-start", "on-start", "day-after-start",
"during", "on-end", "end-plus-30", "end-plus-31", "long-after"],
)
def test_treatment_emergent_window(ae_start, expected):
assert derive_trtemfl(ae_start, TRT_START, TRT_END) == expected
# ---- Missing data --------------------------------------------------------
def test_untreated_subject_is_never_treatment_emergent():
assert derive_trtemfl(date(2026, 4, 1), None, None) == "N"
assert derive_trtemfl(None, None, None) == "N"
def test_missing_ae_date_is_conservative_for_treated_subjects():
# An event with an unknown date in a treated subject is assumed emergent.
assert derive_trtemfl(None, TRT_START, TRT_END) == "Y"
def test_ongoing_treatment_has_no_upper_bound():
assert derive_trtemfl(date(2027, 6, 1), TRT_START, None) == "Y"
assert derive_trtemfl(date(2026, 3, 14), TRT_START, None) == "N"
# ---- The window argument -------------------------------------------------
@pytest.mark.parametrize("window,ae_offset,expected", [
(0, 0, "Y"), # window 0: on the last dose date
(0, 1, "N"), # window 0: day after
(7, 7, "Y"),
(7, 8, "N"),
(30, 30, "Y"),
(30, 31, "N"),
(90, 90, "Y"),
])
def test_window_boundary_is_inclusive(window, ae_offset, expected):
from datetime import timedelta
ae = TRT_END + timedelta(days=ae_offset)
assert derive_trtemfl(ae, TRT_START, TRT_END, window=window) == expected
# ---- Validation ----------------------------------------------------------
def test_negative_window_is_rejected():
with pytest.raises(ValueError, match="non-negative"):
derive_trtemfl(date(2026, 4, 1), TRT_START, TRT_END, window=-1)
def test_end_before_start_is_rejected():
with pytest.raises(ValueError, match="precedes"):
derive_trtemfl(date(2026, 4, 1), date(2026, 6, 15), date(2026, 3, 15))
@pytest.mark.parametrize("bad", ["2026-04-01", 20260401, 1.5])
def test_non_date_input_is_rejected(bad):
with pytest.raises(TypeError, match="must be a date"):
derive_trtemfl(bad, TRT_START, TRT_END)
# ---- Contract ------------------------------------------------------------
def test_returns_character_flag_not_boolean():
# The downstream CDISC output requires "Y"/"N", not True/False.
result = derive_trtemfl(date(2026, 4, 1), TRT_START, TRT_END)
assert isinstance(result, str)
assert result in {"Y", "N"}$ pytest tests/test_trtemfl.py -v
test_treatment_emergent_window[long-before] PASSED
test_treatment_emergent_window[on-start] PASSED
test_treatment_emergent_window[end-plus-30] PASSED
test_treatment_emergent_window[end-plus-31] PASSED
test_window_boundary_is_inclusive[0-0-Y] PASSED
...
========================= 27 passed in 0.06s =========================The design points:
Named parametrisation IDs. test_treatment_emergent_window[end-plus-31] tells you exactly which case failed without opening the file. Without ids=, pytest generates [ae_start8-N], which tells you nothing.
The on-start and end-plus-30 cases are the whole point. Everything else is a sanity check; these two encode the boundary conventions that two independent implementations will disagree about. They are the tests that would have caught the differences in the clinical validation lesson.
window=0 is tested. Not a realistic value, but it is the degenerate case that reveals whether the comparison is > or >=, and it costs one parametrised row.
The missing-ae_start behaviour is asserted with an explanatory comment. That is a convention, not a fact — some SAPs would flag it "N". The test documents the choice, so when a reviewer questions it there is a single place that states the decision.
True/False would pass every other test in the file and break the XPT output. That is exactly the class of bug a contract test catches.
Recap
- pytest needs no boilerplate —
test_*.py,test_*(), plainassert pytest.approx()for floats, always- Fixtures in
conftest.pyare available without import;monkeypatchreverses global changes - Parametrise boundary cases with named
ids= - Assert properties rather than whole DataFrames unless the exact output is the contract
- Mock only what you do not control
--cov-report=term-missingnames the untested lines — that is the useful output
Next: Virtual environments.