gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #462 from christianlouis/copilot/add-webhook-support

Browse Source 
feat(webhooks): implement webhook support for external integrations
This commit is contained in:
Christian Krakau-Louis
2026-03-01 22:05:57 +01:00
committed by GitHub
parent 49eb2dd8e5 a85992ee79
commit 1b3f35d77d
12 changed files with 1071 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.demo
+4
View File
@@ -319,6 +319,10 @@ NOTIFY_ON_STARTUP=True
NOTIFY_ON_SHUTDOWN=False
NOTIFY_ON_FILE_PROCESSED=True
# Webhooks – Notify external systems via HTTP POST on document events.
# Individual webhooks (URL, events, secret) are managed via /api/webhooks/.
WEBHOOK_ENABLED=True
# Uptime Kuma
UPTIME_KUMA_URL=https://status.example.com/api/push/abcdef123456?status=up
UPTIME_KUMA_PING_INTERVAL=5
app/api/__init__.py
+2
View File
@@ -23,6 +23,7 @@ from app.api.url_upload import router as url_upload_router
# Import all the individual routers
from app.api.user import router as user_router
from app.api.webhooks import router as webhooks_router
# Set up logging
logger = logging.getLogger(__name__)
@@ -46,3 +47,4 @@ router.include_router(url_upload_router)
router.include_router(search_router)
router.include_router(queue_router)
router.include_router(saved_searches_router)
router.include_router(webhooks_router)
app/api/webhooks.py
+206
View File
@@ -0,0 +1,206 @@
"""API endpoints for managing webhook configurations.
Provides CRUD operations for webhook configs that notify external systems
when document events occur (``document.uploaded``, ``document.processed``,
``document.failed``).
"""
import json
import logging
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import WebhookConfig
from app.utils.webhook import VALID_EVENTS
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/webhooks", tags=["webhooks"])
DbSession = Annotated[Session, Depends(get_db)]
# ---------------------------------------------------------------------------
# Auth helper (reuse the pattern from settings API)
# ---------------------------------------------------------------------------
def _require_admin(request: Request) -> dict:
"""Ensure the caller is an admin. Raises 403 otherwise."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
return user
AdminUser = Annotated[dict, Depends(_require_admin)]
# ---------------------------------------------------------------------------
# Pydantic schemas
# ---------------------------------------------------------------------------
class WebhookCreate(BaseModel):
"""Schema for creating a new webhook configuration."""
url: str = Field(..., min_length=1, max_length=2048, description="Target URL for webhook delivery")
secret: str | None = Field(default=None, max_length=512, description="Shared secret for HMAC-SHA256 signatures")
events: list[str] = Field(..., min_length=1, description="List of events to subscribe to")
is_active: bool = Field(default=True, description="Whether the webhook is active")
description: str | None = Field(default=None, max_length=500, description="Optional human-readable description")
class WebhookUpdate(BaseModel):
"""Schema for updating an existing webhook configuration."""
url: str | None = Field(default=None, min_length=1, max_length=2048)
secret: str | None = Field(default=None, max_length=512)
events: list[str] | None = Field(default=None, min_length=1)
is_active: bool | None = None
description: str | None = Field(default=None, max_length=500)
class WebhookResponse(BaseModel):
"""Schema returned to clients (secret is never exposed)."""
id: int
url: str
events: list[str]
is_active: bool
description: str | None
has_secret: bool
model_config = {"from_attributes": True}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _validate_events(events: list[str]) -> None:
"""Raise 422 if any event name is not recognised."""
invalid = set(events) - VALID_EVENTS
if invalid:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Invalid event(s): {', '.join(sorted(invalid))}. Valid events: {', '.join(sorted(VALID_EVENTS))}",
)
def _to_response(cfg: WebhookConfig) -> dict[str, Any]:
"""Convert a DB model instance to a response dict."""
try:
events = json.loads(cfg.events)
except (json.JSONDecodeError, TypeError):
events = []
return {
"id": cfg.id,
"url": cfg.url,
"events": events,
"is_active": cfg.is_active,
"description": cfg.description,
"has_secret": cfg.secret is not None and len(cfg.secret) > 0,
}
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@router.get("/", summary="List all webhook configurations")
def list_webhooks(db: DbSession, _admin: AdminUser) -> list[dict[str, Any]]:
"""Return all webhook configurations. Secrets are never included."""
configs = db.query(WebhookConfig).order_by(WebhookConfig.id).all()
return [_to_response(c) for c in configs]
@router.get("/{webhook_id}", summary="Get a single webhook configuration")
def get_webhook(webhook_id: int, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Return a single webhook configuration by ID."""
cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
if not cfg:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
return _to_response(cfg)
@router.post("/", status_code=status.HTTP_201_CREATED, summary="Create a webhook configuration")
def create_webhook(body: WebhookCreate, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Create a new webhook configuration."""
_validate_events(body.events)
cfg = WebhookConfig(
url=body.url,
secret=body.secret,
events=json.dumps(sorted(body.events)),
is_active=body.is_active,
description=body.description,
)
try:
db.add(cfg)
db.commit()
db.refresh(cfg)
except Exception:
db.rollback()
raise
logger.info("Webhook %d created for events %s", cfg.id, body.events)
return _to_response(cfg)
@router.put("/{webhook_id}", summary="Update a webhook configuration")
def update_webhook(webhook_id: int, body: WebhookUpdate, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Update an existing webhook configuration. Only supplied fields are changed."""
cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
if not cfg:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
if body.url is not None:
cfg.url = body.url
if body.secret is not None:
cfg.secret = body.secret
if body.events is not None:
_validate_events(body.events)
cfg.events = json.dumps(sorted(body.events))
if body.is_active is not None:
cfg.is_active = body.is_active
if body.description is not None:
cfg.description = body.description
try:
db.commit()
db.refresh(cfg)
except Exception:
db.rollback()
raise
logger.info("Webhook %d updated", cfg.id)
return _to_response(cfg)
@router.delete("/{webhook_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a webhook configuration")
def delete_webhook(webhook_id: int, db: DbSession, _admin: AdminUser) -> None:
"""Delete a webhook configuration."""
cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
if not cfg:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
try:
db.delete(cfg)
db.commit()
except Exception:
db.rollback()
raise
logger.info("Webhook %d deleted", cfg.id)
@router.get("/events/", summary="List valid webhook event types")
def list_events(_admin: AdminUser) -> list[str]:
"""Return the list of valid event types that can be subscribed to."""
return sorted(VALID_EVENTS)
app/config.py
+6
View File
@@ -274,6 +274,12 @@ class Settings(BaseSettings):
description="Send notifications when files are successfully processed",
)
# Webhook settings
webhook_enabled: bool = Field(
default=True,
description="Enable webhook delivery for document events",
)
# File upload size limits (for security - see SECURITY_AUDIT.md)
max_upload_size: int = Field(
default=1073741824, # 1GB in bytes (1024 * 1024 * 1024)
app/models.py
+15
View File
@@ -143,3 +143,18 @@ class SavedSearch(Base):
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
__table_args__ = (UniqueConstraint("user_id", "name", name="unique_user_search_name"),)
class WebhookConfig(Base):
"""Webhook configuration for notifying external systems of document events."""
__tablename__ = "webhook_configs"
id = Column(Integer, primary_key=True, index=True)
url = Column(String, nullable=False) # Target URL for webhook delivery
secret = Column(String, nullable=True) # Shared secret for HMAC-SHA256 signature
events = Column(Text, nullable=False) # JSON list of subscribed events
is_active = Column(Boolean, default=True, nullable=False) # Whether the webhook is active
description = Column(String, nullable=True) # Optional human-readable description
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
app/tasks/webhook_tasks.py
+39
View File
@@ -0,0 +1,39 @@
"""Celery task for asynchronous webhook delivery with retry and backoff.
Uses :class:`~app.tasks.retry_config.BaseTaskWithRetry` so failed deliveries
are automatically retried with exponential backoff (default: 60 s, 300 s,
900 s) and ±20 % jitter.
"""
import logging
from typing import Any
from app.celery_app import celery
from app.tasks.retry_config import BaseTaskWithRetry
from app.utils.webhook import deliver_webhook
logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry, bind=True, name="webhook.deliver")
def deliver_webhook_task(self, url: str, payload: dict[str, Any], secret: str | None = None) -> dict[str, Any]:
"""Deliver a webhook payload to *url* with automatic retries.
Args:
url: Target webhook URL.
payload: The full webhook payload envelope.
secret: Optional shared secret for HMAC-SHA256 signing.
Returns:
A dict with ``status`` and ``url`` on success.
Raises:
RuntimeError: Re-raised to trigger Celery retry on delivery failure.
"""
logger.info("Delivering webhook to %s (attempt %d/%d)", url, self.request.retries + 1, self.max_retries + 1)
success = deliver_webhook(url, payload, secret)
if success:
return {"status": "delivered", "url": url}
raise RuntimeError(f"Webhook delivery to {url} failed")
app/utils/webhook.py
+165
View File
@@ -0,0 +1,165 @@
"""Webhook delivery utility for notifying external systems of document events.
Provides functions to dispatch webhook payloads with HMAC-SHA256 signatures
and to query active webhook configurations from the database.
Supported events:
- ``document.uploaded`` – a new document has been ingested
- ``document.processed`` – a document finished processing successfully
- ``document.failed`` – document processing failed
"""
import hashlib
import hmac
import json
import logging
import time
from typing import Any
import requests
from app.database import SessionLocal
from app.models import WebhookConfig
logger = logging.getLogger(__name__)
#: Events recognised by the webhook subsystem.
VALID_EVENTS: frozenset[str] = frozenset(
{
"document.uploaded",
"document.processed",
"document.failed",
}
)
#: Timeout (seconds) for outgoing webhook HTTP requests.
WEBHOOK_TIMEOUT = 10
def compute_signature(payload_bytes: bytes, secret: str) -> str:
"""Compute an HMAC-SHA256 hex-digest for *payload_bytes* using *secret*.
Args:
payload_bytes: The raw JSON body to sign.
secret: The shared secret string.
Returns:
``sha256=<hex-digest>`` signature string.
"""
mac = hmac.new(secret.encode("utf-8"), payload_bytes, hashlib.sha256)
return f"sha256={mac.hexdigest()}"
def deliver_webhook(url: str, payload: dict[str, Any], secret: str | None = None) -> bool:
"""Send a single webhook POST request.
Args:
url: Target URL.
payload: JSON-serialisable dictionary.
secret: If provided, an ``X-Webhook-Signature`` header is included.
Returns:
``True`` when the remote server responds with a 2xx status.
"""
body = json.dumps(payload, default=str, sort_keys=True)
body_bytes = body.encode("utf-8")
headers: dict[str, str] = {"Content-Type": "application/json"}
if secret:
headers["X-Webhook-Signature"] = compute_signature(body_bytes, secret)
try:
resp = requests.post(url, data=body_bytes, headers=headers, timeout=WEBHOOK_TIMEOUT)
if resp.ok:
logger.info("Webhook delivered to %s (status %d)", url, resp.status_code)
return True
logger.warning("Webhook to %s returned status %d", url, resp.status_code)
return False
except requests.RequestException as exc:
logger.error("Webhook delivery to %s failed: %s", url, exc)
return False
def build_payload(event: str, data: dict[str, Any]) -> dict[str, Any]:
"""Build a standardised webhook payload envelope.
Args:
event: The event name (e.g. ``document.uploaded``).
data: Event-specific data.
Returns:
Dictionary with ``event``, ``timestamp``, and ``data`` keys.
"""
return {
"event": event,
"timestamp": time.time(),
"data": data,
}
def get_active_webhooks_for_event(event: str) -> list[dict[str, Any]]:
"""Return all active webhook configs subscribed to *event*.
Queries the database directly so this helper can be called from both the
API layer and Celery tasks.
Args:
event: The event name to filter on.
Returns:
A list of dicts with ``id``, ``url``, ``secret``, and ``events`` keys.
"""
db = SessionLocal()
try:
configs = db.query(WebhookConfig).filter(WebhookConfig.is_active.is_(True)).all()
result: list[dict[str, Any]] = []
for cfg in configs:
try:
subscribed = json.loads(cfg.events)
except (json.JSONDecodeError, TypeError):
subscribed = []
if event in subscribed:
result.append(
{
"id": cfg.id,
"url": cfg.url,
"secret": cfg.secret,
"events": subscribed,
}
)
return result
finally:
db.close()
def dispatch_webhook_event(event: str, data: dict[str, Any]) -> None:
"""Fan-out a webhook event to all matching active configurations.
This is the main entry-point used by application code to trigger webhooks.
It delegates to :func:`deliver_webhook_task` (Celery) for each matching
webhook so delivery happens asynchronously with automatic retries.
Args:
event: Event name (must be in :data:`VALID_EVENTS`).
data: Event-specific payload data.
"""
if event not in VALID_EVENTS:
logger.warning("Ignoring unknown webhook event: %s", event)
return
webhooks = get_active_webhooks_for_event(event)
if not webhooks:
logger.debug("No active webhooks for event %s", event)
return
payload = build_payload(event, data)
# Import here to avoid circular dependency with celery_app
from app.tasks.webhook_tasks import deliver_webhook_task
for wh in webhooks:
try:
deliver_webhook_task.delay(wh["url"], payload, wh["secret"])
logger.debug("Queued webhook delivery to %s for event %s", wh["url"], event)
except Exception as exc:
logger.error("Failed to queue webhook to %s: %s", wh["url"], exc)
docs/API.md
+123
View File
@@ -704,6 +704,129 @@ Send a processed file to Google Drive.
}
```
## Webhooks
Manage webhook configurations for notifying external systems when document events occur. All webhook endpoints require admin access.
### Supported Events
| Event | Description |
|-------|-------------|
| `document.uploaded` | A new document has been ingested |
| `document.processed` | A document finished processing successfully |
| `document.failed` | Document processing failed |
### GET /api/webhooks/events/
List all valid webhook event types.
**Response (200):**
```json
["document.failed", "document.processed", "document.uploaded"]
```
### GET /api/webhooks/
List all webhook configurations. Secrets are never included in responses.
**Response (200):**
```json
[
{
"id": 1,
"url": "https://example.com/webhook",
"events": ["document.processed", "document.uploaded"],
"is_active": true,
"description": "Production webhook",
"has_secret": true
}
]
```
### POST /api/webhooks/
Create a new webhook configuration.
**Request:**
```json
{
"url": "https://example.com/webhook",
"secret": "my-shared-secret",
"events": ["document.uploaded", "document.processed", "document.failed"],
"is_active": true,
"description": "My integration"
}
```
**Response (201):**
```json
{
"id": 1,
"url": "https://example.com/webhook",
"events": ["document.failed", "document.processed", "document.uploaded"],
"is_active": true,
"description": "My integration",
"has_secret": true
}
```
### GET /api/webhooks/{webhook_id}
Get a single webhook configuration.
**Response (200):** Same shape as list items above.
### PUT /api/webhooks/{webhook_id}
Update an existing webhook. Only supplied fields are changed.
**Request:**
```json
{
"url": "https://new-url.example.com/webhook",
"is_active": false
}
```
### DELETE /api/webhooks/{webhook_id}
Delete a webhook configuration. Returns `204 No Content` on success.
### Webhook Payload Format
When a subscribed event occurs, a JSON POST request is sent to the configured URL:
```json
{
"event": "document.processed",
"timestamp": 1709322559.123456,
"data": {
"file_id": 42,
"filename": "invoice.pdf"
}
}
```
### HMAC Signature
If a secret is configured, an `X-Webhook-Signature` header is included with each request. The signature is computed as `sha256=<hex-digest>` using HMAC-SHA256 over the raw JSON body.
To verify in Python:
```python
import hashlib, hmac
def verify_signature(body: bytes, secret: str, signature: str) -> bool:
expected = "sha256=" + hmac.new(
secret.encode(), body, hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, signature)
```
### Retry Behavior
Failed deliveries (non-2xx responses or network errors) are automatically retried with exponential backoff: 60 s, 300 s, then 900 s (up to 3 retries with ±20 % jitter).
## Error Handling
Errors follow standard HTTP status codes with descriptive messages:
docs/ConfigurationGuide.md
+11
View File
@@ -800,6 +800,17 @@ For detailed setup instructions, see the [Amazon S3 Setup Guide](AmazonS3Setup.m
For detailed setup instructions, see the [Notifications Setup Guide](NotificationsSetup.md).
### Webhooks
Webhooks notify external systems via HTTP POST when document events occur.
Configurations are stored in the database and managed through the API (see [API docs](API.md#webhooks)).
| **Variable** | **Description** | **Default** |
|---------------------|------------------------------------------------------------------|-------------|
| `WEBHOOK_ENABLED` | Enable or disable webhook delivery globally (`True`/`False`) | `True` |
Webhook URLs, secrets, and subscribed events are configured per-webhook via the `/api/webhooks/` endpoints (admin access required). Each delivery includes an optional HMAC-SHA256 signature for verification and is retried with exponential backoff on failure.
### Uptime Kuma
| **Variable** | **Description** |
migrations/versions/009_add_webhook_configs.py
+37
View File
@@ -0,0 +1,37 @@
"""Add webhook_configs table for external integrations
Revision ID: 009_add_webhook_configs
Revises: 008_add_performance_indexes
Create Date: 2026-03-01
"""
from typing import Union
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "009_add_webhook_configs"
down_revision: Union[str, None] = "008_add_performance_indexes"
depends_on: Union[str, None] = None
def upgrade() -> None:
"""Create webhook_configs table."""
op.create_table(
"webhook_configs",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("url", sa.String(), nullable=False),
sa.Column("secret", sa.String(), nullable=True),
sa.Column("events", sa.Text(), nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("1")),
sa.Column("description", sa.String(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
def downgrade() -> None:
"""Drop webhook_configs table."""
op.drop_table("webhook_configs")
tests/conftest.py
+1 -1
View File
@@ -59,7 +59,7 @@ from app.database import Base # noqa: E402
from app.main import app as fastapi_app # noqa: E402
# Import models to register them with SQLAlchemy Base
from app.models import DocumentMetadata, FileRecord, ProcessingLog, SavedSearch # noqa: F401, E402
from app.models import DocumentMetadata, FileRecord, ProcessingLog, SavedSearch, WebhookConfig # noqa: F401, E402
@pytest.fixture(scope="session")
tests/test_webhooks.py
+462
View File
@@ -0,0 +1,462 @@
"""Tests for the webhook subsystem (utility, API, and task)."""
import hashlib
import hmac
import json
import time
from unittest.mock import MagicMock, patch
import pytest
from app.models import WebhookConfig
from app.utils.webhook import (
build_payload,
compute_signature,
deliver_webhook,
dispatch_webhook_event,
get_active_webhooks_for_event,
)
# ---------------------------------------------------------------------------
# Unit tests – compute_signature
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestComputeSignature:
"""Tests for HMAC-SHA256 signature generation."""
def test_signature_format(self):
"""Signature starts with 'sha256=' and is a hex digest."""
sig = compute_signature(b'{"event":"test"}', "mysecret")
assert sig.startswith("sha256=")
# hex digest should be 64 chars
assert len(sig.split("=")[1]) == 64
def test_signature_matches_manual_hmac(self):
"""Signature matches a manually computed HMAC-SHA256."""
payload = b'{"key":"value"}'
secret = "s3cret"
expected = "sha256=" + hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
assert compute_signature(payload, secret) == expected
def test_different_secrets_produce_different_signatures(self):
"""Different secrets must produce different signatures."""
payload = b"same"
assert compute_signature(payload, "secret1") != compute_signature(payload, "secret2")
# ---------------------------------------------------------------------------
# Unit tests – build_payload
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestBuildPayload:
"""Tests for the webhook payload builder."""
def test_contains_required_keys(self):
"""Payload contains event, timestamp, and data."""
payload = build_payload("document.uploaded", {"file_id": 1})
assert payload["event"] == "document.uploaded"
assert "timestamp" in payload
assert payload["data"] == {"file_id": 1}
def test_timestamp_is_recent(self):
"""Timestamp should be close to current time."""
before = time.time()
payload = build_payload("document.processed", {})
after = time.time()
assert before <= payload["timestamp"] <= after
# ---------------------------------------------------------------------------
# Unit tests – deliver_webhook
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestDeliverWebhook:
"""Tests for the HTTP delivery function."""
def test_success_returns_true(self, mocker):
"""A 200 response returns True."""
mock_post = mocker.patch("app.utils.webhook.requests.post")
mock_post.return_value = MagicMock(ok=True, status_code=200)
result = deliver_webhook("https://example.com/hook", {"event": "test"})
assert result is True
mock_post.assert_called_once()
def test_non_2xx_returns_false(self, mocker):
"""A non-2xx response returns False."""
mock_post = mocker.patch("app.utils.webhook.requests.post")
mock_post.return_value = MagicMock(ok=False, status_code=500)
result = deliver_webhook("https://example.com/hook", {"event": "test"})
assert result is False
def test_request_exception_returns_false(self, mocker):
"""A network error returns False."""
import requests
mocker.patch("app.utils.webhook.requests.post", side_effect=requests.ConnectionError("fail"))
result = deliver_webhook("https://example.com/hook", {"event": "test"})
assert result is False
def test_signature_header_included_when_secret(self, mocker):
"""X-Webhook-Signature header is present when a secret is supplied."""
mock_post = mocker.patch("app.utils.webhook.requests.post")
mock_post.return_value = MagicMock(ok=True, status_code=200)
deliver_webhook("https://example.com/hook", {"event": "test"}, secret="abc")
call_kwargs = mock_post.call_args
headers = call_kwargs.kwargs.get("headers") or call_kwargs[1].get("headers")
assert "X-Webhook-Signature" in headers
assert headers["X-Webhook-Signature"].startswith("sha256=")
def test_no_signature_header_without_secret(self, mocker):
"""X-Webhook-Signature header is absent when no secret is supplied."""
mock_post = mocker.patch("app.utils.webhook.requests.post")
mock_post.return_value = MagicMock(ok=True, status_code=200)
deliver_webhook("https://example.com/hook", {"event": "test"})
call_kwargs = mock_post.call_args
headers = call_kwargs.kwargs.get("headers") or call_kwargs[1].get("headers")
assert "X-Webhook-Signature" not in headers
# ---------------------------------------------------------------------------
# Unit tests – get_active_webhooks_for_event (DB)
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestGetActiveWebhooks:
"""Tests that active webhook configs are correctly filtered by event."""
def test_returns_matching_webhooks(self, db_session):
"""Only webhooks subscribed to the given event are returned."""
cfg1 = WebhookConfig(
url="https://a.com/hook",
secret="s1",
events=json.dumps(["document.uploaded", "document.processed"]),
is_active=True,
)
cfg2 = WebhookConfig(
url="https://b.com/hook",
events=json.dumps(["document.failed"]),
is_active=True,
)
db_session.add_all([cfg1, cfg2])
db_session.commit()
with patch("app.utils.webhook.SessionLocal", return_value=db_session):
results = get_active_webhooks_for_event("document.uploaded")
assert len(results) == 1
assert results[0]["url"] == "https://a.com/hook"
def test_inactive_webhooks_excluded(self, db_session):
"""Inactive webhooks are not returned."""
cfg = WebhookConfig(
url="https://inactive.com/hook",
events=json.dumps(["document.uploaded"]),
is_active=False,
)
db_session.add(cfg)
db_session.commit()
with patch("app.utils.webhook.SessionLocal", return_value=db_session):
results = get_active_webhooks_for_event("document.uploaded")
assert results == []
def test_returns_empty_when_no_match(self, db_session):
"""No webhooks returned when none match the event."""
cfg = WebhookConfig(
url="https://c.com/hook",
events=json.dumps(["document.failed"]),
is_active=True,
)
db_session.add(cfg)
db_session.commit()
with patch("app.utils.webhook.SessionLocal", return_value=db_session):
results = get_active_webhooks_for_event("document.uploaded")
assert results == []
# ---------------------------------------------------------------------------
# Unit tests – dispatch_webhook_event
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestDispatchWebhookEvent:
"""Tests for the high-level dispatch function."""
def test_unknown_event_is_ignored(self, mocker):
"""Unknown events are silently ignored."""
mock_get = mocker.patch("app.utils.webhook.get_active_webhooks_for_event")
dispatch_webhook_event("unknown.event", {})
mock_get.assert_not_called()
def test_no_webhooks_does_not_fail(self, mocker):
"""No error when there are no matching webhooks."""
mocker.patch("app.utils.webhook.get_active_webhooks_for_event", return_value=[])
dispatch_webhook_event("document.uploaded", {"file_id": 1})
def test_queues_celery_task_for_each_webhook(self, mocker):
"""A Celery task is queued for each matching webhook."""
mocker.patch(
"app.utils.webhook.get_active_webhooks_for_event",
return_value=[
{"id": 1, "url": "https://a.com", "secret": "s", "events": ["document.uploaded"]},
{"id": 2, "url": "https://b.com", "secret": None, "events": ["document.uploaded"]},
],
)
mock_task = mocker.patch("app.tasks.webhook_tasks.deliver_webhook_task.delay")
dispatch_webhook_event("document.uploaded", {"file_id": 42})
assert mock_task.call_count == 2
# ---------------------------------------------------------------------------
# Integration tests – API endpoints
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestWebhookAPI:
"""Tests for the /api/webhooks/ endpoints."""
def _with_admin(self, client):
"""Return headers / approach to authenticate as admin for test client."""
# FastAPI TestClient + SessionMiddleware: we can set session data by
# using the app's dependency override or the session directly.
# Simplest: override the _require_admin dependency.
from app.api.webhooks import _require_admin
client.app.dependency_overrides[_require_admin] = lambda: {"is_admin": True, "name": "test-admin"}
return client
def test_create_webhook(self, client):
"""POST /api/webhooks/ creates a new webhook."""
self._with_admin(client)
resp = client.post(
"/api/webhooks/",
json={
"url": "https://example.com/webhook",
"secret": "my-secret",
"events": ["document.uploaded"],
"description": "Test hook",
},
)
assert resp.status_code == 201
data = resp.json()
assert data["url"] == "https://example.com/webhook"
assert data["events"] == ["document.uploaded"]
assert data["is_active"] is True
assert data["has_secret"] is True
assert "secret" not in data # secret must not be exposed
def test_list_webhooks(self, client, db_session):
"""GET /api/webhooks/ returns all webhooks."""
self._with_admin(client)
# seed one
cfg = WebhookConfig(
url="https://list.example.com/hook",
events=json.dumps(["document.processed"]),
is_active=True,
)
db_session.add(cfg)
db_session.commit()
resp = client.get("/api/webhooks/")
assert resp.status_code == 200
items = resp.json()
assert any(w["url"] == "https://list.example.com/hook" for w in items)
def test_get_webhook(self, client, db_session):
"""GET /api/webhooks/{id} returns a single webhook."""
self._with_admin(client)
cfg = WebhookConfig(
url="https://get.example.com/hook",
events=json.dumps(["document.failed"]),
is_active=True,
)
db_session.add(cfg)
db_session.commit()
resp = client.get(f"/api/webhooks/{cfg.id}")
assert resp.status_code == 200
assert resp.json()["url"] == "https://get.example.com/hook"
def test_get_webhook_not_found(self, client):
"""GET /api/webhooks/9999 returns 404."""
self._with_admin(client)
resp = client.get("/api/webhooks/9999")
assert resp.status_code == 404
def test_update_webhook(self, client, db_session):
"""PUT /api/webhooks/{id} updates the webhook."""
self._with_admin(client)
cfg = WebhookConfig(
url="https://old.example.com/hook",
events=json.dumps(["document.uploaded"]),
is_active=True,
)
db_session.add(cfg)
db_session.commit()
resp = client.put(
f"/api/webhooks/{cfg.id}",
json={"url": "https://new.example.com/hook", "is_active": False},
)
assert resp.status_code == 200
data = resp.json()
assert data["url"] == "https://new.example.com/hook"
assert data["is_active"] is False
def test_update_webhook_not_found(self, client):
"""PUT /api/webhooks/9999 returns 404."""
self._with_admin(client)
resp = client.put("/api/webhooks/9999", json={"url": "https://x.com"})
assert resp.status_code == 404
def test_delete_webhook(self, client, db_session):
"""DELETE /api/webhooks/{id} removes the webhook."""
self._with_admin(client)
cfg = WebhookConfig(
url="https://del.example.com/hook",
events=json.dumps(["document.uploaded"]),
is_active=True,
)
db_session.add(cfg)
db_session.commit()
wid = cfg.id
resp = client.delete(f"/api/webhooks/{wid}")
assert resp.status_code == 204
# confirm it's gone
resp2 = client.get(f"/api/webhooks/{wid}")
assert resp2.status_code == 404
def test_delete_webhook_not_found(self, client):
"""DELETE /api/webhooks/9999 returns 404."""
self._with_admin(client)
resp = client.delete("/api/webhooks/9999")
assert resp.status_code == 404
def test_create_webhook_invalid_event(self, client):
"""POST /api/webhooks/ rejects invalid events."""
self._with_admin(client)
resp = client.post(
"/api/webhooks/",
json={"url": "https://x.com/hook", "events": ["bad.event"]},
)
assert resp.status_code == 422
def test_list_events(self, client):
"""GET /api/webhooks/events/ returns valid events."""
self._with_admin(client)
resp = client.get("/api/webhooks/events/")
assert resp.status_code == 200
events = resp.json()
assert "document.uploaded" in events
assert "document.processed" in events
assert "document.failed" in events
def test_requires_admin(self, client):
"""Endpoints return 403 without admin session."""
# Remove the admin override
from app.api.webhooks import _require_admin
client.app.dependency_overrides.pop(_require_admin, None)
resp = client.get("/api/webhooks/")
assert resp.status_code == 403
def test_create_webhook_without_secret(self, client):
"""POST /api/webhooks/ works without a secret."""
self._with_admin(client)
resp = client.post(
"/api/webhooks/",
json={"url": "https://nosecret.com/hook", "events": ["document.failed"]},
)
assert resp.status_code == 201
data = resp.json()
assert data["has_secret"] is False
def test_update_webhook_events(self, client, db_session):
"""PUT /api/webhooks/{id} can update events."""
self._with_admin(client)
cfg = WebhookConfig(
url="https://evup.example.com/hook",
events=json.dumps(["document.uploaded"]),
is_active=True,
)
db_session.add(cfg)
db_session.commit()
resp = client.put(
f"/api/webhooks/{cfg.id}",
json={"events": ["document.processed", "document.failed"]},
)
assert resp.status_code == 200
assert set(resp.json()["events"]) == {"document.processed", "document.failed"}
def test_update_webhook_invalid_event(self, client, db_session):
"""PUT /api/webhooks/{id} rejects invalid events."""
self._with_admin(client)
cfg = WebhookConfig(
url="https://bdev.example.com/hook",
events=json.dumps(["document.uploaded"]),
is_active=True,
)
db_session.add(cfg)
db_session.commit()
resp = client.put(
f"/api/webhooks/{cfg.id}",
json={"events": ["invalid.event"]},
)
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# Unit tests – Celery task
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestDeliverWebhookTask:
"""Tests for the Celery webhook delivery task."""
def test_success_returns_status_dict(self, mocker):
"""Task returns a dict on successful delivery."""
mocker.patch("app.tasks.webhook_tasks.deliver_webhook", return_value=True)
from app.tasks.webhook_tasks import deliver_webhook_task
# Mock the task's request context for the log line
deliver_webhook_task.request.retries = 0
result = deliver_webhook_task.__wrapped__("https://example.com/hook", {"event": "test"}, None)
assert result["status"] == "delivered"
assert result["url"] == "https://example.com/hook"
def test_failure_raises_for_retry(self, mocker):
"""Task raises RuntimeError on delivery failure to trigger retry."""
mocker.patch("app.tasks.webhook_tasks.deliver_webhook", return_value=False)
from app.tasks.webhook_tasks import deliver_webhook_task
deliver_webhook_task.request.retries = 0
with pytest.raises(RuntimeError, match="Webhook delivery.*failed"):
deliver_webhook_task.__wrapped__("https://example.com/hook", {"event": "test"}, None)