Testing a Regulatory Pipeline — Bad Data, Bad Roles, and What Should Fail

Every data governance book tells you to implement controls. Almost none of them tell you how to prove those controls work under adversarial conditions. For a regulatory pipeline submitted to the ECB, “we have Great Expectations checkpoints” is not an answer. The answer is: here is the test run where we injected a negative CET1 amount and the pipeline refused to proceed.

Day 16 is about building that proof. We will write a full tests/ suite in three categories:

1. Bad data tests — rows that violate GX expectations must raise QualityGateError and quarantine the DAG branch
2. Bad role tests — Ranger-denied queries must return Access Denied, masked columns must return hashed values, not plaintext
3. Pipeline integrity tests — broken XBRL and tampered calculation totals must fail validation with specific error codes

The mental shift: you are not testing the happy path. You are testing that the failure path is reliable. A governance control that silently passes bad data is worse than no control at all — you have false confidence.

Why Testing a Governance Pipeline Is Different

In application testing you assert expected == actual. In governance testing you assert bad_input → specific_error. The taxonomy of what you are testing shifts:

This distinction drives the two-tier test strategy used here: unit tests for GX quality gates and XBRL validation (no live services needed), and integration tests for Ranger security policies (Docker stack required).

Test Directory Structure

conftest.py — Shared Fixtures

The fixtures follow a clear contract: pg_conn gives a real psycopg2 connection for data injection; trino_conn_as is a factory that opens a Trino session as a specific user (so Ranger evaluates the right role); gx_context gives a Great Expectations DataContext pointed at the test database.

# tests/conftest.py
import os
import pytest
import psycopg2
import trino


# ── PostgreSQL connection (admin — for fixture setup / teardown) ────────────
@pytest.fixture(scope="session")
def pg_conn():
    conn = psycopg2.connect(
        host=os.environ.get("POSTGRES_HOST", "localhost"),
        port=os.environ.get("POSTGRES_PORT", "5432"),
        dbname=os.environ.get("POSTGRES_DB", "corep"),
        user=os.environ.get("POSTGRES_USER", "corep_admin"),
        password=os.environ.get("POSTGRES_PASSWORD", "corep_secret"),
    )
    conn.autocommit = True
    yield conn
    conn.close()


# ── Trino connection factory — opens session as a named user ───────────────
@pytest.fixture(scope="session")
def trino_conn_as():
    def _factory(username: str):
        return trino.dbapi.connect(
            host=os.environ.get("TRINO_HOST", "localhost"),
            port=int(os.environ.get("TRINO_PORT", "8080")),
            user=username,            # Ranger evaluates based on this user
            http_scheme="http",
            auth=None,
        )
    return _factory


# ── XBRL fixture paths ─────────────────────────────────────────────────────
@pytest.fixture
def broken_xbrl_path(tmp_path):
    # Truncated XML — parser cannot find closing tag
    p = tmp_path / "broken_instance.xbrl"
    p.write_text("<?xml version='1.0'?><xbrl xmlns='http://www.xbrl.org/2003/instance'")
    return str(p)


@pytest.fixture
def tampered_xbrl_path():
    # Use the pre-built fixture — c0010 ≠ c0020 + c0030 + c0040
    base = os.path.dirname(__file__)
    return os.path.join(base, "fixtures", "tampered_instance.xbrl")


@pytest.fixture
def low_cet1_xbrl_path():
    # CET1 ratio = 3.2 % — below 4.5 % regulatory floor
    base = os.path.dirname(__file__)
    return os.path.join(base, "fixtures", "low_cet1_instance.xbrl")


# ── GX context (unit-test friendly — no live DB required) ─────────────────
@pytest.fixture
def gx_context():
    import great_expectations as ge
    return ge.get_context(context_root_dir="gx/")

Category 1 — Bad Data Tests

The GX checkpoints sit at two layers: raw (structural validation immediately after ingest) and mart (business-rule validation after dbt transforms). Both layers must reject specific categories of bad data. The approach: inject a known-bad row directly into PostgreSQL, run the quality module, assert QualityGateError is raised.

Fixture: bad_capital_instruments.csv

