10Testing · warehouse invariants

from dwh.jobs import silver_customer

# Wall-clock only. If you have to add anything else to this list to make the
# test pass, you have found the bug, not the exception.
VOLATILE = ["_tech_loaded_at"]


def snapshot(spark, table: str) -> list:
    return (
        spark.read.table(table)
        .drop(*VOLATILE)
        .orderBy("customer_id", "_tech_valid_from")
        .collect()                       # collect NOW — see the gotchas
    )


def test_silver_customer_load_is_idempotent(spark, catalog):
    silver_customer.run(spark, catalog)
    first = snapshot(spark, f"{catalog}.silver.customer")

    silver_customer.run(spark, catalog)  # same bronze input, nothing new landed
    second = snapshot(spark, f"{catalog}.silver.customer")

    assert second == first

import pytest
from dwh.jobs import gold_load

TABLES = ["gold.dim_customer", "gold.fct_order_line", "gold.agg_revenue_month"]


def fingerprint(spark, catalog: str) -> dict:
    """Order-independent content hash per table, ignoring load timestamps."""
    out = {}
    for t in TABLES:
        row = spark.sql(f"""
            SELECT count(*)                              AS n,
                   bit_xor(xxhash64(to_json(struct(*)))) AS h
            FROM  (SELECT * EXCEPT (_tech_loaded_at) FROM {catalog}.{t})
        """).first()
        out[t] = (row.n, row.h)
    return out


class Boom(RuntimeError):
    pass


def explode(*_args, **_kwargs):
    raise Boom("simulated executor loss between tables")


def test_retry_after_partial_failure(spark, catalog, monkeypatch):
    gold_load.run(spark, catalog)
    clean = fingerprint(spark, catalog)

    # die AFTER dim_customer is written, BEFORE fct_order_line
    monkeypatch.setattr(gold_load, "build_fct_order_line", explode)
    with pytest.raises(Boom):
        gold_load.run(spark, catalog)
    monkeypatch.undo()

    gold_load.run(spark, catalog)          # the retry, from the top
    assert fingerprint(spark, catalog) == clean

-- vbap_typed is the rename-only view from the ingestion route: bronze keeps
-- SAP's field names (VBELN, POSNR, MATNR, …), the view maps them to canonical
-- ones and passes change_seq and the _tech_ columns through untouched.
MERGE INTO silver.sales_order_line AS t
USING (
  SELECT order_id, line_number, product_id, quantity, unit_price, currency, status
  FROM (
    SELECT *,
           row_number() OVER (
             PARTITION BY order_id, line_number
             -- SAP's change counter, NOT _tech_loaded_at: every row in this
             -- batch shares the load time. See the gotchas.
             ORDER BY     change_seq DESC
           ) AS rn
    FROM vbap_typed
    WHERE _tech_batch_id = :batch_id
  )
  WHERE rn = 1
) AS s
ON  t.order_id    = s.order_id
AND t.line_number = s.line_number
WHEN MATCHED THEN UPDATE SET
  t.product_id      = s.product_id,
  t.quantity        = s.quantity,
  t.unit_price      = s.unit_price,
  t.currency        = s.currency,
  t.status          = s.status,
  t._tech_loaded_at = current_timestamp()
WHEN NOT MATCHED THEN INSERT
  (order_id, line_number, product_id, quantity, unit_price, currency,
   status, _tech_loaded_at)
  VALUES
  (s.order_id, s.line_number, s.product_id, s.quantity, s.unit_price,
   s.currency, s.status, current_timestamp())

import datetime as dt
import pytest
from dwh.common.scd2 import apply_scd2

KEYS = ["customer_id"]
SCHEMA = "customer_id STRING, customer_segment STRING, street STRING"


def T(day: int, hour: int) -> dt.datetime:
    return dt.datetime(2026, 3, day, hour)


def cust(segment="A", street="Bahnhofstr. 1"):
    return {"customer_id": "C1", "customer_segment": segment, "street": street}


CASES = [
    # name              (as_of, extract) in processing order        versions open
    ("no-change",       [(T(1, 6), [cust()]), (T(2, 6), [cust()])],        1, 1),
    ("segment-change",  [(T(1, 6), [cust()]), (T(2, 6), [cust("B")])],     2, 1),
    ("address-only",    [(T(1, 6), [cust()]),
                         (T(2, 6), [cust(street="Domplatz 4")])],          1, 1),
    ("reopen-A-B-A",    [(T(1, 6), [cust()]), (T(2, 6), [cust("B")]),
                         (T(3, 6), [cust()])],                             3, 1),
    ("twice-one-day",   [(T(1, 6), [cust()]), (T(1, 18), [cust("B")])],    2, 1),
    ("late-arrival",    [(T(2, 6), [cust("B")]), (T(1, 6), [cust()])],     2, 1),
    ("deleted",         [(T(1, 6), [cust()]), (T(2, 6), [])],              1, 0),
]


@pytest.mark.parametrize("case,extracts,versions,open_rows", CASES)
def test_scd2_boundary(spark, case, extracts, versions, open_rows):
    state = None
    for as_of, extract in extracts:
        state = apply_scd2(
            state, spark.createDataFrame(extract, SCHEMA), KEYS, as_of=as_of
        )

    rows = sorted(
        state.where("customer_id = 'C1'").collect(),
        key=lambda r: r._tech_valid_from,
    )
    assert len(rows) == versions
    assert sum(1 for r in rows if r._tech_valid_to is None) == open_rows

    # the two assertions that catch every case above, whatever the counts say
    assert all(r._tech_valid_from < (r._tech_valid_to or dt.datetime.max)
               for r in rows)                                    # no zero-length
    assert all(a._tech_valid_to == b._tech_valid_from
               for a, b in zip(rows, rows[1:]))                  # no gap, no overlap

