The chapter that gets violated

from typing import Any

from pyspark.sql.types import StringType, StructField, StructType

# RFC 2606 reserves example.com, so a stray test mail can never reach an inbox.
# The national regulator reserves a number block per area code for drama and
# testing. Both live in one constant each: that is what lets the privacy proof
# below stay two LIKEs instead of a regex nobody maintains.
TEST_DOMAIN = "example.com"
DRAMA_BLOCK = "+49 30 23125"

CUSTOMER_SCHEMA = StructType([
    StructField("customer_id",      StringType(), False),
    StructField("customer_name",    StringType(), False),
    StructField("email",            StringType(), True),
    StructField("phone",            StringType(), True),
    StructField("street",           StringType(), True),
    StructField("postal_code",      StringType(), True),
    StructField("city",             StringType(), True),
    StructField("country",          StringType(), False),
    StructField("customer_segment", StringType(), False),
])

# A boring customer: country in scope, segment set, contact details
# unreachable by construction. Each argument to the factory is a delta from it.
_CUSTOMER: dict[str, Any] = {
    "customer_id":      "C-0001",
    "customer_name":    "Muster Handel GmbH",
    "email":            f"kunde0001@{TEST_DOMAIN}",
    "phone":            f"{DRAMA_BLOCK}001",
    "street":           "Teststrasse 1",
    "postal_code":      "10115",
    "city":             "Musterstadt",
    "country":          "DE",
    "customer_segment": "B",
}


def customer_rows(*overrides: dict[str, Any]) -> list[dict[str, Any]]:
    """One row per argument; each argument is the delta from a valid customer."""
    return [{**_CUSTOMER, **o} for o in (overrides or ({},))]

import random

from tests.factories import CUSTOMER_SCHEMA, DRAMA_BLOCK, TEST_DOMAIN, customer_rows

COUNTRIES = ["DE", "AT", "CH", "NL"]     # business rule 3
SEGMENTS = ["A", "B", "C"]


def seed_customers(spark, catalog: str, n: int, seed: int = 42) -> int:
    """Seed <catalog>.fixtures.customer from the unit tests' own factory.

    Returns rows written. Only ever pointed at dwh_test.
    """
    rnd = random.Random(seed)
    rows = customer_rows(*(
        {
            "customer_id":      f"C-{i:05d}",
            "customer_name":    f"Muster Handel {i:05d} GmbH",
            "email":            f"kunde{i:05d}@{TEST_DOMAIN}",
            "phone":            f"{DRAMA_BLOCK}{i % 1000:03d}",
            "street":           f"Teststrasse {i % 200 + 1}",
            "postal_code":      f"{rnd.randrange(10_000, 99_999)}",
            "country":          rnd.choice(COUNTRIES),
            "customer_segment": rnd.choice(SEGMENTS),
        }
        for i in range(n)
    ))

    # Explicit schema, as everywhere else: inference is how a test ends up
    # exercising a type the warehouse does not have.
    df = spark.createDataFrame(rows, CUSTOMER_SCHEMA)
    df.write.mode("overwrite").saveAsTable(f"{catalog}.fixtures.customer")
    return df.count()

-- Every address in a non-production catalog must sit on the reserved test
-- domain and every number in the regulator's drama block. Anything else is a
-- production copy or a hand-typed real customer, and both are the same finding.
--
-- NULL is a finding too, deliberately. The factory always fills both columns,
-- so a NULL here means either a row nobody generated or a column mask standing
-- between this principal and the data. A mask that returns NULL would
-- otherwise empty the result and let the check pass having read nothing.
--
-- The two literals below repeat TEST_DOMAIN and DRAMA_BLOCK from
-- tests/factories.py. Change one and this goes red, which is the intent.
SELECT 'dwh_dev' AS catalog_name, customer_id, email, phone
FROM dwh_dev.gold.dim_customer
WHERE email IS NULL OR email NOT LIKE '%@example.com'
   OR phone IS NULL OR phone NOT LIKE '+49 30 23125%'
UNION ALL
SELECT 'dwh_test' AS catalog_name, customer_id, email, phone
FROM dwh_test.gold.dim_customer
WHERE email IS NULL OR email NOT LIKE '%@example.com'
   OR phone IS NULL OR phone NOT LIKE '+49 30 23125%'