# tests/fixtures/bad_capital_instruments.csv
# Every row here violates at least one GX expectation
instrument_id,reporting_date,tier,amount_eur,currency,issuer_id
# Row 1: negative amount — violates expect_column_values_to_be_between(min=0)
CI_BAD_001,2025-12-31,CET1,-500000,EUR,BANK_01
# Row 2: invalid tier code — violates expect_column_values_to_be_in_set
CI_BAD_002,2025-12-31,INVALID_TIER,1000000,EUR,BANK_01
# Row 3: NULL reporting_date — violates expect_column_values_to_not_be_null
CI_BAD_003,,CET1,750000,EUR,BANK_01
# Row 4: amount = 0 with NULL currency — two violations
CI_BAD_004,2025-12-31,AT1,0,,BANK_01

test_quality_gates.py

"""
tests/test_quality_gates.py
Test that Great Expectations checkpoints reject bad data as expected.
These are unit-level integration tests — they need PostgreSQL but not Trino/Ranger.
Run with:  pytest tests/test_quality_gates.py -v
"""
import pytest
import pandas as pd
from modules.quality import QualityModule, QualityGateError


# ── Helper: inject a DataFrame into a raw table and truncate after test ────
def _inject_rows(conn, table: str, df: pd.DataFrame):
    from sqlalchemy import create_engine
    import os
    url = os.environ.get("COREP_DB_URL", "postgresql+psycopg2://corep_admin:corep_secret@localhost:5432/corep")
    engine = create_engine(url)
    df.to_sql(table, con=engine, schema="raw", if_exists="append", index=False)
    engine.dispose()


def _truncate_table(conn, table: str):
    cur = conn.cursor()
    cur.execute(f"TRUNCATE raw.{table}")


# ── Raw Layer Gate ─────────────────────────────────────────────────────────
class TestRawQualityGate:

    def test_negative_capital_amount_fails_gate(self, pg_conn):
        """
        A capital instrument with amount_eur < 0 must trigger the raw GX gate.
        Expectation: expect_column_values_to_be_between(column='amount_eur', min_value=0)
        """
        bad_row = pd.DataFrame([{
            "instrument_id": "CI_BAD_NEG",
            "reporting_date": "2025-12-31",
            "tier": "CET1",
            "amount_eur": -500_000,
            "currency": "EUR",
            "issuer_id": "BANK_01",
        }])
        _inject_rows(pg_conn, "capital_instruments", bad_row)
        try:
            with pytest.raises(QualityGateError, match=r"raw.*capital"):
                QualityModule(layer="raw").run()
        finally:
            _truncate_table(pg_conn, "capital_instruments")

    def test_invalid_tier_code_fails_gate(self, pg_conn):
        """
        tier must be in {'CET1', 'AT1', 'T2'}.
        Expectation: expect_column_values_to_be_in_set
        """
        bad_row = pd.DataFrame([{
            "instrument_id": "CI_BAD_TIER",
            "reporting_date": "2025-12-31",
            "tier": "STRUCTURED_NOTE",   # not a valid Basel III tier
            "amount_eur": 1_000_000,
            "currency": "EUR",
            "issuer_id": "BANK_01",
        }])
        _inject_rows(pg_conn, "capital_instruments", bad_row)
        try:
            with pytest.raises(QualityGateError):
                QualityModule(layer="raw").run()
        finally:
            _truncate_table(pg_conn, "capital_instruments")

    def test_null_reporting_date_fails_gate(self, pg_conn):
        """
        reporting_date is required (NOT NULL).
        Expectation: expect_column_values_to_not_be_null
        """
        bad_row = pd.DataFrame([{
            "instrument_id": "CI_BAD_NULL",
            "reporting_date": None,
            "tier": "CET1",
            "amount_eur": 500_000,
            "currency": "EUR",
            "issuer_id": "BANK_01",
        }])
        _inject_rows(pg_conn, "capital_instruments", bad_row)
        try:
            with pytest.raises(QualityGateError):
                QualityModule(layer="raw").run()
        finally:
            _truncate_table(pg_conn, "capital_instruments")

    def test_null_exposure_class_rwa_fails_gate(self, pg_conn):
        """
        exposure_class is required on rwa_exposures.
        A NULL breaks the dbt grouping and must be caught before dbt runs.
        """
        bad_row = pd.DataFrame([{
            "exposure_id": "RWA_BAD_001",
            "reporting_date": "2025-12-31",
            "exposure_class": None,   # NULL — will break int_rwa_by_exposure_class
            "exposure_amount_eur": 2_000_000,
            "rw_percentage": 75.0,
        }])
        _inject_rows(pg_conn, "rwa_exposures", bad_row)
        try:
            with pytest.raises(QualityGateError):
                QualityModule(layer="raw").run()
        finally:
            _truncate_table(pg_conn, "rwa_exposures")


