Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,12 @@

All notable changes to this project are documented here.

## [Unreleased]

### Added

- Webhook alert integration tests for `WebhookAlertChannel` and router fan-out

## [0.1.0] - 2026-07-14

### Added
Expand Down
9 changes: 8 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ Scheduling options are documented in [`docs/scheduling.md`](docs/scheduling.md).
- [x] Failure triage runbook in [`docs/operations.md`](docs/operations.md)
- [x] Unit and integration tests with CI
- [x] Airflow DAG `dqo_contract_checks` for scheduled contract runs
- [ ] Webhook alert integration tests against mock server
- [x] Webhook alert integration tests against mock server

## Technology stack

Expand Down Expand Up @@ -161,6 +161,13 @@ Coverage includes contract loading, each check type, end-to-end runs, history pe

Complements [`production-data-pipeline`](https://github.com/br413/production-data-pipeline), which focuses on incremental ingestion and transformation. This repository isolates the quality and observability boundary.

## Writing

| Article | Topic |
|---------|-------|
| [Building a Production Data Pipeline with Incremental Loading and dbt](https://dev.to/bobby_ray_581732c715283b2/building-a-production-data-pipeline-with-incremental-loading-and-dbt-2e2c) | Incremental ingestion and medallion layering — companion pipeline repo |
| [Data Quality Contracts in Production Pipelines](https://dev.to/bobby_ray_581732c715283b2/data-quality-contracts-in-production-pipelines-without-a-separate-platform-team-f3) | YAML contracts, quarantine/DLQ stack, alert routing — uses this repo |

## Topics

`data-quality` · `data-observability` · `data-contracts` · `data-engineering` · `data-platform` · `airflow` · `python` · `schema-validation` · `monitoring` · `alerting`
Expand Down
90 changes: 89 additions & 1 deletion tests/test_alerts.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,15 @@
from src.dqo.alerts import AlertRouter, ConsoleAlertChannel, FileAlertChannel
from urllib.error import URLError
from unittest.mock import MagicMock, patch

import pytest

from src.dqo.alerts import (
AlertEvent,
AlertRouter,
ConsoleAlertChannel,
FileAlertChannel,
WebhookAlertChannel,
)
from src.dqo.models import CheckResult, CheckStatus, RunSummary, Severity
from datetime import datetime, timezone

Expand Down Expand Up @@ -30,3 +41,80 @@ def test_alert_router_writes_console_and_file(tmp_path, capsys) -> None:
captured = capsys.readouterr()
assert "CRITICAL" in captured.out
assert alert_file.read_text(encoding="utf-8").strip()


def test_webhook_alert_channel_posts_json_payload() -> None:
channel = WebhookAlertChannel("https://example.test/hook")
event = AlertEvent(
contract_name="orders",
run_id="run-alert-1",
check_type="nulls",
severity=Severity.CRITICAL,
message="missing order_total",
)

with patch("src.dqo.alerts.request.urlopen") as urlopen:
urlopen.return_value.__enter__ = MagicMock(return_value=MagicMock())
urlopen.return_value.__exit__ = MagicMock(return_value=False)
channel.send(event)

request_obj = urlopen.call_args.args[0]
assert request_obj.full_url == "https://example.test/hook"
assert request_obj.headers["Content-type"] == "application/json"
assert b'"contract_name": "orders"' in request_obj.data
assert b'"check_type": "nulls"' in request_obj.data
assert b'"severity": "critical"' in request_obj.data


def test_webhook_alert_channel_raises_on_delivery_failure() -> None:
channel = WebhookAlertChannel("https://example.test/hook")
event = AlertEvent(
contract_name="orders",
run_id="run-alert-1",
check_type="nulls",
severity=Severity.CRITICAL,
message="missing order_total",
)

with patch("src.dqo.alerts.request.urlopen", side_effect=URLError("network down")):
with pytest.raises(RuntimeError, match="webhook delivery failed"):
channel.send(event)


def test_alert_router_delivers_failed_checks_to_webhook() -> None:
router = AlertRouter([WebhookAlertChannel("https://example.test/hook")])

with patch("src.dqo.alerts.request.urlopen") as urlopen:
urlopen.return_value.__enter__ = MagicMock(return_value=MagicMock())
urlopen.return_value.__exit__ = MagicMock(return_value=False)
events = router.route(_failed_summary())

assert len(events) == 1
assert urlopen.call_count == 1
request_obj = urlopen.call_args.args[0]
assert b"missing order_total" in request_obj.data


def test_alert_router_skips_info_severity_for_webhook() -> None:
summary = RunSummary(
contract_name="orders",
run_id="run-alert-2",
started_at=datetime(2026, 7, 14, 10, 0, tzinfo=timezone.utc),
finished_at=datetime(2026, 7, 14, 10, 1, tzinfo=timezone.utc),
results=(
CheckResult(
contract_name="orders",
check_type="freshness",
status=CheckStatus.FAILED,
message="slightly stale",
severity=Severity.INFO,
),
),
)
router = AlertRouter([WebhookAlertChannel("https://example.test/hook")])

with patch("src.dqo.alerts.request.urlopen") as urlopen:
events = router.route(summary)

assert events == []
urlopen.assert_not_called()
Loading