20Written concepts

-- The role matrix from chapter 3, checked into the repo as a VALUES list.
-- Symmetric EXCEPT: one direction finds grants nobody declared, the other
-- finds declarations nobody granted. One direction alone catches half.
--
-- Run this as a principal that can see every grant in the catalog.
-- information_schema is filtered to the objects the caller has privileges on,
-- so a narrowly-granted service principal reads an empty actual set, both
-- EXCEPTs collapse, and the check goes green having compared nothing.
WITH concept(grantee, table_name, privilege_type) AS (
  VALUES
    ('grp_dwh_bi_reader', 'mv_sales',          'SELECT'),
    ('grp_dwh_bi_reader', 'agg_revenue_month', 'SELECT'),
    ('grp_dwh_analyst',   'fct_order_line',    'SELECT'),
    ('grp_dwh_analyst',   'dim_customer',      'SELECT'),
    ('grp_dwh_engineer',  'fct_order_line',    'MODIFY')
),
actual AS (
  SELECT grantee, table_name, privilege_type
  FROM dwh_prod.information_schema.table_privileges
  WHERE table_schema = 'gold'
    AND grantee <> 'grp_dwh_owner'      -- ownership grants are implicit
)
SELECT 'undeclared' AS problem, * FROM (SELECT * FROM actual  EXCEPT SELECT * FROM concept)
UNION ALL
SELECT 'missing'    AS problem, * FROM (SELECT * FROM concept EXCEPT SELECT * FROM actual)

---
title:         Berechtigungskonzept - dwh_prod
owner:         A. Weber (Data Owner, Vertrieb)   # a named person, never a group
authors:       lead engineer + platform
last_reviewed: 2026-07-14
review_cycle:  90d
status:        active                      # draft | active | superseded
proof:         tests/access/grants_match_concept.sql
applies_to:    dwh_prod.gold, dwh_prod.silver
---

import re
from datetime import date, timedelta
from pathlib import Path

import pytest
import yaml

CONCEPTS = sorted(Path("docs/concepts").glob("*.md"))


def test_concepts_are_found() -> None:
    """An empty parametrize list reports as skipped, not failed. Without this
    line the whole check passes vacuously the day the path or the CWD moves."""
    assert CONCEPTS, "docs/concepts/ is empty - nothing was checked"


@pytest.mark.parametrize("path", CONCEPTS, ids=lambda p: p.name)
def test_header_is_current(path: Path) -> None:
    front = re.match(r"^---\n(.*?)\n---", path.read_text(encoding="utf-8"), re.S)
    assert front, f"{path.name}: no header block"
    head = yaml.safe_load(front.group(1))

    if head["status"] != "active":
        return

    cycle = timedelta(days=int(head["review_cycle"].rstrip("d")))
    assert head["last_reviewed"] + cycle >= date.today(), (
        f"{path.name}: last reviewed {head['last_reviewed']}, cycle {head['review_cycle']}"
    )
    assert Path(head["proof"]).exists(), f"{path.name}: proof {head['proof']} is missing"

owner:     lead engineer
reviewers: two engineers + platform/architecture
timebox:   3 days initial, half a day per quarter
proof:     tests/standards/

1  Scope and non-goals
2  Architecture
   2.1  Layer contract: what may enter and leave bronze / silver / gold
   2.2  Topology: dwh_dev / dwh_test / dwh_prod, layers as schemas
3  Modelling
   3.1  Star schema in gold; silver stays source-shaped
   3.2  Surrogate keys: sha2 over normalised business keys
   3.3  SCD2 interval semantics, exclusive end, the UNKNOWN member
4  Naming
   4.1  Prefixes: dim_ fct_ agg_ brg_ stg_ mv_ _tech_
   4.2  ^[a-z][a-z0-9_]*$, ASCII only; reserved words (sales_order, not order)