# ── Mart Layer Gate ────────────────────────────────────────────────────────
class TestMartQualityGate:

    def test_cet1_ratio_below_floor_fails_mart_gate(self, pg_conn):
        """
        CET1 ratio < 4.5 % violates the EBA Pillar 1 floor.
        The mart GX suite checks: expect_column_values_to_be_between(
            column='cet1_ratio', min_value=0.045)
        NOTE: This test also validates that the pipeline does not submit
        data that would fail the ECB's automated plausibility checks.
        """
        cur = pg_conn.cursor()
        # Temporarily overwrite a mart row with a below-floor ratio
        cur.execute("""
            INSERT INTO mart.corep_c0300
                (reporting_date, cet1_ratio, t1_ratio, total_capital_ratio)
            VALUES ('2025-12-31', 0.032, 0.045, 0.080)
            ON CONFLICT (reporting_date) DO UPDATE
              SET cet1_ratio = EXCLUDED.cet1_ratio
        """)
        try:
            with pytest.raises(QualityGateError, match=r"mart.*cet1"):
                QualityModule(layer="mart").run()
        finally:
            cur.execute("DELETE FROM mart.corep_c0300 WHERE cet1_ratio = 0.032")

Category 2 — Bad Role Tests (Ranger Security)

Ranger enforces access control at query time through the Trino plugin. The key insight: Ranger evaluates the Trino user field, not a password or token. In development, passing a username to the Trino DBAPI connector is sufficient to exercise Ranger policies without setting up full Kerberos or LDAP.

"""
tests/test_security_policies.py
Integration tests — REQUIRE the full Docker stack (Trino + Ranger + PostgreSQL).
Run with:  pytest tests/test_security_policies.py -v -m integration
"""
import pytest
import trino.exceptions


@pytest.mark.integration
class TestRawSchemaAccess:
    """raw.* schema is off-limits for all roles except data_engineer and pipeline_service."""

    def test_risk_analyst_cannot_read_raw_counterparties(self, trino_conn_as):
        conn = trino_conn_as("risk_analyst")
        cur = conn.cursor()
        with pytest.raises(
            trino.exceptions.TrinoUserError,
            match=r"Access Denied"
        ):
            cur.execute("SELECT name FROM postgresql.raw.counterparties LIMIT 1")
            cur.fetchall()   # Ranger denial surfaces on fetch, not execute

    def test_corep_reporting_cannot_read_raw_schema(self, trino_conn_as):
        conn = trino_conn_as("corep_reporting")
        cur = conn.cursor()
        with pytest.raises(trino.exceptions.TrinoUserError, match=r"Access Denied"):
            cur.execute("SELECT * FROM postgresql.raw.capital_instruments LIMIT 1")
            cur.fetchall()

    def test_unknown_user_denied_everywhere(self, trino_conn_as):
        conn = trino_conn_as("ghost_user_not_in_any_policy")
        cur = conn.cursor()
        with pytest.raises(trino.exceptions.TrinoUserError, match=r"Access Denied"):
            cur.execute("SELECT * FROM postgresql.mart.corep_c0300 LIMIT 1")
            cur.fetchall()


