TEST_PLAN.md

ObservaKit Test Plan

This is my test plan for the ObservaKit data observability project assignment. I have analyzed the codebase, run the tests, and documented how the core features work.

---

1. What is included and what is not

What is working / in-scope:

• Core Monitoring Features: Checking data freshness, tracking row counts (volume checks) with backfill detection, running quality checks (Soda Core & Custom SQL queries), detecting schema drift, validating data contracts, and recording pipeline runs.
• Alerting: Managing alert channels (Slack, Teams, PagerDuty, Email, Discord, Webhooks), setting up mute windows (suppressions), and automatic alert deduplication (noise reduction).
• API endpoints: Standard endpoints to trigger and query results.
• Local environment: Runs on SQLite for local development and PostgreSQL for Docker.

What is not implemented / out-of-scope:

• Great Expectations: The code has a function for Great Expectations quality checks, but it is just a placeholder and returns a fake failure. It is not implemented.
• Roadmap features: Future items like the dashboard UI editor, incident timelines, and column lineage are not in the current codebase.

---

2. The 7 core flows I checked

F1: Booting the project and checking API health

• What I checked: Setting up the docker containers, running migrations, and hit the health endpoints.
• Files and folders involved:
  • docker-compose.yml
  • backend/main.py (FastAPI app setup)
  • backend/models.py (Database tables definition)
  • backend/scheduler.py (Runs background checks)
  • alembic/ (Database migrations folder)
• Endpoints:
  • GET /healthz (Check if DB connection is active)
  • GET /status (Summary of table health in last 24 hours)
• Inputs needed:
  • Env variables like OBSERVAKIT_API_KEY and DB credentials.
• Expected result:
  • Running docker compose up starts everything and automatically runs database migrations.
  • /healthz returns {"status": "ok"} when the database is connected.
• Status: ✅ appears to work

---

F2: Running a data freshness check

• What I checked: How the system monitors if tables are receiving fresh data based on SLA rules.
• Files and folders involved:
  • backend/routers/freshness.py
  • connectors/base.py (Used to query target databases)
  • config/kit.yml (Where we configure freshness SLAs)
• Endpoints:
  • POST /freshness/poll (Runs the freshness checks)
• Inputs needed:
  • Configuration in config/kit.yml indicating the table name, timestamp_column, and SLA thresholds (warn_after, fail_after).
• Expected result:
  • The code queries the target database for the newest timestamp in the column.
  • It calculates lag (now - max_timestamp) and records a FreshnessRecord in the DB.
  • If the lag is too high, it sends an alert to the configured channel.
• Status: ✅ appears to work

---

F3: Running data quality checks

• What I checked: Checking data quality rules using Soda Core, custom SQL queries, consistency comparisons, and row count anomalies.
• Files and folders involved:
  • backend/routers/checks.py
  • checks/ (Where Soda Core YAML files are kept)
  • backend/models.py (CheckResult, VolumeRecord, BackfillEvent tables)
• Endpoints:
  • POST /checks/run (Runs quality checks)
  • POST /checks/volume (Runs volume checks)
• Inputs needed:
  • Soda Core YAML configurations, custom SQL queries in kit.yml, and historical row counts to detect volume anomalies.
• Expected result:
  • Soda Core: The code starts a subprocess to run soda scan, reads the JSON output, and logs the results.
  • Custom SQL: Executes SQL and checks if assertions pass (e.g. result == 0).
  • Volume checks: Compares row count to 7-day average. If there's a big change, it checks if it's a "backfill" (historical data imports) to avoid false alert alarms.
• Status:
  • Soda Core Scan: ⚠️ unclear (Requires Soda CLI to be installed on the machine; if missing, the scan fails).
  • Great Expectations: ❌ broken/missing (Not implemented in code; returns a dummy failure).
  • Custom SQL, Consistency & Volume: ✅ appears to work

---

F4: Detecting schema drift

• What I checked: Finding out if someone modified columns in the warehouse tables.
• Files and folders involved:
  • backend/routers/schema_diff.py
  • connectors/base.py (To fetch table metadata)
  • backend/models.py (SchemaSnapshot and SchemaDiff tables)
• Endpoints:
  • POST /schema/snapshot (Runs comparison)
• Inputs needed:
  • List of tables to watch in config/kit.yml under schema_drift:.
• Expected result:
  • Saves the current list of columns and types as a SchemaSnapshot.
  • Compares it to the previous snapshot. If columns are added, removed, or type is changed, it logs the change into SchemaDiff and alerts the user.
• Status: ✅ appears to work

---

F5: Data contract validation

• What I checked: Validating tables against a YAML schema contract defined by developers.
• Files and folders involved:
  • backend/routers/contracts.py
  • config/contracts/ (YAML files detailing the contracts)
• Endpoints:
  • POST /contracts/validate (Validates contract files)
