Generated structure, and the one test the business can read

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, SCD2_PARENT_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, SCD2 parents)."""
    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()],
                {f"{r.table_schema}.{r.table_name}" for r
                 in spark.sql(SCD2_PARENT_QUERY.format(catalog=CATALOG)).collect()})


def pytest_generate_tests(metafunc):
    pks, fks, historised = _declared()
    if "pk" in metafunc.fixturenames:
        metafunc.parametrize("pk", pks, ids=[f"{s}.{t}" for s, t, _ in pks])
    if "fk" in metafunc.fixturenames:
        # A declared FK onto an SCD2 parent is a modelling error, not a test
        # case: the parent's PK is (sk, _tech_valid_from) and a membership test
        # on the sk alone would be green on exactly the rows that are wrong.
        # Fail collection — loudly — instead of generating the weaker test.
        bad = [f for f in fks if f[2] in historised]
        assert not bad, (
            f"FK declared onto a historised parent: {bad}. "
            "Drop the declaration and assert it with the interval-aware "
            "tests/sql/fct_order_line_orphans.sql instead."
        )
        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):
    """MEMBERSHIP only: does the value exist in the parent at all.

    Sound because the generator has already excluded historised parents —
    for those, membership is not the question the report asks.
    """
    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,
         count_if(o.order_id IS NULL)                              AS lines_without_header
  FROM   silver.sales_order_line l
  -- LEFT on BOTH lookups. An inner join to the header is the same trap as an
  -- inner join to the rate: the line disappears from the expected value and
  -- from gold, both sides shrink by the same amount, and delta reads 0,00.
  -- A headerless line has no order_ts, so it lands in a NULL month — which is
  -- exactly what an unattributable line should look like in the output.
  LEFT 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,
       coalesce(s.lines_without_header, 0)                   AS lines_without_header
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
   -- an unresolved lookup makes BOTH sides smaller by the same amount, so the
   -- delta stays zero and the defect is invisible. These are the rows that
   -- show it — one column per lookup that can fail to resolve.
   OR coalesce(s.lines_without_rate, 0) > 0
   OR coalesce(s.lines_without_header, 0) > 0
ORDER BY 1