@pytest.mark.integration
class TestColumnMasking:
    """
    Ranger column masks: corep_reporting sees SHA256(lei), not plaintext.
    The mask policy uses: MASK_HASH on raw.counterparties.lei
    """

    def test_corep_reporting_sees_hashed_lei(self, trino_conn_as):
        # corep_reporting has row-level access to mart only,
        # but to test the mask we need a role that can see raw but gets masked.
        # The data_engineer role can read raw AND has the mask applied.
        conn = trino_conn_as("data_engineer")
        cur = conn.cursor()
        cur.execute("SELECT lei FROM postgresql.raw.counterparties LIMIT 1")
        rows = cur.fetchall()
        assert rows, "Expected at least one row in raw.counterparties"
        lei_value = rows[0][0]
        # Ranger MASK_HASH produces a 64-char hex string (SHA-256)
        assert len(lei_value) == 64, f"Expected SHA-256 hash (64 chars), got: {lei_value!r}"
        assert all(c in "0123456789abcdef" for c in lei_value.lower()), \
            "Not a hex string — Ranger mask may not be applied"

    def test_pipeline_service_sees_plaintext_lei(self, trino_conn_as):
        """pipeline_service is exempt from the mask — it writes data, needs plaintext."""
        conn = trino_conn_as("pipeline_service")
        cur = conn.cursor()
        cur.execute("SELECT lei FROM postgresql.raw.counterparties LIMIT 1")
        rows = cur.fetchall()
        assert rows
        lei_value = rows[0][0]
        # Real LEI is 20 chars, not a 64-char hash
        assert len(lei_value) != 64, "pipeline_service should see plaintext, not hash"


@pytest.mark.integration
class TestDMLRestrictions:
    """Only pipeline_service may write to the mart schema."""

    def test_corep_reporting_cannot_insert_into_mart(self, trino_conn_as):
        conn = trino_conn_as("corep_reporting")
        cur = conn.cursor()
        with pytest.raises(trino.exceptions.TrinoUserError, match=r"Access Denied"):
            cur.execute("""
                INSERT INTO postgresql.mart.corep_c0300
                    (reporting_date, cet1_ratio, t1_ratio, total_capital_ratio)
                VALUES ('2099-12-31', 0.15, 0.16, 0.17)
            """)
            cur.fetchall()

    def test_risk_analyst_cannot_delete_from_mart(self, trino_conn_as):
        conn = trino_conn_as("risk_analyst")
        cur = conn.cursor()
        with pytest.raises(trino.exceptions.TrinoUserError, match=r"Access Denied"):
            cur.execute("DELETE FROM postgresql.mart.corep_c0100")
            cur.fetchall()

    def test_pipeline_service_can_insert_mart(self, trino_conn_as, pg_conn):
        """Positive test — pipeline_service must succeed or pipeline itself breaks."""
        conn = trino_conn_as("pipeline_service")
        cur = conn.cursor()
        try:
            cur.execute("""
                INSERT INTO postgresql.mart.corep_c0300
                    (reporting_date, cet1_ratio, t1_ratio, total_capital_ratio)
                VALUES ('2099-06-30', 0.15, 0.17, 0.19)
            """)
            cur.fetchall()
        finally:
            pg_conn.cursor().execute(
                "DELETE FROM mart.corep_c0300 WHERE reporting_date = '2099-06-30'"
            )

Category 3 — Pipeline Integrity Tests (XBRL Validation)

The XBRL validation layer has three distinct failure modes, each caught by a different Arelle validation pass:

"""
tests/test_xbrl_validation.py
Unit tests — these do not need a live Docker stack.
They call validate_xbrl_instance() directly with prepared fixture files.
Run with:  pytest tests/test_xbrl_validation.py -v
"""
import pytest
from xbrl.validate_instance import validate_xbrl_instance
from modules.xbrl_valid import XbrlValidModule, XbrlValidationError


class TestStructuralValidation:

    def test_truncated_xml_fails_structural_pass(self, broken_xbrl_path):
        """
        A file that is cut off mid-element cannot be parsed by lxml.
        Arelle surfaces this as an xml:syntaxError or similar structural error.
        The validate_xbrl_instance function must return passed=False.
        """
        result = validate_xbrl_instance(broken_xbrl_path)
        assert not result.passed, "Broken XBRL should fail validation"
        assert result.errors, "Should have at least one error message"
        error_codes = [m.code.lower() for m in result.errors]
        assert any(
            "syntax" in c or "parse" in c or "error" in c
            for c in error_codes
        ), f"Expected structural error, got codes: {error_codes}"

    def test_xbrl_valid_module_raises_on_broken_file(self, broken_xbrl_path, monkeypatch):
        """
        XbrlValidModule.run() must raise XbrlValidationError when the
        underlying validator returns passed=False.
        """
        monkeypatch.setenv("XBRL_OUTPUT_DIR", str(broken_xbrl_path.parent))
        with pytest.raises(XbrlValidationError):
            XbrlValidModule().run()