5  Code structure
   5.1  The transform is a pure function; IO at the edges
   5.2  Module layout under src/dwh/
   5.3  Workspace files and the sys.path bootstrap — no wheel
6  Testing
   6.1  The four kinds: PySpark unit, SQL assertion, pipeline, warehouse invariant
   6.2  What every merge runs; what runs nightly
7  Delivery
   7.1  Bundle targets dev / test / prod; bundle validate first in CI
   7.2  Branching, review rules, OAuth service principals
8  Definition of done
9  Not testable: on-call rota, escalation, team topology

owner:     data owner (Vertrieb) - a named person, drafted by the lead engineer
reviewers: Security/IT, Datenschutzbeauftragter
timebox:   2 days initial, reviewed on every new schema
proof:     tests/access/

1  Scope: which catalogs, and what is explicitly out of scope
2  Principals
   2.1  Groups only - no grant to a named user, ever
   2.2  One service principal per job; OAuth, never personal access tokens
   2.3  Group membership: source of truth in the IdP, who approves a join
3  Role matrix
   3.1  grp_dwh_engineer / grp_dwh_analyst / grp_dwh_bi_reader / grp_dwh_owner
   3.2  gold granted per object; silver by exception; bronze engineers only
4  Attribute-based access
   4.1  Governed tags: pii, classification, domain, environment
   4.2  Column mask policies attached to tags, not to tables
   4.3  Row filter on country, driven by the group's region attribute
5  Environment isolation
   5.1  Workspace-catalog binding: a dev workspace cannot reach dwh_prod
   5.2  Who may create external locations (a very short list)
6  Ownership: production objects owned by groups, never individuals
7  Review
   7.1  Quarterly access review over pii-tagged tables - the query, and who signs
   7.2  Alerting on permission changes in dwh_prod
8  Exceptions register: who, what, and until when

-- "email, phone and street are personal data" - that sentence, as three
-- statements. The mask policy hangs off the tag, so a column added to any
-- table next quarter is protected the moment it is tagged.
--
-- Precondition: pii must already exist as a GOVERNED tag defined at account
-- level. Run these before that and they still succeed - they just create an
-- ordinary free-form tag, which no ABAC policy can attach to. The columns
-- then look tagged in Catalog Explorer and are masked by nothing.
ALTER TABLE dwh_prod.gold.dim_customer ALTER COLUMN email  SET TAGS ('pii' = 'true');
ALTER TABLE dwh_prod.gold.dim_customer ALTER COLUMN phone  SET TAGS ('pii' = 'true');
ALTER TABLE dwh_prod.gold.dim_customer ALTER COLUMN street SET TAGS ('pii' = 'true');

owner:     Datenschutzbeauftragter
reviewers: lead engineer, legal
timebox:   2 days initial, reviewed on every new PII column
proof:     tests/privacy/

1  Legal basis and scope: which processing, which catalogs
2  Inventory of personal data
   2.1  gold.dim_customer.email / .phone / .street - directly identifying
   2.2  country + postal_code + customer_segment - quasi-identifiers
   2.3  Keeping the inventory current: automated classification + governed tags
3  Techniques and what each one legally achieves
   3.1  Pseudonymisation: hashing, tokenisation, masking - still personal data
   3.2  Anonymisation: aggregation above a k-threshold, generalisation, suppression
   3.3  Where each is used in this warehouse, and where neither is
4  Test data
   4.1  No production copy in dwh_dev or dwh_test, in any form
   4.2  Synthetic generation from the factories, referential integrity preserved
   4.3  The check that proves it
5  Access: masks, row filters, and the one group that sees unmasked values
6  Erasure requests
   6.1  Finding every SCD2 version of a subject in gold.dim_customer
   6.2  What is erased, and what is retained under a statutory obligation
   6.3  Deletion vectors, VACUUM, and the moment the bytes actually go
7  Deletion from bronze - the hard part, because bronze is append-only by contract
8  Sharing: Delta Sharing, never a CSV export

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%'