From 11ea4ea5f66f46da63f191cb83f29bc7a2f89de3 Mon Sep 17 00:00:00 2001 From: Benjamin De Vierno <54540257+bdevierno1@users.noreply.github.com> Date: Thu, 18 Jun 2026 20:39:54 -0400 Subject: [PATCH 1/2] Add data-quality checks, contracts, and CI for core datasets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - backend/data_quality/checks.py — four-dimension check primitives (completeness <0.5% null, validity allowed-values + non-negative, uniqueness zero-dup PK, freshness ≤ Tier-1 SLA 3600 s) - backend/data_quality/contracts/ — YAML data contracts for `operations` and `billing_sessions` (owner, schema, quality rules, SLAs) - backend/data_quality/runner.py — CLI that loads contracts, runs all checks, exits 1 on any failure (DATABASE_URL-aware) - backend/tests/test_data_quality.py — 14 unit tests covering all pass/fail paths per dimension + contract schema smoke-test - .github/workflows/data-quality.yml — CI: contract YAML lint, DQ unit tests, full backend suite on every push/PR - backend/requirements.txt — add pyyaml>=6.0.0 Co-Authored-By: Vecna --- .github/workflows/data-quality.yml | 67 ++++++ backend/data_quality/__init__.py | 1 + backend/data_quality/checks.py | 183 ++++++++++++++ .../contracts/billing_sessions.yaml | 77 ++++++ .../data_quality/contracts/operations.yaml | 84 +++++++ backend/data_quality/runner.py | 105 ++++++++ backend/requirements.txt | 1 + backend/tests/test_data_quality.py | 226 ++++++++++++++++++ 8 files changed, 744 insertions(+) create mode 100644 .github/workflows/data-quality.yml create mode 100644 backend/data_quality/__init__.py create mode 100644 backend/data_quality/checks.py create mode 100644 backend/data_quality/contracts/billing_sessions.yaml create mode 100644 backend/data_quality/contracts/operations.yaml create mode 100644 backend/data_quality/runner.py create mode 100644 backend/tests/test_data_quality.py diff --git a/.github/workflows/data-quality.yml b/.github/workflows/data-quality.yml new file mode 100644 index 0000000..d2a14be --- /dev/null +++ b/.github/workflows/data-quality.yml @@ -0,0 +1,67 @@ +name: data-quality + +on: + push: + branches: ["**"] + pull_request: + branches: [main] + +jobs: + checks: + name: Data-quality checks & contract lint + runs-on: ubuntu-latest + + defaults: + run: + working-directory: backend + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + + - name: Install dependencies + run: pip install -r requirements.txt pyyaml + + - name: Lint contracts (YAML parse + required keys) + run: | + python - <<'EOF' + import sys, pathlib, yaml + + REQUIRED_TOP = {"dataset", "ownership", "schema", "quality_rules", "slas"} + REQUIRED_QR = {"primary_key", "not_null_cols", "validity_rules"} + REQUIRED_SLAS = {"freshness_timestamp_col", "freshness_max_seconds"} + errors = [] + + for p in sorted(pathlib.Path("data_quality/contracts").glob("*.yaml")): + doc = yaml.safe_load(p.read_text()) + missing_top = REQUIRED_TOP - doc.keys() + if missing_top: + errors.append(f"{p.name}: missing top-level keys {missing_top}") + continue + qr = doc["quality_rules"] + missing_qr = REQUIRED_QR - qr.keys() + if missing_qr: + errors.append(f"{p.name}: quality_rules missing {missing_qr}") + sla = doc["slas"] + missing_sla = REQUIRED_SLAS - sla.keys() + if missing_sla: + errors.append(f"{p.name}: slas missing {missing_sla}") + if sla.get("freshness_max_seconds", 0) <= 0: + errors.append(f"{p.name}: freshness_max_seconds must be > 0") + + if errors: + print("\n".join(errors), file=sys.stderr) + sys.exit(1) + print(f"OK — {len(list(pathlib.Path('data_quality/contracts').glob('*.yaml')))} contract(s) valid") + EOF + + - name: Run data-quality unit tests + run: pytest tests/test_data_quality.py -v + + - name: Run full backend test suite + run: pytest --tb=short diff --git a/backend/data_quality/__init__.py b/backend/data_quality/__init__.py new file mode 100644 index 0000000..d414cc9 --- /dev/null +++ b/backend/data_quality/__init__.py @@ -0,0 +1 @@ +"""Data-quality checks for Vecna Operations Console core datasets.""" diff --git a/backend/data_quality/checks.py b/backend/data_quality/checks.py new file mode 100644 index 0000000..b5d10a1 --- /dev/null +++ b/backend/data_quality/checks.py @@ -0,0 +1,183 @@ +""" +Data-quality check primitives. + +Dimensions covered (per data-quality-dimensions.md): + completeness — null rate on NOT-NULL cols < 0.5 % + validity — zero constraint violations (allowed values, non-negative) + uniqueness — zero duplicate PKs + freshness — latest row age ≤ SLA window (per pipeline-sla-and-freshness.md) + +All functions accept a *synchronous* sqlalchemy.engine.Connection so they +compose cleanly in tests (sqlite:///:memory:) and in the CLI runner. +""" +from __future__ import annotations + +import datetime +from dataclasses import dataclass, field +from typing import Any + +import sqlalchemy as sa + + +# --------------------------------------------------------------------------- +# Result type +# --------------------------------------------------------------------------- + +@dataclass +class CheckResult: + dataset: str + check: str # completeness | validity | uniqueness | freshness + passed: bool + detail: str + value: Any = None + threshold: Any = None + + def __str__(self) -> str: + status = "PASS" if self.passed else "FAIL" + return f"[{status}] {self.dataset}.{self.check}: {self.detail}" + + +# --------------------------------------------------------------------------- +# Completeness +# --------------------------------------------------------------------------- + +def check_completeness( + conn: sa.engine.Connection, + table: str, + not_null_cols: list[str], + threshold_pct: float = 0.5, +) -> list[CheckResult]: + """Null rate on each NOT-NULL column must be < threshold_pct %.""" + results: list[CheckResult] = [] + total: int = conn.execute(sa.text(f"SELECT COUNT(*) FROM {table}")).scalar() # type: ignore[assignment] + if total == 0: + for col in not_null_cols: + results.append(CheckResult( + dataset=table, check="completeness", passed=True, + detail=f"{col}: empty table — skipped", + value=0.0, threshold=threshold_pct, + )) + return results + + for col in not_null_cols: + null_count: int = conn.execute( # type: ignore[assignment] + sa.text(f"SELECT COUNT(*) FROM {table} WHERE {col} IS NULL") + ).scalar() + null_pct = (null_count / total) * 100 + passed = null_pct < threshold_pct + results.append(CheckResult( + dataset=table, check="completeness", passed=passed, + detail=f"{col}: null_rate={null_pct:.4f}% (threshold<{threshold_pct}%)", + value=round(null_pct, 6), + threshold=threshold_pct, + )) + return results + + +# --------------------------------------------------------------------------- +# Uniqueness +# --------------------------------------------------------------------------- + +def check_uniqueness( + conn: sa.engine.Connection, + table: str, + primary_key: str, +) -> CheckResult: + """Zero duplicate values on the primary-key column.""" + total: int = conn.execute(sa.text(f"SELECT COUNT(*) FROM {table}")).scalar() # type: ignore[assignment] + distinct: int = conn.execute( # type: ignore[assignment] + sa.text(f"SELECT COUNT(DISTINCT {primary_key}) FROM {table}") + ).scalar() + duplicates = total - distinct + return CheckResult( + dataset=table, check="uniqueness", + passed=(duplicates == 0), + detail=f"{primary_key}: {duplicates} duplicate(s) in {total} rows", + value=duplicates, + threshold=0, + ) + + +# --------------------------------------------------------------------------- +# Validity +# --------------------------------------------------------------------------- + +def check_validity( + conn: sa.engine.Connection, + table: str, + rules: list[dict[str, Any]], +) -> list[CheckResult]: + """Zero schema violations. + + Supported rule types: + {"type": "allowed_values", "col": "status", "allowed": ["a", "b"]} + {"type": "non_negative", "col": "cost_usd"} + """ + results: list[CheckResult] = [] + for rule in rules: + rtype = rule.get("type", "allowed_values") + col = rule["col"] + + if rtype == "allowed_values": + quoted = ", ".join(f"'{v}'" for v in rule["allowed"]) + n: int = conn.execute( # type: ignore[assignment] + sa.text( + f"SELECT COUNT(*) FROM {table} " + f"WHERE {col} IS NOT NULL AND {col} NOT IN ({quoted})" + ) + ).scalar() + results.append(CheckResult( + dataset=table, check="validity", passed=(n == 0), + detail=f"{col}: {n} value(s) outside {rule['allowed']}", + value=n, threshold=0, + )) + elif rtype == "non_negative": + n = conn.execute( # type: ignore[assignment] + sa.text(f"SELECT COUNT(*) FROM {table} WHERE {col} < 0") + ).scalar() + results.append(CheckResult( + dataset=table, check="validity", passed=(n == 0), + detail=f"{col}: {n} negative value(s)", + value=n, threshold=0, + )) + return results + + +# --------------------------------------------------------------------------- +# Freshness +# --------------------------------------------------------------------------- + +def check_freshness( + conn: sa.engine.Connection, + table: str, + timestamp_col: str, + max_age_seconds: int, +) -> CheckResult: + """Latest row timestamp must be within max_age_seconds of now (UTC).""" + raw = conn.execute( + sa.text(f"SELECT MAX({timestamp_col}) FROM {table}") + ).scalar() + + if raw is None: + return CheckResult( + dataset=table, check="freshness", passed=True, + detail="empty table — no staleness to measure", + value=None, threshold=max_age_seconds, + ) + + latest: datetime.datetime = ( + datetime.datetime.fromisoformat(raw) if isinstance(raw, str) else raw + ) + # Normalise to naive UTC for comparison + if latest.tzinfo is not None: + latest = latest.utctimetuple() # type: ignore[assignment] + latest = datetime.datetime(*latest[:6]) + + age = (datetime.datetime.utcnow() - latest).total_seconds() + passed = age <= max_age_seconds + return CheckResult( + dataset=table, check="freshness", passed=passed, + detail=f"{timestamp_col}: age={age:.0f}s (SLA≤{max_age_seconds}s)", + value=round(age, 1), + threshold=max_age_seconds, + ) diff --git a/backend/data_quality/contracts/billing_sessions.yaml b/backend/data_quality/contracts/billing_sessions.yaml new file mode 100644 index 0000000..77b63f2 --- /dev/null +++ b/backend/data_quality/contracts/billing_sessions.yaml @@ -0,0 +1,77 @@ +# Data contract — billing_sessions +# Version: 1.0.0 +# Template: data-contract-template.md + +dataset: + name: billing_sessions + table: billing_sessions + version: "1.0.0" + description: > + Cumulative billing counters for a client session (browser tab). + Aggregates token usage, LLM cost, scan fees, and tool fees. + +ownership: + owner: ben@vecna.ai + team: vecna-platform + producer_service: backend/app/services/billing.py + consumers: + - frontend (billing panel) + - cost reporting + +schema: + fields: + - name: id + type: VARCHAR(36) + nullable: false + description: UUID primary key + - name: total_cost_usd + type: REAL + nullable: false + constraints: ">= 0" + - name: total_credits + type: REAL + nullable: false + constraints: ">= 0" + - name: total_input_tokens + type: INTEGER + nullable: false + constraints: ">= 0" + - name: total_output_tokens + type: INTEGER + nullable: false + constraints: ">= 0" + - name: total_billable_usd + type: REAL + nullable: false + constraints: ">= 0" + - name: created_at + type: DATETIME + nullable: false + - name: updated_at + type: DATETIME + nullable: false + +quality_rules: + primary_key: id + not_null_cols: + - id + - total_cost_usd + - total_credits + - total_input_tokens + - total_output_tokens + - created_at + - updated_at + completeness_threshold_pct: 0.5 # null rate must stay below 0.5 % + validity_rules: + - type: non_negative + col: total_cost_usd + - type: non_negative + col: total_credits + - type: non_negative + col: total_billable_usd + +slas: + tier: 1 + freshness_timestamp_col: updated_at + freshness_max_seconds: 3600 # Tier-1: rows must arrive within 1 h of real-time + availability_target_pct: 99.9 diff --git a/backend/data_quality/contracts/operations.yaml b/backend/data_quality/contracts/operations.yaml new file mode 100644 index 0000000..584fc03 --- /dev/null +++ b/backend/data_quality/contracts/operations.yaml @@ -0,0 +1,84 @@ +# Data contract — operations +# Version: 1.0.0 +# Template: data-contract-template.md + +dataset: + name: operations + table: operations + version: "1.0.0" + description: > + Records every passive-recon scan: lifecycle status, token usage, + cost attribution, findings, and timing. + +ownership: + owner: ben@vecna.ai + team: vecna-platform + producer_service: backend/app/services/runner.py + consumers: + - frontend (operations list + report views) + - billing service (cost roll-up into billing_sessions) + - telemetry service + +schema: + fields: + - name: id + type: VARCHAR(36) + nullable: false + description: UUID primary key + - name: session_id + type: VARCHAR(36) + nullable: true + description: FK → billing_sessions.id (SET NULL on delete) + - name: target_url + type: VARCHAR(2048) + nullable: false + - name: status + type: VARCHAR(32) + nullable: false + allowed_values: [pending, running, completed, failed] + - name: cost_usd + type: REAL + nullable: false + constraints: ">= 0" + - name: credits_used + type: REAL + nullable: false + constraints: ">= 0" + - name: billable_total_usd + type: REAL + nullable: false + constraints: ">= 0" + - name: created_at + type: DATETIME + nullable: false + - name: updated_at + type: DATETIME + nullable: false + +quality_rules: + primary_key: id + not_null_cols: + - id + - target_url + - status + - cost_usd + - credits_used + - created_at + - updated_at + completeness_threshold_pct: 0.5 # null rate must stay below 0.5 % + validity_rules: + - type: allowed_values + col: status + allowed: [pending, running, completed, failed] + - type: non_negative + col: cost_usd + - type: non_negative + col: credits_used + - type: non_negative + col: billable_total_usd + +slas: + tier: 1 + freshness_timestamp_col: updated_at + freshness_max_seconds: 3600 # Tier-1: rows must arrive within 1 h of real-time + availability_target_pct: 99.9 diff --git a/backend/data_quality/runner.py b/backend/data_quality/runner.py new file mode 100644 index 0000000..111fcac --- /dev/null +++ b/backend/data_quality/runner.py @@ -0,0 +1,105 @@ +""" +CLI runner — loads contracts from YAML and runs all checks against the live DB. + +Usage: + python -m data_quality.runner [--db-url sqlite:///./vecna.db] + +Exit codes: + 0 all checks passed + 1 one or more checks failed + 2 usage / configuration error +""" +from __future__ import annotations + +import os +import sys +from pathlib import Path + +import sqlalchemy as sa +import yaml + +from data_quality.checks import ( + CheckResult, + check_completeness, + check_freshness, + check_uniqueness, + check_validity, +) + +CONTRACTS_DIR = Path(__file__).parent / "contracts" + + +def load_contracts() -> list[dict]: + contracts = [] + for path in sorted(CONTRACTS_DIR.glob("*.yaml")): + with path.open() as fh: + contracts.append(yaml.safe_load(fh)) + return contracts + + +def run_contract(conn: sa.engine.Connection, contract: dict) -> list[CheckResult]: + table = contract["dataset"]["table"] + quality = contract.get("quality_rules", {}) + sla = contract.get("slas", {}) + results: list[CheckResult] = [] + + # Completeness + not_null_cols = quality.get("not_null_cols", []) + if not_null_cols: + results.extend( + check_completeness( + conn, table, not_null_cols, + threshold_pct=quality.get("completeness_threshold_pct", 0.5), + ) + ) + + # Uniqueness + pk = quality.get("primary_key") + if pk: + results.append(check_uniqueness(conn, table, pk)) + + # Validity + validity_rules = quality.get("validity_rules", []) + if validity_rules: + results.extend(check_validity(conn, table, validity_rules)) + + # Freshness + ts_col = sla.get("freshness_timestamp_col") + max_age = sla.get("freshness_max_seconds") + if ts_col and max_age is not None: + results.append(check_freshness(conn, table, ts_col, int(max_age))) + + return results + + +def main(db_url: str | None = None) -> int: + url = db_url or os.environ.get("DATABASE_URL", "sqlite:///./vecna.db") + # Strip async driver prefix for sync runner + url = url.replace("+aiosqlite", "") + + engine = sa.create_engine(url) + contracts = load_contracts() + if not contracts: + print("ERROR: no contracts found in", CONTRACTS_DIR, file=sys.stderr) + return 2 + + all_results: list[CheckResult] = [] + with engine.connect() as conn: + for contract in contracts: + all_results.extend(run_contract(conn, contract)) + + failed = [r for r in all_results if not r.passed] + for r in all_results: + print(r) + + if failed: + print(f"\n{len(failed)}/{len(all_results)} check(s) FAILED", file=sys.stderr) + return 1 + + print(f"\n{len(all_results)}/{len(all_results)} check(s) passed") + return 0 + + +if __name__ == "__main__": + db_url_arg = sys.argv[1] if len(sys.argv) > 1 else None + sys.exit(main(db_url_arg)) diff --git a/backend/requirements.txt b/backend/requirements.txt index a0213d3..5470f87 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -10,3 +10,4 @@ sse-starlette>=2.0.0 litellm>=1.80.0 openai>=1.0.0 pytest>=8.0.0 +pyyaml>=6.0.0 diff --git a/backend/tests/test_data_quality.py b/backend/tests/test_data_quality.py new file mode 100644 index 0000000..82f0858 --- /dev/null +++ b/backend/tests/test_data_quality.py @@ -0,0 +1,226 @@ +""" +Tests for data_quality.checks — run against an in-memory SQLite DB seeded +with known data so each dimension is verifiable without a live Vecna instance. +""" +from __future__ import annotations + +import datetime + +import pytest +import sqlalchemy as sa + +from data_quality.checks import ( + CheckResult, + check_completeness, + check_freshness, + check_uniqueness, + check_validity, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture() +def conn(): + """Sync in-memory SQLite connection with both core tables pre-created.""" + engine = sa.create_engine("sqlite:///:memory:", future=True) + with engine.begin() as c: + c.execute(sa.text(""" + CREATE TABLE operations ( + id TEXT PRIMARY KEY, + target_url TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + cost_usd REAL NOT NULL DEFAULT 0.0, + credits_used REAL NOT NULL DEFAULT 0.0, + billable_total_usd REAL NOT NULL DEFAULT 0.0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """)) + c.execute(sa.text(""" + CREATE TABLE billing_sessions ( + id TEXT PRIMARY KEY, + total_cost_usd REAL NOT NULL DEFAULT 0.0, + total_credits REAL NOT NULL DEFAULT 0.0, + total_input_tokens INTEGER NOT NULL DEFAULT 0, + total_output_tokens INTEGER NOT NULL DEFAULT 0, + total_billable_usd REAL NOT NULL DEFAULT 0.0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """)) + with engine.connect() as c: + yield c + + +def _now() -> str: + return datetime.datetime.utcnow().isoformat() + + +def _op(conn, id_="op-1", status="completed", cost=0.5, credits=1.0, + billable=0.5, target_url="https://example.com", updated_at=None): + conn.execute(sa.text(""" + INSERT INTO operations (id, target_url, status, cost_usd, credits_used, + billable_total_usd, created_at, updated_at) + VALUES (:id, :url, :status, :cost, :credits, :billable, :now, :upd) + """), { + "id": id_, "url": target_url, "status": status, "cost": cost, + "credits": credits, "billable": billable, + "now": _now(), "upd": updated_at or _now(), + }) + conn.commit() + + +def _session(conn, id_="sess-1", cost=0.0, credits=0.0, billable=0.0, + updated_at=None): + conn.execute(sa.text(""" + INSERT INTO billing_sessions + (id, total_cost_usd, total_credits, total_input_tokens, + total_output_tokens, total_billable_usd, created_at, updated_at) + VALUES (:id, :cost, :credits, 0, 0, :billable, :now, :upd) + """), { + "id": id_, "cost": cost, "credits": credits, "billable": billable, + "now": _now(), "upd": updated_at or _now(), + }) + conn.commit() + + +# --------------------------------------------------------------------------- +# Completeness +# --------------------------------------------------------------------------- + +class TestCompleteness: + def test_passes_on_full_data(self, conn): + _op(conn) + results = check_completeness(conn, "operations", ["id", "target_url", "status"]) + assert all(r.passed for r in results) + + def test_empty_table_passes(self, conn): + results = check_completeness(conn, "operations", ["id", "target_url"]) + assert all(r.passed for r in results) + assert all("empty table" in r.detail for r in results) + + def test_fails_when_null_rate_exceeds_threshold(self, conn): + # Simulate a column that carries nulls in an analytics replica — use a + # separate table without NOT NULL on the tested column. + conn.execute(sa.text(""" + CREATE TABLE replica_ops (id TEXT PRIMARY KEY, url TEXT, ts TEXT) + """)) + conn.execute(sa.text("INSERT INTO replica_ops VALUES ('r1', NULL, :now)"), {"now": _now()}) + conn.commit() + results = check_completeness(conn, "replica_ops", ["url"]) + assert len(results) == 1 + assert not results[0].passed + assert results[0].value == 100.0 + + +# --------------------------------------------------------------------------- +# Uniqueness +# --------------------------------------------------------------------------- + +class TestUniqueness: + def test_passes_unique_pk(self, conn): + _op(conn, id_="op-a") + _op(conn, id_="op-b") + r = check_uniqueness(conn, "operations", "id") + assert r.passed + assert r.value == 0 + + def test_empty_table_passes(self, conn): + r = check_uniqueness(conn, "operations", "id") + assert r.passed + + def test_fails_on_duplicate_pk(self, conn): + # Force duplicate via INSERT OR IGNORE bypass: use a raw insert trick. + # SQLite enforces PK uniqueness, so we'll create a temp table to simulate. + conn.execute(sa.text(""" + CREATE TABLE ops_dup (id TEXT, val TEXT) + """)) + conn.execute(sa.text("INSERT INTO ops_dup VALUES ('x', 'a')")) + conn.execute(sa.text("INSERT INTO ops_dup VALUES ('x', 'b')")) + conn.commit() + r = check_uniqueness(conn, "ops_dup", "id") + assert not r.passed + assert r.value == 1 + + +# --------------------------------------------------------------------------- +# Validity +# --------------------------------------------------------------------------- + +class TestValidity: + def test_valid_status_passes(self, conn): + _op(conn, status="completed") + results = check_validity(conn, "operations", [ + {"type": "allowed_values", "col": "status", + "allowed": ["pending", "running", "completed", "failed"]}, + ]) + assert all(r.passed for r in results) + + def test_invalid_status_fails(self, conn): + _op(conn, status="unknown_status") + results = check_validity(conn, "operations", [ + {"type": "allowed_values", "col": "status", + "allowed": ["pending", "running", "completed", "failed"]}, + ]) + assert not results[0].passed + assert results[0].value == 1 + + def test_non_negative_passes(self, conn): + _op(conn, cost=0.0, credits=5.0) + results = check_validity(conn, "operations", [ + {"type": "non_negative", "col": "cost_usd"}, + {"type": "non_negative", "col": "credits_used"}, + ]) + assert all(r.passed for r in results) + + def test_negative_cost_fails(self, conn): + _op(conn, cost=-1.0) + results = check_validity(conn, "operations", [ + {"type": "non_negative", "col": "cost_usd"}, + ]) + assert not results[0].passed + assert results[0].value == 1 + + +# --------------------------------------------------------------------------- +# Freshness +# --------------------------------------------------------------------------- + +class TestFreshness: + def test_recent_row_passes(self, conn): + _op(conn) # updated_at = now + r = check_freshness(conn, "operations", "updated_at", max_age_seconds=3600) + assert r.passed + + def test_empty_table_passes(self, conn): + r = check_freshness(conn, "operations", "updated_at", max_age_seconds=60) + assert r.passed + assert "empty table" in r.detail + + def test_stale_row_fails(self, conn): + two_hours_ago = ( + datetime.datetime.utcnow() - datetime.timedelta(hours=2) + ).isoformat() + _op(conn, updated_at=two_hours_ago) + r = check_freshness(conn, "operations", "updated_at", max_age_seconds=3600) + assert not r.passed + assert r.value > 3600 + + +# --------------------------------------------------------------------------- +# Contract smoke-test: contracts load without error +# --------------------------------------------------------------------------- + +def test_contracts_load(): + from data_quality.runner import load_contracts + contracts = load_contracts() + assert len(contracts) == 2 + tables = {c["dataset"]["table"] for c in contracts} + assert "operations" in tables + assert "billing_sessions" in tables + for c in contracts: + assert "quality_rules" in c + assert "slas" in c + assert c["slas"]["freshness_max_seconds"] > 0 From 1d8cc7ca9f8f133f780b238a9b9ff640545fac00 Mon Sep 17 00:00:00 2001 From: Benjamin De Vierno <54540257+bdevierno1@users.noreply.github.com> Date: Mon, 13 Jul 2026 19:38:48 -0400 Subject: [PATCH 2/2] fix(ci): add pyproject.toml so pytest can import data_quality package CI runs pytest from backend/ working-directory but there was no pythonpath configuration, so `from data_quality.checks import ...` raised ModuleNotFoundError at collection time (exit 2). Adding [tool.pytest.ini_options] pythonpath = ["."] tells pytest to prepend the backend/ directory to sys.path, making data_quality and any sibling packages importable without an install step. Co-Authored-By: Claude Sonnet 4.6 --- backend/pyproject.toml | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 backend/pyproject.toml diff --git a/backend/pyproject.toml b/backend/pyproject.toml new file mode 100644 index 0000000..b673220 --- /dev/null +++ b/backend/pyproject.toml @@ -0,0 +1,5 @@ +[tool.pytest.ini_options] +# Add the backend/ directory to sys.path so that `data_quality` (and other +# top-level packages that live alongside `tests/`) are importable without +# a pip-install step in CI. +pythonpath = ["."]