class TestCalculationLinkbase:

    def test_tampered_totals_trigger_inconsistency(self, tampered_xbrl_path):
        """
        tampered_instance.xbrl has c0010 (Own Funds) set to a value that does
        NOT equal c0020 (CET1) + c0030 (AT1) + c0040 (T2).
        EBA calculation linkbase must catch this as xbrlCalcs:inconsistency.
        """
        result = validate_xbrl_instance(tampered_xbrl_path)
        assert not result.passed
        calc_errors = [
            m for m in result.errors
            if "xbrlCalcs:inconsistency" in m.code
        ]
        assert calc_errors, (
            "Expected xbrlCalcs:inconsistency error for tampered totals. "
            f"Got errors: {[m.code for m in result.errors]}"
        )
        # Verify the inconsistency is on Own Funds (c0010)
        c0010_errors = [m for m in calc_errors if "c0010" in m.message.lower()]
        assert c0010_errors, "c0010 (Own Funds) should be the inconsistent concept"

    def test_rounding_tolerance_not_treated_as_error(self, tmp_path):
        """
        xbrlCalcs:insignificantRounding is a WARNING, not an error.
        A difference of 1 thousand EUR (within decimals=-3 tolerance) must NOT
        cause result.passed to be False.
        This test guards against over-strict validation that would reject
        instances Arelle itself considers valid.
        """
        import shutil, os
        good_instance = os.path.join("output", "xbrl",
                                       "corep_c0100_2025-12-31.xbrl")
        if not os.path.exists(good_instance):
            pytest.skip("No generated XBRL instance available — run pipeline first")
        result = validate_xbrl_instance(good_instance)
        rounding_only_errors = [
            m for m in result.errors
            if "insignificantRounding" not in m.code
        ]
        assert not rounding_only_errors, \
            f"Non-rounding errors found in good instance: {rounding_only_errors}"


class TestFormulaLinkbase:

    def test_cet1_below_minimum_triggers_formula_assertion(self, low_cet1_xbrl_path):
        """
        EBA formula linkbase contains an assertion that CET1 ratio ≥ 4.5%.
        An instance with CET1 = 3.2% must fail this assertion.
        The error message will reference the EBA assertion concept ID.
        """
        result = validate_xbrl_instance(low_cet1_xbrl_path)
        assert not result.passed
        formula_errors = [
            m for m in result.errors
            if "formula" in m.code.lower() or "assertion" in m.code.lower()
        ]
        assert formula_errors, (
            "Expected formula linkbase assertion failure for CET1 = 3.2%. "
            f"Got codes: {[m.code for m in result.errors]}"
        )

Category 4 — DAG Branch Routing Tests

The Airflow DAG uses BranchPythonOperator to route to a quarantine task when quality or validation fails. These tests verify the routing logic itself — without needing a running Airflow instance — by calling the branch functions directly with synthetic XCom returns.

"""
tests/test_pipeline_branches.py
Test DAG branch routing logic in isolation (no Airflow required).
The branch functions are extracted from the DAG and testable as pure Python.
"""
import pytest
from unittest.mock import MagicMock


# ── Import branch logic from the DAG module ────────────────────────────────
from dags.corep_pipeline_dag import (
    _branch_quality_layer1,
    _branch_quality_layer2,
    _branch_xbrl_valid,
)


def _mock_ti(xcom_value: str) -> MagicMock:
    """Create a mock TaskInstance that returns xcom_value from xcom_pull."""
    ti = MagicMock()
    ti.xcom_pull.return_value = xcom_value
    return ti