SELECT tc.table_schema,
       tc.table_name,
       array_join(sort_array(collect_list(kcu.column_name)), ', ') AS pk_columns
FROM   {catalog}.information_schema.table_constraints tc
JOIN   {catalog}.information_schema.key_column_usage  kcu
       ON  kcu.constraint_catalog = tc.constraint_catalog
       AND kcu.constraint_schema  = tc.constraint_schema
       AND kcu.constraint_name    = tc.constraint_name
WHERE  tc.constraint_type = 'PRIMARY KEY'
  AND  tc.table_schema IN ('silver', 'gold')
GROUP BY tc.table_schema, tc.table_name
ORDER BY 1, 2

import os
from functools import lru_cache

from dwh.testing.constraints import FK_QUERY, PK_QUERY
from dwh.testing.session import metadata_session   # pytest fixtures do not
                                                   # exist at collection time

# The generator and the tests MUST read the same catalog: the "catalog"
# fixture resolves from this same variable. See the gotchas.
CATALOG = os.environ["DWH_CATALOG"]


@lru_cache(maxsize=1)
def _declared():
    """Read once at collection: (primary keys, foreign keys)."""
    with metadata_session() as spark:
        return ([tuple(r) for r in spark.sql(PK_QUERY.format(catalog=CATALOG)).collect()],
                [tuple(r) for r in spark.sql(FK_QUERY.format(catalog=CATALOG)).collect()])


def pytest_generate_tests(metafunc):
    pks, fks = _declared()
    if "pk" in metafunc.fixturenames:
        metafunc.parametrize("pk", pks, ids=[f"{s}.{t}" for s, t, _ in pks])
    if "fk" in metafunc.fixturenames:
        metafunc.parametrize("fk", fks, ids=[f"{c}.{col}" for c, col, _, _ in fks])


def test_declared_pk_is_unique(spark, catalog, pk):
    schema, table, cols = pk
    dupes = spark.sql(f"""
        SELECT {cols} FROM {catalog}.{schema}.{table}
        GROUP BY {cols} HAVING count(*) > 1
    """).count()
    assert dupes == 0, f"{schema}.{table}: {dupes} duplicate {cols} groups"


def test_declared_fk_has_no_orphans(spark, catalog, fk):
    child, child_col, parent, parent_col = fk
    orphans = spark.sql(f"""
        SELECT c.{child_col}
        FROM   {catalog}.{child} c
        LEFT ANTI JOIN {catalog}.{parent} p ON p.{parent_col} = c.{child_col}
        WHERE  c.{child_col} IS NOT NULL
    """).count()
    assert orphans == 0, f"{child}.{child_col} -> {parent}: {orphans} orphans"


def test_every_gold_table_declares_a_grain(spark, catalog):
    missing = spark.sql(f"""
        SELECT t.table_name
        FROM   {catalog}.information_schema.tables t
        LEFT ANTI JOIN {catalog}.information_schema.table_constraints tc
               ON  tc.table_schema    = t.table_schema
               AND tc.table_name      = t.table_name
               AND tc.constraint_type = 'PRIMARY KEY'
        WHERE  t.table_schema = 'gold'
          -- gold.mv_sales is a metric view and cannot carry a PK
          AND  t.table_type IN ('MANAGED', 'EXTERNAL')
    """).collect()
    assert not missing, f"no declared grain: {[r.table_name for r in missing]}"

WITH silver_revenue AS (
  SELECT date_trunc('MONTH', o.order_ts)                          AS month,
         -- EUR is the base currency and is absent from the ECB file, so it is
         -- 1 by definition. Any OTHER currency with no rate must NOT vanish
         -- into sum()'s NULL-skipping: it is counted on the next line.
         sum(round(l.quantity * l.unit_price *
                   CASE WHEN l.currency = 'EUR' THEN CAST(1 AS DECIMAL(18,6))
                        ELSE f.fx_rate END, 2))                    AS amount_eur,
         count_if(l.currency IS DISTINCT FROM 'EUR'
                  AND f.fx_rate IS NULL)                           AS lines_without_rate
  FROM   silver.sales_order_line l
  JOIN   silver.sales_order      o ON o.order_id  = l.order_id
  LEFT JOIN silver.fx_rate       f ON f.currency  = l.currency
                                  AND f.rate_date = date(o.order_ts)
  WHERE  l.status IS DISTINCT FROM 'CANCELLED'       -- rule 1, NULL-safe
  GROUP BY 1
),
gold_revenue AS (
  SELECT date_trunc('MONTH', order_ts) AS month,
         sum(amount_eur)               AS amount_eur
  FROM   gold.fct_order_line
  WHERE  status IS DISTINCT FROM 'CANCELLED'
  GROUP BY 1
)
SELECT coalesce(s.month, g.month)                            AS month,
       coalesce(s.amount_eur, 0)                             AS silver_eur,
       coalesce(g.amount_eur, 0)                             AS gold_eur,
       coalesce(g.amount_eur, 0) - coalesce(s.amount_eur, 0) AS delta,
       coalesce(s.lines_without_rate, 0)                     AS lines_without_rate
FROM        silver_revenue s
FULL OUTER JOIN gold_revenue g ON g.month = s.month
-- rule 2 rounds per line on both sides from the same three DECIMAL columns, so
-- an exact match is the expectation and one cent is pure arithmetic noise.
-- Anything larger is a difference, not drift.
WHERE abs(coalesce(g.amount_eur, 0) - coalesce(s.amount_eur, 0)) > 0.01
   -- a missing rate makes BOTH sides smaller by the same amount, so the delta
   -- stays zero and the defect is invisible. This is the row that shows it.
   OR coalesce(s.lines_without_rate, 0) > 0
ORDER BY 1