• Inputs needed:
  • Contract YAML definitions stating column types, null limits, allowed values, min rows, and custom rules.
• Expected result:
  • Queries database to verify if all columns exist and have the correct types.
  • Performs counts on nulls, duplicate values, value ranges, and custom business rules.
  • Logs the validation results. If any check fails, triggers a contract violation alert.
• Status: ✅ appears to work

---

F6: Alert suppression and noise reduction

• What I checked: How the system avoids sending too many alerts during maintenance or when a check fails repeatedly.
• Files and folders involved:
  • backend/routers/suppressions.py (Mute endpoints)
  • backend/routers/alert_noise.py (Noise tracking endpoints)
  • alerts/base.py (Dispatches alerts and checks deduplication)
• Endpoints:
  • POST /suppress/ (To mute a table's alerts)
  • GET /alerts/noise/summary (See noisy alerts)
• Inputs needed:
  • Suppression details (table name, mute time, reason) and historical alert counts.
• Expected result:
  • Before sending an alert, the system checks if the table is muted or if a similar alert was sent recently (using an adaptive time window based on a calculated noise score).
  • If yes, the alert is skipped. If sent, it increments the noise counters.
• Status: ✅ appears to work

---

F7: Pipeline webhook integration

• What I checked: Receiving callback requests from Airflow or Prefect to log pipeline runs.
• Files and folders involved:
  • backend/routers/webhooks.py
  • backend/models.py (PipelineRun table)
• Endpoints:
  • POST /webhooks/airflow
  • POST /webhooks/prefect
• Inputs needed:
  • JSON payloads from Airflow or Prefect with DAG details and states.
• Expected result:
  • Parses the payload, normalizes status values, and inserts a PipelineRun record.
• Status: ⚠️ unclear

---

3. Major gaps and things that might break

• Great Expectations is missing: If someone configures engine: great_expectations in the config, it will fail because the runner is not coded.
• SQLite vs PostgreSQL locking: In PostgreSQL, the scheduler locks tasks so multiple servers don't run them at the same time. On SQLite, it doesn't do anything, which might cause double runs during testing.
• No alert retry logic: If Slack or Teams is down, the alert fails. It records it in AlertLog, but it won't try to send it again later.
• Soda Core CLI dependency: The quality check depends on having the soda command line tool installed on the server. If it isn't, the tests fail.

---

4. Things I couldn't test or verify

• Database performance: I couldn't test how fast the schema diff fetches column data if the database has thousands of tables.
• Third-party API rate limits: I am not sure if Slack or Teams will block us if we send hundreds of alerts before the noise throttling turns on.

---

5. Future Testing Considerations / Improvements

• Edge Case Coverage: Expand test scenarios to cover failure modes and edge cases, such as:
  • Database connection failures (bad DB connection).
  • Malformed or invalid configurations (bad config).
  • Schema type mismatches and wrong data types.
  • Other potential system failure modes.

Detailed Integration Testing Matrix

Step	Function	Detailed Description & Expected Outcome	File (click)	Line range
Task 1 – Ingestion & Freshness	poll_freshness	Executes freshness checks for every table defined in kit.yml. For each table it creates a FreshnessRecord with status: ok when the data is fresh. No alert is generated in this happy‑path scenario.	freshness.py	76‑84
Task 2 – Schema Drift (Initial Snapshot)	take_snapshot	Captures the current schema of each configured table and persists a SchemaSnapshot. Since this is the first run, no SchemaDiff records are produced.	schema_diff.py	21‑27
Task 3 – Schema Change & Drift Alert	take_snapshot (re‑run) + _trigger_schema_alert	After a deliberate schema modification (e.g., adding a column), a second take_snapshot detects the differences, creates one or more SchemaDiff entries and dispatches a drift alert via _trigger_schema_alert.	schema_diff.py & base.py	21‑27, 326‑336
Task 4 – Data Contract Validation	validate_contracts	Loads contract YAML files, validates column presence, data types, nullability, uniqueness, allowed values and any custom SQL rules. Persists a ContractValidationResult; if any rule fails, a contract‑failure alert is emitted.	contracts.py	83‑90
Task 5 – Alert Suppression & Noise Throttling	dispatch_alert (uses is_alert_suppressed & is_alert_deduped)	Before routing an alert, checks for a manual suppression window (is_alert_suppressed) and applies adaptive deduplication (is_alert_deduped) based on the current noise score. If allowed, the alert is sent and the noise record is refreshed.	base.py	328‑340
Supporting Helpers	is_alert_suppressed	Queries the CheckSuppression table to determine whether alerts for a given table are currently muted.	base.py	226‑236
	is_alert_deduped	Performs adaptive deduplication using the noise‑aware window, preventing duplicate alerts within the calculated timeframe.	base.py	254‑266
	_refresh_noise_record	After a successful alert dispatch, recomputes the noise score and updates AlertNoiseRecord for future deduplication calculations.	base.py	132‑213