Priya Nair·
Point this at any table schema and it generates the data-quality test suite we should have written months ago. Caught a null-rate regression in staging the same week we adopted it.
Generates a pytest-based data quality suite from a table schema: nulls, uniqueness, referential integrity, distribution drift.
Data Quality Test Suite Generator
You are a data quality engineer. Generate a complete pytest test suite for a warehouse table.
Table schema (DDL or column list with types):
{{table_schema}}
Business rules I know about:
{{business_rules}} (e.g. amounts are never negative, status has 5 allowed values, one row per order per day)
Row count and update cadence: {{table_profile}}
Warehouse and test runner context: {{stack}} (e.g. Snowflake + pytest + a get_connection() fixture that exists already)
Generate tests in four tiers, each a clearly named test class:
1. **Schema contract:** columns exist with expected types; fail loudly listing missing/extra columns. Generate this from the DDL I gave you.
2. **Row-level invariants:** for each column infer sensible checks from name + type + business rules — null rate thresholds (exact 0% only where the rules justify it, otherwise a stated tolerance), allowed value sets for status-like columns, non-negativity for amounts, date sanity (nothing in the future, nothing before company founding {{founding_date}}).
3. **Set-level invariants:** primary key uniqueness, referential integrity to parent tables I mentioned, and the grain check from my business rules (one row per X per Y) written as a GROUP BY HAVING count > 1 query.
4. **Drift monitors (warning, not failure):** row count vs. 7-day trailing average within {{drift_tolerance}}, null-rate delta per column vs. last run, and new unseen values in enum-like columns. These emit warnings via pytest.warns-compatible custom warning, not hard failures.
Conventions: every test docstring states WHY the check exists in business terms. SQL in readable triple-quoted strings, parameterized where repeated across columns. No ORM. Include the conftest snippet for any new fixture you introduce. End with a short README section: how to run only tier 1-3 in CI (drift monitors run on schedule instead) and what to do when each tier fails.
Ergebnisse
## Generated suite: `test_fct_orders.py` (Snowflake, 4 tiers)
### Tier 1 — Schema contract
```python
class TestSchemaContract:
EXPECTED = {
"order_id": "NUMBER", "customer_id": "NUMBER",
"order_date": "DATE", "status": "VARCHAR",
"amount_chf": "NUMBER", "loaded_at": "TIMESTAMP_NTZ",
}
def test_columns_match(self, conn):
"""Downstream dashboards break silently on renamed columns —
fail loudly here instead."""
actual = fetch_columns(conn, "analytics.fct_orders")
missing = set(self.EXPECTED) - set(actual)
extra = set(actual) - set(self.EXPECTED)
assert not missing and not extra, f"missing={missing} extra={extra}"
```
### Tier 2 — Row-level invariants (excerpt)
```python
@pytest.mark.parametrize("column,max_null_pct", [
("order_id", 0.0), # PK — never null
("customer_id", 0.0), # business rule: guest checkout removed 2024
("status", 0.0),
("amount_chf", 0.5), # refund rows may lag amounts by one load
])
def test_null_rate(conn, column, max_null_pct):
"""Null spikes here historically meant a broken upstream join."""
...
def test_status_allowed_values(conn):
"""Exactly 5 states per order lifecycle spec v3."""
sql = '''SELECT DISTINCT status FROM analytics.fct_orders
WHERE status NOT IN ('placed','paid','shipped','delivered','cancelled')'''
assert fetch_all(conn, sql) == []
def test_amount_non_negative(conn):
"""Negative amounts belong in fct_refunds, never here."""
```
Date sanity included: `order_date <= CURRENT_DATE` and `>= '2019-03-01'` (founding).
### Tier 3 — Set-level invariants
```python
def test_grain_one_row_per_order(conn):
"""Grain guarantee: revenue is summed downstream — duplicates
inflate reported revenue directly."""
sql = '''SELECT order_id, COUNT(*) FROM analytics.fct_orders
GROUP BY order_id HAVING COUNT(*) > 1 LIMIT 10'''
dupes = fetch_all(conn, sql)
assert dupes == [], f"duplicate order_ids: {dupes}"
def test_fk_customer_exists(conn):
"""Orphan orders mean the customer dimension load raced this table."""
```
### Tier 4 — Drift monitors (warnings)
```python
class DataDriftWarning(UserWarning): ...
def test_rowcount_drift(conn):
"""Volume drop >20% vs 7-day average usually means a stalled
upstream extract, not a real business change."""
if abs(delta_pct) > 20:
warnings.warn(f"row count {delta_pct:+.1f}% vs 7d avg", DataDriftWarning)
```
New-enum-value and null-delta monitors follow the same pattern.
### conftest addition
```python
@pytest.fixture(scope="session")
def conn():
yield get_connection(role="TESTER_RO") # read-only role, reuses your fixture
```
### README
- CI (every merge): `pytest test_fct_orders.py -m "not drift"` — tiers 1-3, hard gates.
- Schedule (hourly): `pytest -m drift -W error::DataDriftWarning` if you want paging, plain run for Slack digest.
- Tier 1 failure → schema change upstream: update EXPECTED consciously, never blindly. Tier 3 grain failure → stop downstream jobs first, then debug the merge keys.
Modell: GitHub Copilot
6 Likes3 SavesScore: 5
1 Kommentar
Daniel Cohen·
Read-only role on the test fixture — small detail, but it's the difference between a test suite and an incident waiting to happen.