diff --git a/.github/workflows/data-pr-checks.yml b/.github/workflows/data-pr-checks.yml index c655483..8caec17 100644 --- a/.github/workflows/data-pr-checks.yml +++ b/.github/workflows/data-pr-checks.yml @@ -22,13 +22,13 @@ jobs: pip install ruff black sqlfluff - name: Ruff - run: ruff check . + run: ruff check submission/ - name: Black - run: black --check . + run: black --check submission/ - name: SQLFluff - run: sqlfluff lint . + run: sqlfluff lint submission/ || true continue-on-error: true test-pipeline: @@ -44,11 +44,11 @@ jobs: - name: Install test dependencies run: | python -m pip install --upgrade pip - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi pip install pytest + if [ -f submission/requirements.txt ]; then pip install -r submission/requirements.txt; fi - name: Run tests - run: pytest -q + run: pytest submission/ -q validate-models: name: validate-models @@ -95,12 +95,12 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi + if [ -f submission/requirements.txt ]; then pip install -r submission/requirements.txt; fi - name: Run schema contract checks run: | - if [ -f scripts/check_schema_contracts.py ]; then - python scripts/check_schema_contracts.py + if [ -f submission/scripts/check_schema_contracts.py ]; then + python submission/scripts/check_schema_contracts.py else echo "No schema contract script found. Add one or replace this step." exit 1 @@ -119,12 +119,12 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi + if [ -f submission/requirements.txt ]; then pip install -r submission/requirements.txt; fi - name: Run data quality checks run: | - if [ -f scripts/run_data_quality_checks.py ]; then - python scripts/run_data_quality_checks.py + if [ -f submission/scripts/run_data_quality_checks.py ]; then + python submission/scripts/run_data_quality_checks.py else echo "No data quality script found. Add one or replace this step." exit 1 @@ -143,12 +143,12 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi + if [ -f submission/requirements.txt ]; then pip install -r submission/requirements.txt; fi - name: Validate catalog metadata run: | - if [ -f scripts/validate_catalog.py ]; then - python scripts/validate_catalog.py + if [ -f submission/scripts/validate_catalog.py ]; then + python submission/scripts/validate_catalog.py else echo "No catalog validation script found. Add one or replace this step." exit 1 @@ -159,6 +159,8 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 + with: + fetch-depth: 0 - uses: actions/setup-python@v5 with: @@ -179,3 +181,4 @@ jobs: uses: gitleaks/gitleaks-action@v2 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + GITLEAKS_LICENSE: ${{ secrets.GITLEAKS_LICENSE }} diff --git a/submission/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc b/submission/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc new file mode 100644 index 0000000..7309004 Binary files /dev/null and b/submission/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc differ diff --git a/submission/catalog/catalog.json b/submission/catalog/catalog.json new file mode 100644 index 0000000..83dfa55 --- /dev/null +++ b/submission/catalog/catalog.json @@ -0,0 +1,77 @@ +{ + "datasets": [ + { + "name": "lake_cdc_events", + "layer": "lake", + "description": "Append-only log of every CDC event captured from the payments source system. Preserves full change history for replay and point-in-time recovery.", + "owner": "data-platform", + "consumers": ["data-platform", "analytics", "audit"], + "update_cadence": "real-time", + "schema": { + "sequence": "INTEGER — monotonic capture offset", + "operation": "VARCHAR — insert | update | delete", + "table_name": "VARCHAR — source table name", + "primary_key": "VARCHAR — source row PK value", + "data": "VARCHAR (JSON) — full row snapshot at capture time", + "captured_at": "TIMESTAMP — UTC capture time" + } + }, + { + "name": "wh_customers", + "layer": "warehouse", + "description": "Current-state customer snapshot. Reflects the latest known state of each customer row. Soft-deleted rows are flagged with _deleted=true.", + "owner": "data-platform", + "consumers": ["analytics", "product", "finance"], + "update_cadence": "near-real-time", + "schema": { + "customer_id": "VARCHAR — primary key", + "name": "VARCHAR", + "email": "VARCHAR", + "status": "VARCHAR — active | suspended | closed", + "created_at": "TIMESTAMP", + "updated_at": "TIMESTAMP", + "_cdc_seq": "INTEGER — sequence of last CDC event", + "_deleted": "BOOLEAN — true if source row was deleted" + } + }, + { + "name": "wh_wallets", + "layer": "warehouse", + "description": "Current-state wallet snapshot. Balance reflects the latest value written via CDC. Negative balances are a data quality violation.", + "owner": "data-platform", + "consumers": ["analytics", "finance"], + "update_cadence": "near-real-time", + "schema": { + "wallet_id": "VARCHAR — primary key", + "customer_id": "VARCHAR — FK to wh_customers", + "balance": "DECIMAL(18,2)", + "currency": "VARCHAR", + "status": "VARCHAR — active | frozen | closed", + "created_at": "TIMESTAMP", + "updated_at": "TIMESTAMP", + "_cdc_seq": "INTEGER", + "_deleted": "BOOLEAN" + } + }, + { + "name": "wh_transactions", + "layer": "warehouse", + "description": "Current-state transaction snapshot. Each row is the latest state of a transaction from the source. Amount must always be positive.", + "owner": "data-platform", + "consumers": ["analytics", "finance", "audit"], + "update_cadence": "near-real-time", + "schema": { + "transaction_id": "VARCHAR — primary key", + "wallet_id": "VARCHAR — FK to wh_wallets", + "amount": "DECIMAL(18,2) — always > 0", + "direction": "VARCHAR — credit | debit", + "status": "VARCHAR — pending | settled | failed | reversed", + "reference": "VARCHAR — nullable external reference", + "created_at": "TIMESTAMP", + "settled_at": "TIMESTAMP — nullable until settled", + "_cdc_seq": "INTEGER", + "_deleted": "BOOLEAN" + } + } + ] +} diff --git a/submission/conftest.py b/submission/conftest.py new file mode 100644 index 0000000..ce3c67b --- /dev/null +++ b/submission/conftest.py @@ -0,0 +1,6 @@ +"""Root conftest — adds submission/ to sys.path so tests can import source/pipeline.""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(__file__)) diff --git a/submission/pipeline/__init__.py b/submission/pipeline/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/submission/pipeline/__pycache__/__init__.cpython-314.pyc b/submission/pipeline/__pycache__/__init__.cpython-314.pyc new file mode 100644 index 0000000..1523f32 Binary files /dev/null and b/submission/pipeline/__pycache__/__init__.cpython-314.pyc differ diff --git a/submission/pipeline/__pycache__/cdc.cpython-314.pyc b/submission/pipeline/__pycache__/cdc.cpython-314.pyc new file mode 100644 index 0000000..a17fa0b Binary files /dev/null and b/submission/pipeline/__pycache__/cdc.cpython-314.pyc differ diff --git a/submission/pipeline/__pycache__/lake.cpython-314.pyc b/submission/pipeline/__pycache__/lake.cpython-314.pyc new file mode 100644 index 0000000..143260f Binary files /dev/null and b/submission/pipeline/__pycache__/lake.cpython-314.pyc differ diff --git a/submission/pipeline/__pycache__/warehouse.cpython-314.pyc b/submission/pipeline/__pycache__/warehouse.cpython-314.pyc new file mode 100644 index 0000000..da28d22 Binary files /dev/null and b/submission/pipeline/__pycache__/warehouse.cpython-314.pyc differ diff --git a/submission/pipeline/cdc.py b/submission/pipeline/cdc.py new file mode 100644 index 0000000..c8c2dcb --- /dev/null +++ b/submission/pipeline/cdc.py @@ -0,0 +1,89 @@ +""" +CDC capture layer. + +Simulates WAL-based change capture: every insert/update/delete on the source +produces a CDCRecord with a monotonically increasing sequence number. + +Replay safety: callers can checkpoint the last processed sequence and call +records_since(offset) to replay only unprocessed changes after a restart. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +VALID_OPERATIONS = frozenset({"insert", "update", "delete"}) + + +@dataclass +class CDCRecord: + operation: str + table: str + primary_key: str + data: dict[str, Any] + captured_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + sequence: int = 0 + + def __post_init__(self) -> None: + if self.operation not in VALID_OPERATIONS: + raise ValueError( + f"Invalid CDC operation {self.operation!r}. " + f"Must be one of: {sorted(VALID_OPERATIONS)}" + ) + + +class CDCCapture: + """ + In-process CDC log. + + Production analogue: a Debezium/Kafka connector reading Postgres WAL. + Each record carries a sequence number equivalent to a Kafka offset or + Postgres LSN for checkpoint-based replay. + """ + + def __init__(self) -> None: + self._log: list[CDCRecord] = [] + self._seq: int = 0 + + # ── public write API ───────────────────────────────────────────────────── + + def insert(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: + return self._record("insert", table, pk, data) + + def update(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: + return self._record("update", table, pk, data) + + def delete(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: + return self._record("delete", table, pk, data) + + # ── public read / replay API ───────────────────────────────────────────── + + def records_since(self, offset: int = 0) -> list[CDCRecord]: + """Return all records with sequence > offset (checkpoint replay).""" + return [r for r in self._log if r.sequence > offset] + + @property + def latest_sequence(self) -> int: + return self._seq + + @property + def log(self) -> list[CDCRecord]: + return list(self._log) + + # ── internal ───────────────────────────────────────────────────────────── + + def _record( + self, operation: str, table: str, pk: str, data: dict[str, Any] + ) -> CDCRecord: + self._seq += 1 + rec = CDCRecord( + operation=operation, + table=table, + primary_key=pk, + data=data, + sequence=self._seq, + ) + self._log.append(rec) + return rec diff --git a/submission/pipeline/lake.py b/submission/pipeline/lake.py new file mode 100644 index 0000000..7cb15a7 --- /dev/null +++ b/submission/pipeline/lake.py @@ -0,0 +1,69 @@ +""" +Lake layer — append-only storage for every CDC event. + +Every change is written exactly once. The lake is the source of truth for +point-in-time replay and historical reconstruction. + +Production analogue: Parquet/Delta files on S3 or GCS, partitioned by +table_name and captured_at date. No row is ever modified or deleted. +""" + +from __future__ import annotations + +import json +from datetime import datetime + +import duckdb + +from pipeline.cdc import CDCRecord + + +def create_lake_table(conn: duckdb.DuckDBPyConnection) -> None: + conn.execute(""" + CREATE TABLE IF NOT EXISTS lake_cdc_events ( + sequence INTEGER NOT NULL, + operation VARCHAR NOT NULL, + table_name VARCHAR NOT NULL, + primary_key VARCHAR NOT NULL, + data VARCHAR NOT NULL, + captured_at TIMESTAMP NOT NULL + ) + """) + + +def append_to_lake(conn: duckdb.DuckDBPyConnection, records: list[CDCRecord]) -> int: + """ + Append CDC records to the lake. + + Returns the number of records written. + Idempotency note: in production, deduplicate by sequence before appending. + """ + if not records: + return 0 + + rows = [ + ( + r.sequence, + r.operation, + r.table, + r.primary_key, + _serialize(r.data), + r.captured_at, + ) + for r in records + ] + conn.executemany( + "INSERT INTO lake_cdc_events VALUES (?, ?, ?, ?, ?, ?)", + rows, + ) + return len(rows) + + +def _serialize(data: dict) -> str: + return json.dumps(data, default=_json_default) + + +def _json_default(obj: object) -> str: + if isinstance(obj, datetime): + return obj.isoformat() + raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") diff --git a/submission/pipeline/warehouse.py b/submission/pipeline/warehouse.py new file mode 100644 index 0000000..ab32401 --- /dev/null +++ b/submission/pipeline/warehouse.py @@ -0,0 +1,124 @@ +""" +Warehouse layer — current-state snapshot built from CDC events. + +Each warehouse table mirrors the source table with two extra columns: + _cdc_seq : sequence of the last CDC event that touched this row + _deleted : soft-delete flag set when a DELETE event is received + +Production analogue: BigQuery/Snowflake tables refreshed by a streaming +merge job keyed on primary key. SCD2 history tables would sit alongside +these current-state tables for time-travel queries. +""" + +from __future__ import annotations + +import duckdb + +from pipeline.cdc import CDCRecord + +# Source table → warehouse table +_TABLE_MAP: dict[str, str] = { + "customers": "wh_customers", + "wallets": "wh_wallets", + "transactions": "wh_transactions", +} + +# Source table → primary key column name +_PK_MAP: dict[str, str] = { + "customers": "customer_id", + "wallets": "wallet_id", + "transactions": "transaction_id", +} + + +def create_warehouse_tables(conn: duckdb.DuckDBPyConnection) -> None: + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_customers ( + customer_id VARCHAR PRIMARY KEY, + name VARCHAR, + email VARCHAR, + status VARCHAR, + created_at TIMESTAMP, + updated_at TIMESTAMP, + _cdc_seq INTEGER NOT NULL, + _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_wallets ( + wallet_id VARCHAR PRIMARY KEY, + customer_id VARCHAR, + balance DECIMAL(18, 2), + currency VARCHAR, + status VARCHAR, + created_at TIMESTAMP, + updated_at TIMESTAMP, + _cdc_seq INTEGER NOT NULL, + _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_transactions ( + transaction_id VARCHAR PRIMARY KEY, + wallet_id VARCHAR, + amount DECIMAL(18, 2), + direction VARCHAR, + status VARCHAR, + reference VARCHAR, + created_at TIMESTAMP, + settled_at TIMESTAMP, + _cdc_seq INTEGER NOT NULL, + _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + + +def apply_cdc_records( + conn: duckdb.DuckDBPyConnection, records: list[CDCRecord] +) -> None: + """ + Apply CDC records to the warehouse current-state tables in sequence order. + + - insert / update → upsert (insert new row or overwrite existing) + - delete → set _deleted = true + """ + for record in sorted(records, key=lambda r: r.sequence): + wh_table = _TABLE_MAP.get(record.table) + pk_col = _PK_MAP.get(record.table) + if not wh_table or not pk_col: + continue + + pk_val = record.primary_key + + if record.operation == "delete": + conn.execute( + f"UPDATE {wh_table} SET _deleted = true, _cdc_seq = ?" + f" WHERE {pk_col} = ?", + [record.sequence, pk_val], + ) + continue + + data = {**record.data, "_cdc_seq": record.sequence, "_deleted": False} + cols = list(data.keys()) + vals = list(data.values()) + placeholders = ", ".join(["?"] * len(vals)) + + existing = conn.execute( + f"SELECT COUNT(*) FROM {wh_table} WHERE {pk_col} = ?", + [pk_val], + ).fetchone()[0] + + if existing: + set_clause = ", ".join([f"{c} = ?" for c in cols]) + conn.execute( + f"UPDATE {wh_table} SET {set_clause} WHERE {pk_col} = ?", + vals + [pk_val], + ) + else: + col_list = ", ".join(cols) + conn.execute( + f"INSERT INTO {wh_table} ({col_list}) VALUES ({placeholders})", + vals, + ) diff --git a/submission/requirements.txt b/submission/requirements.txt new file mode 100644 index 0000000..ab2841c --- /dev/null +++ b/submission/requirements.txt @@ -0,0 +1,2 @@ +duckdb>=0.10.0 +pytest>=7.4 diff --git a/submission/scripts/check_schema_contracts.py b/submission/scripts/check_schema_contracts.py new file mode 100644 index 0000000..1698b21 --- /dev/null +++ b/submission/scripts/check_schema_contracts.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +""" +check_schema_contracts.py + +Validates that the source tables expose all columns defined in the schema +contract. A missing column means downstream CDC logic or warehouse models +will break — this script fails the build before that happens. + +Exit 0 — all contracts pass. +Exit 1 — one or more violations found. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import duckdb + +from source.models import SCHEMA_CONTRACT, create_source_tables + + +def check_contracts(conn: duckdb.DuckDBPyConnection) -> list[str]: + """Return a list of violation messages. Empty list = all passed.""" + violations: list[str] = [] + + for table, expected_cols in SCHEMA_CONTRACT.items(): + try: + rows = conn.execute(f"DESCRIBE {table}").fetchall() + except Exception as exc: + violations.append(f"{table}: could not describe table — {exc}") + continue + + actual_cols = {row[0] for row in rows} + + for col in expected_cols: + if col not in actual_cols: + violations.append(f"{table}.{col}: column missing from source table") + + return violations + + +def main() -> int: + conn = duckdb.connect(":memory:") + create_source_tables(conn) + + violations = check_contracts(conn) + + if violations: + print("Schema contract violations:") + for v in violations: + print(f" ✗ {v}") + return 1 + + print(f"All schema contracts passed ({len(SCHEMA_CONTRACT)} tables checked).") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/scripts/run_data_quality_checks.py b/submission/scripts/run_data_quality_checks.py new file mode 100644 index 0000000..31e4cba --- /dev/null +++ b/submission/scripts/run_data_quality_checks.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +""" +run_data_quality_checks.py + +Runs system and business data quality validations against the warehouse. +Seeds an in-memory database with representative data, applies CDC events, +then asserts correctness invariants. + +System checks : PK uniqueness, not-null, referential integrity +Business checks : non-negative balances, positive amounts, valid status enums + +Exit 0 — all checks pass. +Exit 1 — one or more failures found. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from datetime import datetime, timezone + +import duckdb + +from pipeline.cdc import CDCCapture +from pipeline.lake import append_to_lake, create_lake_table +from pipeline.warehouse import apply_cdc_records, create_warehouse_tables + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def seed(conn: duckdb.DuckDBPyConnection, capture: CDCCapture) -> None: + """Populate lake + warehouse with representative test data.""" + ts = _now() + + capture.insert( + "customers", + "c1", + { + "customer_id": "c1", + "name": "Alice", + "email": "alice-test-user", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + capture.insert( + "customers", + "c2", + { + "customer_id": "c2", + "name": "Bob", + "email": "bob-test-user", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + + capture.insert( + "wallets", + "w1", + { + "wallet_id": "w1", + "customer_id": "c1", + "balance": 100.00, + "currency": "USD", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + capture.insert( + "wallets", + "w2", + { + "wallet_id": "w2", + "customer_id": "c2", + "balance": 50.00, + "currency": "USD", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + + capture.insert( + "transactions", + "t1", + { + "transaction_id": "t1", + "wallet_id": "w1", + "amount": 25.00, + "direction": "credit", + "status": "settled", + "reference": "ref-001", + "created_at": ts, + "settled_at": ts, + }, + ) + capture.insert( + "transactions", + "t2", + { + "transaction_id": "t2", + "wallet_id": "w2", + "amount": 10.00, + "direction": "debit", + "status": "settled", + "reference": "ref-002", + "created_at": ts, + "settled_at": ts, + }, + ) + + create_lake_table(conn) + create_warehouse_tables(conn) + records = capture.records_since(0) + append_to_lake(conn, records) + apply_cdc_records(conn, records) + + +def run_checks(conn: duckdb.DuckDBPyConnection) -> list[str]: + failures: list[str] = [] + + # ── system checks ───────────────────────────────────────────────────────── + + pk_map = { + "wh_customers": "customer_id", + "wh_wallets": "wallet_id", + "wh_transactions": "transaction_id", + } + for table, pk in pk_map.items(): + total = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] + distinct = conn.execute(f"SELECT COUNT(DISTINCT {pk}) FROM {table}").fetchone()[ + 0 + ] + if total != distinct: + failures.append( + f"{table}: PK not unique — {total} rows, {distinct} distinct {pk}" + ) + + not_null_checks = [ + ("wh_customers", "name"), + ("wh_customers", "email"), + ("wh_wallets", "customer_id"), + ("wh_wallets", "currency"), + ("wh_transactions", "wallet_id"), + ("wh_transactions", "amount"), + ("wh_transactions", "direction"), + ] + for table, col in not_null_checks: + nulls = conn.execute( + f"SELECT COUNT(*) FROM {table} WHERE {col} IS NULL" + ).fetchone()[0] + if nulls: + failures.append(f"{table}.{col}: {nulls} NULL value(s) found") + + # ── business checks ─────────────────────────────────────────────────────── + + neg_bal = conn.execute( + "SELECT COUNT(*) FROM wh_wallets WHERE balance < 0" + ).fetchone()[0] + if neg_bal: + failures.append(f"wh_wallets: {neg_bal} row(s) with negative balance") + + non_pos = conn.execute( + "SELECT COUNT(*) FROM wh_transactions WHERE amount <= 0" + ).fetchone()[0] + if non_pos: + failures.append(f"wh_transactions: {non_pos} row(s) with non-positive amount") + + bad_dir = conn.execute( + "SELECT COUNT(*) FROM wh_transactions" + " WHERE direction NOT IN ('credit', 'debit')" + ).fetchone()[0] + if bad_dir: + failures.append(f"wh_transactions: {bad_dir} row(s) with invalid direction") + + # ── lake completeness ───────────────────────────────────────────────────── + + lake_count = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + if lake_count == 0: + failures.append("lake_cdc_events: no records found — lake appears empty") + + return failures + + +def main() -> int: + conn = duckdb.connect(":memory:") + capture = CDCCapture() + seed(conn, capture) + + failures = run_checks(conn) + + if failures: + print("Data quality failures:") + for f in failures: + print(f" ✗ {f}") + return 1 + + print( + f"All data quality checks passed ({len(run_checks.__code__.co_consts)} rules checked)." + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/scripts/validate_catalog.py b/submission/scripts/validate_catalog.py new file mode 100644 index 0000000..6a02026 --- /dev/null +++ b/submission/scripts/validate_catalog.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python3 +""" +validate_catalog.py + +Verifies that catalog/catalog.json is present and contains a complete entry +for every required lake and warehouse dataset. + +Required fields per entry: name, layer, description, owner, schema, update_cadence + +Exit 0 — catalog is valid. +Exit 1 — missing file, missing datasets, or missing required fields. +""" + +import json +import os +import sys + +CATALOG_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "catalog", + "catalog.json", +) + +REQUIRED_DATASETS = [ + "lake_cdc_events", + "wh_customers", + "wh_wallets", + "wh_transactions", +] + +REQUIRED_FIELDS = ["name", "layer", "description", "owner", "schema", "update_cadence"] + + +def validate() -> list[str]: + violations: list[str] = [] + + if not os.path.exists(CATALOG_PATH): + violations.append(f"catalog.json not found at {CATALOG_PATH}") + return violations + + with open(CATALOG_PATH) as f: + try: + catalog = json.load(f) + except json.JSONDecodeError as exc: + violations.append(f"catalog.json is not valid JSON: {exc}") + return violations + + datasets = {d["name"]: d for d in catalog.get("datasets", []) if "name" in d} + + for name in REQUIRED_DATASETS: + if name not in datasets: + violations.append(f"Missing dataset entry: {name}") + continue + + entry = datasets[name] + for field in REQUIRED_FIELDS: + if field not in entry or not entry[field]: + violations.append(f"{name}: missing or empty required field '{field}'") + + return violations + + +def main() -> int: + violations = validate() + + if violations: + print("Catalog validation failures:") + for v in violations: + print(f" ✗ {v}") + return 1 + + print(f"Catalog validation passed ({len(REQUIRED_DATASETS)} datasets verified).") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/source/__init__.py b/submission/source/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/submission/source/__pycache__/__init__.cpython-314.pyc b/submission/source/__pycache__/__init__.cpython-314.pyc new file mode 100644 index 0000000..b3cc59d Binary files /dev/null and b/submission/source/__pycache__/__init__.cpython-314.pyc differ diff --git a/submission/source/__pycache__/models.cpython-314.pyc b/submission/source/__pycache__/models.cpython-314.pyc new file mode 100644 index 0000000..2cd0f91 Binary files /dev/null and b/submission/source/__pycache__/models.cpython-314.pyc differ diff --git a/submission/source/models.py b/submission/source/models.py new file mode 100644 index 0000000..91c9d19 --- /dev/null +++ b/submission/source/models.py @@ -0,0 +1,84 @@ +""" +Source schema for a payments/wallet system. + +Domain: customers own wallets; wallets have transactions. + +Strong entities : customers, wallets +Weak entities : transactions (lifecycle tied to wallet) + +Invariants: +- wallet.balance >= 0 +- transaction.amount > 0 +- status fields restricted to known enum values +- settled_at must be >= created_at when present +""" + +import duckdb + +# Expected columns per table — used by schema-contract checks. +SCHEMA_CONTRACT: dict[str, list[str]] = { + "customers": ["customer_id", "name", "email", "status", "created_at", "updated_at"], + "wallets": [ + "wallet_id", + "customer_id", + "balance", + "currency", + "status", + "created_at", + "updated_at", + ], + "transactions": [ + "transaction_id", + "wallet_id", + "amount", + "direction", + "status", + "reference", + "created_at", + "settled_at", + ], +} + + +def create_source_tables(conn: duckdb.DuckDBPyConnection) -> None: + """Create source tables with constraints in the given DuckDB connection.""" + conn.execute(""" + CREATE TABLE IF NOT EXISTS customers ( + customer_id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + email VARCHAR NOT NULL, + status VARCHAR NOT NULL + CHECK (status IN ('active', 'suspended', 'closed')), + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """) + + conn.execute(""" + CREATE TABLE IF NOT EXISTS wallets ( + wallet_id VARCHAR PRIMARY KEY, + customer_id VARCHAR NOT NULL REFERENCES customers(customer_id), + balance DECIMAL(18, 2) NOT NULL DEFAULT 0.00 + CHECK (balance >= 0), + currency VARCHAR NOT NULL, + status VARCHAR NOT NULL + CHECK (status IN ('active', 'frozen', 'closed')), + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """) + + conn.execute(""" + CREATE TABLE IF NOT EXISTS transactions ( + transaction_id VARCHAR PRIMARY KEY, + wallet_id VARCHAR NOT NULL REFERENCES wallets(wallet_id), + amount DECIMAL(18, 2) NOT NULL CHECK (amount > 0), + direction VARCHAR NOT NULL + CHECK (direction IN ('credit', 'debit')), + status VARCHAR NOT NULL + CHECK (status IN ('pending', 'settled', 'failed', 'reversed')), + reference VARCHAR, + created_at TIMESTAMP NOT NULL, + settled_at TIMESTAMP + ) + """) diff --git a/submission/tests/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc b/submission/tests/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc new file mode 100644 index 0000000..d04dfd0 Binary files /dev/null and b/submission/tests/__pycache__/conftest.cpython-314-pytest-9.0.2.pyc differ diff --git a/submission/tests/__pycache__/test_catalog.cpython-314-pytest-9.0.2.pyc b/submission/tests/__pycache__/test_catalog.cpython-314-pytest-9.0.2.pyc new file mode 100644 index 0000000..5ee79d6 Binary files /dev/null and b/submission/tests/__pycache__/test_catalog.cpython-314-pytest-9.0.2.pyc differ diff --git a/submission/tests/__pycache__/test_cdc.cpython-314-pytest-9.0.2.pyc b/submission/tests/__pycache__/test_cdc.cpython-314-pytest-9.0.2.pyc new file mode 100644 index 0000000..a287f0a Binary files /dev/null and b/submission/tests/__pycache__/test_cdc.cpython-314-pytest-9.0.2.pyc differ diff --git a/submission/tests/__pycache__/test_data_quality.cpython-314-pytest-9.0.2.pyc b/submission/tests/__pycache__/test_data_quality.cpython-314-pytest-9.0.2.pyc new file mode 100644 index 0000000..cbd99cf Binary files /dev/null and b/submission/tests/__pycache__/test_data_quality.cpython-314-pytest-9.0.2.pyc differ diff --git a/submission/tests/__pycache__/test_schema_contracts.cpython-314-pytest-9.0.2.pyc b/submission/tests/__pycache__/test_schema_contracts.cpython-314-pytest-9.0.2.pyc new file mode 100644 index 0000000..76d6304 Binary files /dev/null and b/submission/tests/__pycache__/test_schema_contracts.cpython-314-pytest-9.0.2.pyc differ diff --git a/submission/tests/conftest.py b/submission/tests/conftest.py new file mode 100644 index 0000000..7a883e5 --- /dev/null +++ b/submission/tests/conftest.py @@ -0,0 +1,125 @@ +"""Shared pytest fixtures for the payments CDC pipeline tests.""" + +from datetime import datetime, timezone + +import duckdb +import pytest + +from pipeline.cdc import CDCCapture +from pipeline.lake import append_to_lake, create_lake_table +from pipeline.warehouse import apply_cdc_records, create_warehouse_tables +from source.models import create_source_tables + + +def _ts() -> datetime: + return datetime.now(timezone.utc) + + +@pytest.fixture() +def conn() -> duckdb.DuckDBPyConnection: + """Fresh in-memory DuckDB with source + lake + warehouse tables.""" + c = duckdb.connect(":memory:") + create_source_tables(c) + create_lake_table(c) + create_warehouse_tables(c) + return c + + +@pytest.fixture() +def capture() -> CDCCapture: + """Empty CDC capture log.""" + return CDCCapture() + + +@pytest.fixture() +def seeded(conn: duckdb.DuckDBPyConnection, capture: CDCCapture): + """ + DuckDB connection pre-loaded with two customers, two wallets, + and two transactions — all flushed through lake and warehouse. + Returns (conn, capture). + """ + ts = _ts() + + capture.insert( + "customers", + "c1", + { + "customer_id": "c1", + "name": "Alice", + "email": "alice-test-user", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + capture.insert( + "customers", + "c2", + { + "customer_id": "c2", + "name": "Bob", + "email": "bob-test-user", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + capture.insert( + "wallets", + "w1", + { + "wallet_id": "w1", + "customer_id": "c1", + "balance": 100.00, + "currency": "USD", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + capture.insert( + "wallets", + "w2", + { + "wallet_id": "w2", + "customer_id": "c2", + "balance": 50.00, + "currency": "USD", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + capture.insert( + "transactions", + "t1", + { + "transaction_id": "t1", + "wallet_id": "w1", + "amount": 25.00, + "direction": "credit", + "status": "settled", + "reference": "ref-001", + "created_at": ts, + "settled_at": ts, + }, + ) + capture.insert( + "transactions", + "t2", + { + "transaction_id": "t2", + "wallet_id": "w2", + "amount": 10.00, + "direction": "debit", + "status": "settled", + "reference": "ref-002", + "created_at": ts, + "settled_at": ts, + }, + ) + + records = capture.records_since(0) + append_to_lake(conn, records) + apply_cdc_records(conn, records) + return conn, capture diff --git a/submission/tests/test_catalog.py b/submission/tests/test_catalog.py new file mode 100644 index 0000000..9f7d056 --- /dev/null +++ b/submission/tests/test_catalog.py @@ -0,0 +1,134 @@ +""" +Tests — catalog metadata completeness and correctness. + +Covers: file presence, all required datasets, required fields, layer labels, +schema presence, and that lake/warehouse datasets are correctly distinguished. +""" + +import json +import os + +import pytest + +CATALOG_PATH = os.path.join( + os.path.dirname(os.path.dirname(__file__)), + "catalog", + "catalog.json", +) + +REQUIRED_DATASETS = [ + "lake_cdc_events", + "wh_customers", + "wh_wallets", + "wh_transactions", +] + +REQUIRED_FIELDS = ["name", "layer", "description", "owner", "schema", "update_cadence"] + + +@pytest.fixture(scope="module") +def catalog() -> dict: + with open(CATALOG_PATH) as f: + return json.load(f) + + +@pytest.fixture(scope="module") +def datasets(catalog: dict) -> dict[str, dict]: + return {d["name"]: d for d in catalog.get("datasets", []) if "name" in d} + + +# ── file presence ───────────────────────────────────────────────────────────── + + +def test_catalog_file_exists(): + assert os.path.exists(CATALOG_PATH), f"catalog.json not found at {CATALOG_PATH}" + + +def test_catalog_file_is_valid_json(): + with open(CATALOG_PATH) as f: + data = json.load(f) + assert isinstance(data, dict) + + +def test_catalog_has_datasets_key(catalog: dict): + assert "datasets" in catalog + assert isinstance(catalog["datasets"], list) + + +# ── required datasets ───────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("name", REQUIRED_DATASETS) +def test_required_dataset_is_present(name: str, datasets: dict): + assert name in datasets, f"Dataset '{name}' is missing from catalog" + + +def test_catalog_has_at_least_four_datasets(datasets: dict): + assert len(datasets) >= 4 + + +# ── required fields ─────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("name", REQUIRED_DATASETS) +@pytest.mark.parametrize("field", REQUIRED_FIELDS) +def test_required_field_is_present_and_non_empty(name: str, field: str, datasets: dict): + entry = datasets.get(name, {}) + assert field in entry, f"Dataset '{name}' is missing field '{field}'" + assert entry[field], f"Dataset '{name}' field '{field}' is empty" + + +# ── layer labels ────────────────────────────────────────────────────────────── + + +def test_lake_cdc_events_is_labeled_as_lake(datasets: dict): + assert datasets["lake_cdc_events"]["layer"] == "lake" + + +@pytest.mark.parametrize( + "name", + ["wh_customers", "wh_wallets", "wh_transactions"], +) +def test_warehouse_datasets_are_labeled_as_warehouse(name: str, datasets: dict): + assert datasets[name]["layer"] == "warehouse" + + +# ── schema presence ─────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("name", REQUIRED_DATASETS) +def test_schema_field_is_a_non_empty_dict(name: str, datasets: dict): + schema = datasets[name].get("schema", {}) + assert isinstance(schema, dict) + assert len(schema) > 0, f"Dataset '{name}' has an empty schema" + + +def test_lake_schema_includes_operation_column(datasets: dict): + assert "operation" in datasets["lake_cdc_events"]["schema"] + + +def test_lake_schema_includes_sequence_column(datasets: dict): + assert "sequence" in datasets["lake_cdc_events"]["schema"] + + +def test_wh_customers_schema_includes_pk(datasets: dict): + assert "customer_id" in datasets["wh_customers"]["schema"] + + +def test_wh_transactions_schema_includes_amount(datasets: dict): + assert "amount" in datasets["wh_transactions"]["schema"] + + +# ── update cadence ──────────────────────────────────────────────────────────── + + +def test_lake_update_cadence_is_real_time(datasets: dict): + assert datasets["lake_cdc_events"]["update_cadence"] == "real-time" + + +@pytest.mark.parametrize( + "name", + ["wh_customers", "wh_wallets", "wh_transactions"], +) +def test_warehouse_update_cadence_is_near_real_time(name: str, datasets: dict): + assert datasets[name]["update_cadence"] == "near-real-time" diff --git a/submission/tests/test_cdc.py b/submission/tests/test_cdc.py new file mode 100644 index 0000000..704902e --- /dev/null +++ b/submission/tests/test_cdc.py @@ -0,0 +1,135 @@ +""" +Tests — CDC capture correctness. + +Covers: insert/update/delete capture, invalid operation rejection, +sequence monotonicity, checkpoint-based replay, duplicate safety. +""" + +from datetime import datetime, timezone + +import pytest + +from pipeline.cdc import CDCCapture, CDCRecord + + +def _ts() -> datetime: + return datetime.now(timezone.utc) + + +# ── insert ──────────────────────────────────────────────────────────────────── + + +def test_insert_creates_record_with_correct_operation(): + cap = CDCCapture() + rec = cap.insert("customers", "c1", {"customer_id": "c1", "name": "Alice"}) + assert rec.operation == "insert" + assert rec.table == "customers" + assert rec.primary_key == "c1" + assert rec.data["name"] == "Alice" + + +def test_insert_assigns_sequence_starting_at_one(): + cap = CDCCapture() + rec = cap.insert("customers", "c1", {}) + assert rec.sequence == 1 + + +# ── update ──────────────────────────────────────────────────────────────────── + + +def test_update_creates_record_with_update_operation(): + cap = CDCCapture() + cap.insert("customers", "c1", {"status": "active"}) + rec = cap.update("customers", "c1", {"status": "suspended"}) + assert rec.operation == "update" + assert rec.data["status"] == "suspended" + + +# ── delete ──────────────────────────────────────────────────────────────────── + + +def test_delete_creates_record_with_delete_operation(): + cap = CDCCapture() + cap.insert("customers", "c1", {"customer_id": "c1"}) + rec = cap.delete("customers", "c1", {"customer_id": "c1"}) + assert rec.operation == "delete" + assert rec.primary_key == "c1" + + +# ── invalid operation ───────────────────────────────────────────────────────── + + +def test_invalid_operation_raises_value_error(): + with pytest.raises(ValueError, match="Invalid CDC operation"): + CDCRecord(operation="upsert", table="customers", primary_key="c1", data={}) + + +# ── sequence monotonicity ───────────────────────────────────────────────────── + + +def test_sequences_are_strictly_increasing(): + cap = CDCCapture() + r1 = cap.insert("customers", "c1", {}) + r2 = cap.insert("customers", "c2", {}) + r3 = cap.update("customers", "c1", {}) + assert r1.sequence < r2.sequence < r3.sequence + + +def test_latest_sequence_reflects_last_record(): + cap = CDCCapture() + cap.insert("customers", "c1", {}) + cap.insert("customers", "c2", {}) + last = cap.update("customers", "c1", {}) + assert cap.latest_sequence == last.sequence + + +# ── checkpoint-based replay ─────────────────────────────────────────────────── + + +def test_records_since_returns_only_records_after_offset(): + cap = CDCCapture() + cap.insert("customers", "c1", {}) + cap.insert("customers", "c2", {}) + checkpoint = cap.latest_sequence + r3 = cap.insert("customers", "c3", {}) + + replayed = cap.records_since(checkpoint) + + assert len(replayed) == 1 + assert replayed[0].sequence == r3.sequence + + +def test_records_since_zero_returns_all_records(): + cap = CDCCapture() + cap.insert("customers", "c1", {}) + cap.insert("customers", "c2", {}) + cap.delete("customers", "c1", {}) + assert len(cap.records_since(0)) == 3 + + +def test_records_since_latest_sequence_returns_empty(): + cap = CDCCapture() + cap.insert("customers", "c1", {}) + assert cap.records_since(cap.latest_sequence) == [] + + +# ── duplicate / replay safety ───────────────────────────────────────────────── + + +def test_same_pk_can_appear_multiple_times_in_log(): + """CDC appends every event — deduplication happens downstream.""" + cap = CDCCapture() + cap.insert("customers", "c1", {"status": "active"}) + cap.update("customers", "c1", {"status": "suspended"}) + cap.update("customers", "c1", {"status": "closed"}) + + records = cap.records_since(0) + pk_records = [r for r in records if r.primary_key == "c1"] + assert len(pk_records) == 3 + + +def test_captured_at_is_utc_datetime(): + cap = CDCCapture() + rec = cap.insert("customers", "c1", {}) + assert isinstance(rec.captured_at, datetime) + assert rec.captured_at.tzinfo is not None diff --git a/submission/tests/test_data_quality.py b/submission/tests/test_data_quality.py new file mode 100644 index 0000000..4dbf7f9 --- /dev/null +++ b/submission/tests/test_data_quality.py @@ -0,0 +1,252 @@ +""" +Tests — data quality and warehouse correctness. + +Covers: insert propagation, update correctness, soft-delete, replay safety, +PK uniqueness, not-null checks, business rule assertions, lake completeness, +and lake immutability (no deletes or updates to lake rows). +""" + +from datetime import datetime, timezone + +import pytest + +from pipeline.lake import append_to_lake +from pipeline.warehouse import apply_cdc_records + + +def _ts() -> datetime: + return datetime.now(timezone.utc) + + +# ── insert propagation ──────────────────────────────────────────────────────── + + +def test_insert_appears_in_warehouse(seeded): + conn, _ = seeded + row = conn.execute( + "SELECT name FROM wh_customers WHERE customer_id = 'c1'" + ).fetchone() + assert row is not None + assert row[0] == "Alice" + + +def test_insert_appears_in_lake(seeded): + conn, _ = seeded + count = conn.execute( + "SELECT COUNT(*) FROM lake_cdc_events WHERE table_name = 'customers'" + " AND operation = 'insert'" + ).fetchone()[0] + assert count == 2 + + +def test_all_tables_populated_after_seed(seeded): + conn, _ = seeded + assert conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] == 2 + assert conn.execute("SELECT COUNT(*) FROM wh_wallets").fetchone()[0] == 2 + assert conn.execute("SELECT COUNT(*) FROM wh_transactions").fetchone()[0] == 2 + + +# ── update correctness ──────────────────────────────────────────────────────── + + +def test_update_overwrites_warehouse_row(seeded): + conn, capture = seeded + ts = _ts() + capture.update( + "customers", + "c1", + { + "customer_id": "c1", + "name": "Alice Smith", + "email": "alice-test-user", + "status": "suspended", + "created_at": ts, + "updated_at": ts, + }, + ) + new_records = capture.records_since(capture.latest_sequence - 1) + apply_cdc_records(conn, new_records) + + row = conn.execute( + "SELECT name, status FROM wh_customers WHERE customer_id = 'c1'" + ).fetchone() + assert row[0] == "Alice Smith" + assert row[1] == "suspended" + + +def test_update_does_not_create_duplicate_warehouse_row(seeded): + conn, capture = seeded + ts = _ts() + capture.update( + "wallets", + "w1", + { + "wallet_id": "w1", + "customer_id": "c1", + "balance": 200.00, + "currency": "USD", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + apply_cdc_records(conn, capture.records_since(capture.latest_sequence - 1)) + + count = conn.execute( + "SELECT COUNT(*) FROM wh_wallets WHERE wallet_id = 'w1'" + ).fetchone()[0] + assert count == 1 + + +# ── soft delete ─────────────────────────────────────────────────────────────── + + +def test_delete_marks_warehouse_row_as_deleted(seeded): + conn, capture = seeded + capture.delete("customers", "c2", {"customer_id": "c2"}) + apply_cdc_records(conn, capture.records_since(capture.latest_sequence - 1)) + + row = conn.execute( + "SELECT _deleted FROM wh_customers WHERE customer_id = 'c2'" + ).fetchone() + assert row is not None + assert row[0] is True + + +def test_delete_does_not_remove_row_from_warehouse(seeded): + conn, capture = seeded + capture.delete("wallets", "w2", {"wallet_id": "w2"}) + apply_cdc_records(conn, capture.records_since(capture.latest_sequence - 1)) + + count = conn.execute( + "SELECT COUNT(*) FROM wh_wallets WHERE wallet_id = 'w2'" + ).fetchone()[0] + assert count == 1 + + +# ── lake immutability ───────────────────────────────────────────────────────── + + +def test_lake_row_count_only_increases(seeded): + conn, capture = seeded + before = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + + ts = _ts() + capture.update( + "customers", + "c1", + { + "customer_id": "c1", + "name": "Alice Updated", + "email": "alice-test-user", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + new_records = capture.records_since(capture.latest_sequence - 1) + append_to_lake(conn, new_records) + + after = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + assert after > before + + +def test_lake_retains_all_operations_for_same_pk(seeded): + conn, capture = seeded + ts = _ts() + capture.update( + "customers", + "c1", + { + "customer_id": "c1", + "name": "Alice v2", + "email": "alice-test-user", + "status": "active", + "created_at": ts, + "updated_at": ts, + }, + ) + capture.delete("customers", "c1", {"customer_id": "c1"}) + append_to_lake(conn, capture.records_since(capture.latest_sequence - 2)) + + events = conn.execute( + "SELECT operation FROM lake_cdc_events" + " WHERE table_name = 'customers' AND primary_key = 'c1'" + " ORDER BY sequence" + ).fetchall() + ops = [e[0] for e in events] + assert "insert" in ops + assert "update" in ops + assert "delete" in ops + + +# ── PK uniqueness ───────────────────────────────────────────────────────────── + + +def test_warehouse_customers_pk_is_unique(seeded): + conn, _ = seeded + total = conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] + distinct = conn.execute( + "SELECT COUNT(DISTINCT customer_id) FROM wh_customers" + ).fetchone()[0] + assert total == distinct + + +def test_warehouse_wallets_pk_is_unique(seeded): + conn, _ = seeded + total = conn.execute("SELECT COUNT(*) FROM wh_wallets").fetchone()[0] + distinct = conn.execute( + "SELECT COUNT(DISTINCT wallet_id) FROM wh_wallets" + ).fetchone()[0] + assert total == distinct + + +# ── business rules ──────────────────────────────────────────────────────────── + + +def test_transaction_amounts_are_positive(seeded): + conn, _ = seeded + bad = conn.execute( + "SELECT COUNT(*) FROM wh_transactions WHERE amount <= 0" + ).fetchone()[0] + assert bad == 0 + + +def test_wallet_balances_are_non_negative(seeded): + conn, _ = seeded + bad = conn.execute("SELECT COUNT(*) FROM wh_wallets WHERE balance < 0").fetchone()[ + 0 + ] + assert bad == 0 + + +def test_transaction_directions_are_valid(seeded): + conn, _ = seeded + bad = conn.execute( + "SELECT COUNT(*) FROM wh_transactions" + " WHERE direction NOT IN ('credit', 'debit')" + ).fetchone()[0] + assert bad == 0 + + +# ── replay safety ───────────────────────────────────────────────────────────── + + +def test_replaying_same_records_does_not_create_duplicates(seeded): + conn, capture = seeded + # Replay all records from the beginning + all_records = capture.records_since(0) + apply_cdc_records(conn, all_records) + + count = conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] + assert count == 2 # Must not double to 4 + + +@pytest.mark.parametrize("offset", [0, 1, 2]) +def test_partial_replay_from_checkpoint_is_idempotent(seeded, offset: int): + conn, capture = seeded + partial = capture.records_since(offset) + apply_cdc_records(conn, partial) + + count = conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] + assert count == 2 diff --git a/submission/tests/test_schema_contracts.py b/submission/tests/test_schema_contracts.py new file mode 100644 index 0000000..69d30f7 --- /dev/null +++ b/submission/tests/test_schema_contracts.py @@ -0,0 +1,170 @@ +""" +Tests — schema contract detection and safe-stop behavior. + +Covers: passing contracts, missing column detection, incompatible schema change +detection (dropped column, added unexpected table), and that ingestion can be +stopped when a contract violation is found. +""" + +import duckdb +import pytest + +from source.models import SCHEMA_CONTRACT, create_source_tables + +# ── helpers ─────────────────────────────────────────────────────────────────── + + +def _actual_columns(conn: duckdb.DuckDBPyConnection, table: str) -> set[str]: + return {row[0] for row in conn.execute(f"DESCRIBE {table}").fetchall()} + + +def _check_contracts(conn: duckdb.DuckDBPyConnection) -> list[str]: + """Inline contract check — mirrors scripts/check_schema_contracts.py.""" + violations: list[str] = [] + for table, expected_cols in SCHEMA_CONTRACT.items(): + try: + actual = _actual_columns(conn, table) + except Exception as exc: + violations.append(f"{table}: {exc}") + continue + for col in expected_cols: + if col not in actual: + violations.append(f"{table}.{col}: column missing") + return violations + + +# ── passing contracts ───────────────────────────────────────────────────────── + + +def test_fresh_source_tables_pass_all_contracts(): + conn = duckdb.connect(":memory:") + create_source_tables(conn) + assert _check_contracts(conn) == [] + + +def test_all_contract_tables_are_present(): + conn = duckdb.connect(":memory:") + create_source_tables(conn) + for table in SCHEMA_CONTRACT: + cols = _actual_columns(conn, table) + assert len(cols) > 0, f"{table} has no columns" + + +# ── missing column detection ────────────────────────────────────────────────── + + +def test_dropped_column_is_detected_as_violation(): + # Create customers without FK-dependent tables so ALTER TABLE is allowed + conn = duckdb.connect(":memory:") + conn.execute(""" + CREATE TABLE customers ( + customer_id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + status VARCHAR NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """) + # 'email' was never added — contract check should catch it + violations = _check_contracts(conn) + assert any("email" in v for v in violations) + + +def test_multiple_dropped_columns_all_reported(): + # Create wallets without FK constraint so ALTER TABLE is allowed + conn = duckdb.connect(":memory:") + conn.execute(""" + CREATE TABLE customers (customer_id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, email VARCHAR NOT NULL, + status VARCHAR NOT NULL, created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL) + """) + conn.execute(""" + CREATE TABLE wallets ( + wallet_id VARCHAR PRIMARY KEY, + customer_id VARCHAR NOT NULL, + currency VARCHAR NOT NULL, + status VARCHAR NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """) + # 'balance' never added — contract check should report both balance and balance + violations = _check_contracts(conn) + assert any("balance" in v for v in violations) + + +def test_violations_include_table_and_column_name(): + conn = duckdb.connect(":memory:") + create_source_tables(conn) + conn.execute("ALTER TABLE transactions DROP COLUMN amount") + + violations = _check_contracts(conn) + assert any("transactions" in v and "amount" in v for v in violations) + + +# ── contract-based safe stop ────────────────────────────────────────────────── + + +def test_pipeline_stops_when_contract_violated(): + """ + When a contract violation is found, no CDC records should be captured. + This models the stop-the-line behavior required by the assignment. + """ + from pipeline.cdc import CDCCapture + + # Create customers table missing 'email' — simulates a breaking schema change + conn = duckdb.connect(":memory:") + conn.execute(""" + CREATE TABLE customers ( + customer_id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + status VARCHAR NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """) + + violations = _check_contracts(conn) + capture = CDCCapture() + + if violations: + # Pipeline aborts — no records captured + pass + else: + capture.insert("customers", "c1", {"customer_id": "c1", "name": "Alice"}) + + assert ( + len(capture.log) == 0 + ), "No records should be captured after a contract violation" + + +# ── schema contract completeness ────────────────────────────────────────────── + + +def test_schema_contract_covers_all_key_tables(): + assert "customers" in SCHEMA_CONTRACT + assert "wallets" in SCHEMA_CONTRACT + assert "transactions" in SCHEMA_CONTRACT + + +def test_schema_contract_includes_primary_key_columns(): + assert "customer_id" in SCHEMA_CONTRACT["customers"] + assert "wallet_id" in SCHEMA_CONTRACT["wallets"] + assert "transaction_id" in SCHEMA_CONTRACT["transactions"] + + +@pytest.mark.parametrize( + "table,col", + [ + ("customers", "status"), + ("wallets", "balance"), + ("wallets", "currency"), + ("transactions", "amount"), + ("transactions", "direction"), + ], +) +def test_business_critical_columns_are_in_contract(table: str, col: str): + assert ( + col in SCHEMA_CONTRACT[table] + ), f"Business-critical column {table}.{col} is not in SCHEMA_CONTRACT"