class TestQualityBranchRouting:

    def test_quality_layer1_pass_routes_to_dbt(self):
        ti = _mock_ti("PASS")
        result = _branch_quality_layer1(ti=ti)
        assert result == "run_dbt"

    def test_quality_layer1_fail_routes_to_quarantine(self):
        ti = _mock_ti("FAIL")
        result = _branch_quality_layer1(ti=ti)
        assert result == "quarantine_raw_failure"

    def test_quality_layer1_none_xcom_routes_to_quarantine(self):
        """None from xcom_pull (skipped task) should be treated as failure."""
        ti = _mock_ti(None)
        result = _branch_quality_layer1(ti=ti)
        assert result == "quarantine_raw_failure"

    def test_quality_layer2_pass_routes_to_catalog(self):
        ti = _mock_ti("PASS")
        result = _branch_quality_layer2(ti=ti)
        assert result == "run_catalog"

    def test_xbrl_valid_fail_routes_to_quarantine(self):
        ti = _mock_ti("FAIL")
        result = _branch_xbrl_valid(ti=ti)
        assert result == "quarantine_xbrl_failure"

    def test_xbrl_valid_pass_routes_to_submission(self):
        ti = _mock_ti("PASS")
        result = _branch_xbrl_valid(ti=ti)
        assert result == "build_submission_package"

Running Tests in CI — Two-Tier Strategy

pytest.ini — Markers and Settings

# pytest.ini
[pytest]
testpaths = tests
markers =
    integration: marks tests that require the full Docker stack (deselect with -m "not integration")
log_cli = true
log_cli_level = INFO
addopts = --tb=short -q

# Run unit tests only (CI Stage 1)
pytest -m "not integration" -v

# Run all tests including security integration tests
pytest -m "integration" --timeout=120 -v

# Run a specific failure scenario
pytest tests/test_xbrl_validation.py::TestCalculationLinkbase::test_tampered_totals_trigger_inconsistency -v -s

Building the XBRL Fixture Files

The tampered XBRL fixture cannot be a random file — it must be a structurally valid XBRL instance that passes structural validation but fails the calculation linkbase. The script below generates it from your existing mart data and then deliberately corrupts the Own Funds total:

# tests/fixtures/build_tampered_xbrl.py
# Run once to generate tests/fixtures/tampered_instance.xbrl
# Requires: a passing XBRL instance already in output/xbrl/

import os, shutil
from lxml import etree

SOURCE = "output/xbrl/corep_c0100_2025-12-31.xbrl"
DEST   = "tests/fixtures/tampered_instance.xbrl"

# EBA namespace and concept for Own Funds (c0010)
EBA_NS = "http://www.eba.europa.eu/xbrl/crr/dict/met"
OWN_FUNDS_CONCEPT = "{%s}ei:OwnFunds" % EBA_NS

shutil.copy(SOURCE, DEST)
tree = etree.parse(DEST)
root = tree.getroot()

# Find the Own Funds fact and add 1 billion to it (breaks calculation sum)
for elem in root.iter():
    if "OwnFunds" in elem.tag and elem.text:
        original = int(elem.text)
        elem.text = str(original + 1_000_000)   # +1 billion EUR (×1000 because decimals=-3)
        print(f"Tampered OwnFunds: {original} → {elem.text}")
        break

tree.write(DEST, xml_declaration=True, encoding="UTF-8")
print(f"Written: {DEST}")

The Complete “What Should Fail” Reference Table

The “What Should Pass” Checks (Positive Test Coverage)

Adversarial testing is not complete without positive tests. You must confirm that legitimate users doing legitimate things are not blocked. A Ranger policy that is too broad will deny your own pipeline.

Three Anti-Patterns to Avoid

1. Mocking Ranger in Security Tests

If you mock the Trino connection in security tests, you are not testing Ranger — you are testing your mock. The value of security tests is that they prove Ranger’s policy evaluation works end-to-end. Use the real Trino connector, accept the Docker dependency.

2. Leaving Poison Rows After Test Failures

A test that injects bad data and then fails its own assertion before cleanup leaves bad rows in the database. The next test run starts from a dirty state. Always use try/finally for cleanup, never teardown_method alone.

3. Testing at the Wrong Layer

Testing that dbt correctly computes cet1_ratio is a transformation test. Testing that GX catches a CET1 ratio below 4.5% is a governance control test. They are different tests and both are necessary. Many teams skip the governance control test because the transformation test passes — then they submit a 3% CET1 report to the ECB.

Tomorrow, Day 17: the capstone run — starting from a cold Docker stack, triggering the Airflow DAG, and tracing the complete path from raw CSV to XBRL submission ZIP.