Writing pytest Assertions for k-Anonymity Guarantees

Three assertions cover the guarantee: every published cell holds at least $k$ distinct individuals, no suppressed cell appears in the output in any form, and the released columns contain no identifier. Each needs a paired negative control, because a check that never rejects anything is indistinguishable from a check that is not wired up.

Core Specification # permalink

The property under test is

where $R$ is the released cell set and $U(c)$ the set of distinct individuals in cell $c$. Two failure modes make this harder to test than it looks.

The unit of counting. len(group) counts observations. A single device contributing forty pings to one cell satisfies a k of 20 while one person stands there. Every assertion below counts nunique() on the person column, and the fixture deliberately contains such a device so the test would catch a regression to row counts.

The absence of the suppressed set. Withheld cells must not appear as null rows, as a suppressed flag, or as a gap that can be inferred from a published complete cell list. The assertion is over the shape of the output, not only its values.

The third assertion — no identifier columns — is trivial and catches the most common real regression, which is a join that reintroduces device_id for debugging and is never removed.

Worked example of the fixture # permalink

The fixture is the interesting part of the module. A clean synthetic frame passes every check and proves nothing; the fixture has to contain the cases that fail in production:

A suite whose fixture lacks c-02 will pass with a row-count implementation, and that is precisely the regression the suite exists to prevent.

Python Implementation # permalink

"""test_k_anonymity.py — blocking checks for a k-anonymous spatial release."""
from __future__ import annotations

import pandas as pd
import pytest

K = 20
IDENTIFIER_COLUMNS = {"device_id", "person_id", "uid", "imei", "advertising_id"}

@pytest.fixture
def raw() -> pd.DataFrame:
    """Deliberately contains the cases that fail in production, not a clean sample."""
    rows = []
    rows += [{"cell_id": "c-01", "person_id": f"p{i%31}"} for i in range(44)]
    rows += [{"cell_id": "c-02", "person_id": "p999"} for _ in range(40)]  # one device
    rows += [{"cell_id": "c-03", "person_id": f"q{i}"} for i in range(12)]
    rows += [{"cell_id": "c-04", "person_id": f"r{i%20}"} for i in range(26)]
    return pd.DataFrame(rows)

def released(raw: pd.DataFrame, k: int = K) -> pd.DataFrame:
    """The transform under test: publish cells clearing k, drop the rest entirely."""
    counts = raw.groupby("cell_id")["person_id"].nunique().rename("n_persons")
    keep = counts[counts >= k]
    return keep.reset_index()          # no null rows, no suppressed flag

def test_every_published_cell_clears_k(raw):
    out = released(raw)
    assert not out.empty, "the release is empty — the transform or fixture is wrong"
    worst = int(out["n_persons"].min())
    assert worst >= K, f"cell {out.loc[out['n_persons'].idxmin(), 'cell_id']} holds {worst}"

def test_counts_are_over_people_not_rows(raw):
    """c-02 has 40 rows and one person; a row-count implementation publishes it."""
    out = released(raw)
    assert "c-02" not in set(out["cell_id"]), "a single-person cell was published"

def test_suppressed_cells_are_absent_not_nulled(raw):
    out = released(raw)
    assert "c-03" not in set(out["cell_id"]), "a sub-threshold cell appears in the output"
    assert out["n_persons"].notna().all(), "a null count leaks which cells were withheld"
    assert not (set(out.columns) & {"suppressed", "is_suppressed", "withheld"}), \
        "a suppression flag maps the sparse cells"

def test_no_identifier_columns_survive(raw):
    out = released(raw)
    leaked = set(out.columns) & IDENTIFIER_COLUMNS
    assert not leaked, f"identifier columns in the release: {sorted(leaked)}"

Verification: the negative controls # permalink

Each assertion above needs a companion that proves it can fail. Without them, a refactor that disconnects released() from the real pipeline leaves the suite green.

def test_control_k_gate_rejects_a_thin_release():
    thin = pd.DataFrame({"cell_id": ["x"], "n_persons": [3]})
    with pytest.raises(AssertionError):
        worst = int(thin["n_persons"].min())
        assert worst >= K, "expected the k gate to reject a 3-person cell"

def test_control_identifier_gate_rejects_a_leak():
    leaky = pd.DataFrame({"cell_id": ["x"], "n_persons": [40], "device_id": ["d1"]})
    with pytest.raises(AssertionError):
        leaked = set(leaky.columns) & IDENTIFIER_COLUMNS
        assert not leaked, "expected the identifier gate to reject device_id"

def test_control_boundary_is_inclusive(raw):
    """c-04 holds exactly K people and must be published — an off-by-one here
    silently removes every borderline cell from every release."""
    assert "c-04" in set(released(raw)["cell_id"])

Edge Cases & Adjustments # permalink

• The boundary. >= k and > k differ by one cell class and the difference is invisible in aggregate metrics. Pin it with the c-04 control and state the convention in the release record.
• Structural zeros. A cell with no residents is not a small cell. Exclude true zeros from the suppression rule and test that they survive, or a map of parkland will show holes where the parks are.
• Devices standing in for people. Where identity cannot be resolved, keep the same tests but rename the column and record that the floor is over devices. The assertion is unchanged; the claim it supports is weaker.
• Very large releases. nunique() over a hundred million rows is expensive. Compute the per-cell distinct counts once in the pipeline and assert over that artefact, rather than recomputing in the test.
• Multiple published keys. If the release publishes cell crossed with a time bucket, the floor applies to the crossed key. Testing geography alone passes a release whose real groups are much smaller.

Keeping the Assertions Fast # permalink

On a release of any size, nunique() over the raw frame is the slowest thing in the job, and a slow gate is a gate somebody eventually moves out of the critical path.

The fix is to compute the per-cell distinct counts once, in the pipeline, and have the assertion read that artefact rather than recomputing it. The count becomes a column of the release itself, which has the additional benefit of making the guarantee inspectable: a reviewer can see the minimum without running anything. Where the release must not carry counts, write them to a side artefact that the audit record references.

FAQ # permalink

Should these run on production data or a fixture?

Both, at different stages. The fixture-based tests run on every pull request and protect the logic; the same assertions run inside the release job over the real artefact and protect the publication. The CI topic page sets out which belongs where.

What if a legitimate release fails the k gate?

Coarsen the geometry or raise the suppression, and re-run. Lowering $k$ requires a change to the policy file and a review, which is exactly the friction that should exist.

Is pytest.raises the right way to write a control?

It is the clearest. The alternative — asserting on a returned boolean — passes when the function silently returns False for an unrelated reason. Raising makes the control specific about which assertion fired.

How do I test complementary suppression?

Not with these assertions. Suppression completeness is a property of the published totals and needs the linear-programming check described in applying complementary suppression to choropleth maps.

Related # permalink

• Automating Spatial Privacy Checks in CI — where each check has to run to block anything
• Gating Releases on a Re-identification Risk Budget — the risk-based gate these thresholds sit under
• k-Anonymity Grouping for Location Traces — the guarantee under test
• Setting Minimum Count Thresholds for Published Map Cells — choosing the k the tests enforce

← Back to Automating Spatial Privacy Checks in CI