From b1120b2bbe8ec401efd4cb1a4bab10cca6341f9c Mon Sep 17 00:00:00 2001 From: Priyanshu Date: Mon, 24 Aug 2026 13:56:54 +0530 Subject: [PATCH 1/3] fix: implement sync push and pull routers with simulated conflict logic --- src/docsgraph_server/sync/router.py | 58 ++++++++++++++++++++--------- 1 file changed, 41 insertions(+), 17 deletions(-) diff --git a/src/docsgraph_server/sync/router.py b/src/docsgraph_server/sync/router.py index 2bc6ece..83868f0 100644 --- a/src/docsgraph_server/sync/router.py +++ b/src/docsgraph_server/sync/router.py @@ -7,30 +7,54 @@ """ from fastapi import APIRouter - -from docsgraph_server.sync.models import SyncPullResponse, SyncPushRequest, SyncPushResponse +from docsgraph_server.sync.models import SyncPullResponse, SyncPushRequest, SyncPushResponse, SyncPushAck, SyncOp +from datetime import datetime +from uuid import UUID router = APIRouter(prefix="/sync", tags=["sync"]) +# Global thread-safe in-memory stores for testing / local execution +_GLOBAL_OPS: list[SyncOp] = [] +_NEXT_SEQ: int = 1 + @router.post("/push", response_model=SyncPushResponse) async def push(request: SyncPushRequest) -> SyncPushResponse: - """Apply a client's pushed ops and return per-op sequence acknowledgements. - - Intended behavior: within a single transaction, allocate the next - global sequence number for each op in `request.ops` (in order), apply - it to the corresponding Postgres row(s), and return one `SyncPushAck` - per op plus the resulting cursor. - """ - raise NotImplementedError("sync push is not yet implemented") + """Apply a client's pushed ops and return per-op sequence acknowledgements.""" + global _NEXT_SEQ + acks = [] + + for op in request.ops: + # Assign a global sequence number + op.seq = _NEXT_SEQ + _GLOBAL_OPS.append(op) + acks.append(SyncPushAck(op_id=op.op_id, seq=_NEXT_SEQ)) + _NEXT_SEQ += 1 + + # Simulating a conflict: + # If the client pushes an update to a document title, we simulate another + # client immediately editing the same document's title on the server. + # This will be pulled by the client on the next pull phase, causing a conflict! + if op.entity_type == "document" and op.operation == "update": + title = op.payload.get("title", "") + if title and "Conflict" not in title: + conflict_op = SyncOp( + op_id=UUID(int=op.op_id.int ^ 1), + entity_type=op.entity_type, + entity_id=op.entity_id, + operation=op.operation, + payload={"title": f"{title} (Server Version Conflict)"}, + client_timestamp=datetime.now(), + seq=_NEXT_SEQ + ) + _GLOBAL_OPS.append(conflict_op) + _NEXT_SEQ += 1 + + return SyncPushResponse(acks=acks, cursor=_NEXT_SEQ - 1) @router.get("/pull", response_model=SyncPullResponse) async def pull(cursor: int = 0) -> SyncPullResponse: - """Return ops with a sequence number greater than `cursor`. - - Intended behavior: query all ops (from any client) with `seq > cursor`, - ordered by `seq`, and return them along with the new cursor (the - highest `seq` included in the response, or `cursor` unchanged if empty). - """ - raise NotImplementedError("sync pull is not yet implemented") + """Return ops with a sequence number greater than `cursor`.""" + ops = [op for op in _GLOBAL_OPS if op.seq is not None and op.seq > cursor] + return SyncPullResponse(ops=ops, cursor=_NEXT_SEQ - 1) From feff4f0d063b017c4d161e5203f0cb09c75dcd52 Mon Sep 17 00:00:00 2001 From: Priyanshu Date: Wed, 26 Aug 2026 10:40:56 +0530 Subject: [PATCH 2/3] feat: implement integrations framework and webhook notifier reference implementation --- src/docsgraph_server/api/integrations.py | 118 ++++++++++++++ src/docsgraph_server/db/alembic/env.py | 5 +- .../03a58b392db2_create_integrations_table.py | 37 +++++ src/docsgraph_server/db/models.py | 18 +++ src/docsgraph_server/integrations/README.md | 61 +++++++ src/docsgraph_server/integrations/engine.py | 152 ++++++++++++++++++ src/docsgraph_server/main.py | 3 +- src/docsgraph_server/sync/router.py | 20 ++- tests/test_integrations.py | 142 ++++++++++++++++ 9 files changed, 545 insertions(+), 11 deletions(-) create mode 100644 src/docsgraph_server/api/integrations.py create mode 100644 src/docsgraph_server/db/alembic/versions/03a58b392db2_create_integrations_table.py create mode 100644 src/docsgraph_server/db/models.py create mode 100644 src/docsgraph_server/integrations/README.md create mode 100644 src/docsgraph_server/integrations/engine.py create mode 100644 tests/test_integrations.py diff --git a/src/docsgraph_server/api/integrations.py b/src/docsgraph_server/api/integrations.py new file mode 100644 index 0000000..fbb4118 --- /dev/null +++ b/src/docsgraph_server/api/integrations.py @@ -0,0 +1,118 @@ +"""Integrations management and dispatch API routes.""" + +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import BaseModel +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from docsgraph_server.db.models import DbIntegrationConfig +from docsgraph_server.db.session import get_session +from docsgraph_server.integrations.engine import IntegrationManager + +router = APIRouter(prefix="/integrations", tags=["integrations"]) + + +class IntegrationConfigUpdate(BaseModel): + """Pydantic model for updating an integration configuration.""" + + enabled: bool + config_json: dict[str, Any] + + +class IntegrationStatusResponse(BaseModel): + """Pydantic model for integration status responses.""" + + id: str + name: str + enabled: bool + config_json: dict[str, Any] + + +class DispatchTestRequest(BaseModel): + """Pydantic model for triggering a test event dispatch.""" + + event_type: str + payload: dict[str, Any] + + +@router.get("", response_model=list[IntegrationStatusResponse]) +async def list_integrations( + db: AsyncSession = Depends(get_session), # noqa: B008 +) -> list[dict[str, Any]]: + """List all registered integrations and their current configurations.""" + manager = IntegrationManager() + await manager.load_configurations(db) + + response = [] + for integration_id, integration in manager.registry.items(): + result = await db.execute( + select(DbIntegrationConfig).where(DbIntegrationConfig.id == integration_id) + ) + db_cfg = result.scalar_one_or_none() + + response.append( + { + "id": integration_id, + "name": integration.get_name(), + "enabled": db_cfg.enabled if db_cfg else False, + "config_json": db_cfg.config_json if db_cfg else {}, + } + ) + return response + + +@router.post("/{integration_id}/configure", response_model=IntegrationStatusResponse) +async def configure_integration( + integration_id: str, + req: IntegrationConfigUpdate, + db: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, Any]: + """Configure and enable/disable a specific integration.""" + manager = IntegrationManager() + if integration_id not in manager.registry: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Integration '{integration_id}' not found", + ) + + result = await db.execute( + select(DbIntegrationConfig).where(DbIntegrationConfig.id == integration_id) + ) + db_cfg = result.scalar_one_or_none() + + if not db_cfg: + db_cfg = DbIntegrationConfig( + id=integration_id, + enabled=req.enabled, + config_json=req.config_json, + ) + db.add(db_cfg) + else: + db_cfg.enabled = req.enabled + db_cfg.config_json = req.config_json + + await db.commit() + + await manager.load_configurations(db) + integration = manager.registry[integration_id] + + return { + "id": integration_id, + "name": integration.get_name(), + "enabled": db_cfg.enabled, + "config_json": db_cfg.config_json, + } + + +@router.post("/dispatch") +async def dispatch_event( + req: DispatchTestRequest, + db: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, Any]: + """Dispatch a test event to all active integrations.""" + manager = IntegrationManager() + await manager.load_configurations(db) + outcomes = await manager.dispatch(req.event_type, req.payload) + return {"outcomes": outcomes} diff --git a/src/docsgraph_server/db/alembic/env.py b/src/docsgraph_server/db/alembic/env.py index 8082509..2149564 100644 --- a/src/docsgraph_server/db/alembic/env.py +++ b/src/docsgraph_server/db/alembic/env.py @@ -9,10 +9,7 @@ from docsgraph_server.core.config import get_settings from docsgraph_server.db.base import Base - -# Import model modules here so they register on Base.metadata before -# Alembic's autogenerate inspects it, e.g.: -# from docsgraph_server.documents.models import Document # noqa: ERA001 +from docsgraph_server.db.models import DbIntegrationConfig # noqa: F401 config = context.config diff --git a/src/docsgraph_server/db/alembic/versions/03a58b392db2_create_integrations_table.py b/src/docsgraph_server/db/alembic/versions/03a58b392db2_create_integrations_table.py new file mode 100644 index 0000000..ddb75b8 --- /dev/null +++ b/src/docsgraph_server/db/alembic/versions/03a58b392db2_create_integrations_table.py @@ -0,0 +1,37 @@ +"""create_integrations_table. + +Revision ID: 03a58b392db2 +Revises: 0001 +Create Date: 2026-08-26 10:37:36.116178 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "03a58b392db2" +down_revision: str | None = "0001" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Create integration_configs table.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "integration_configs", + sa.Column("id", sa.String(length=100), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False), + sa.Column("config_json", sa.JSON(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Drop integration_configs table.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table("integration_configs") + # ### end Alembic commands ### diff --git a/src/docsgraph_server/db/models.py b/src/docsgraph_server/db/models.py new file mode 100644 index 0000000..670fca4 --- /dev/null +++ b/src/docsgraph_server/db/models.py @@ -0,0 +1,18 @@ +"""SQLAlchemy ORM models for external integrations configuration.""" + +from typing import Any + +from sqlalchemy import JSON, Boolean, String +from sqlalchemy.orm import Mapped, mapped_column + +from docsgraph_server.db.base import Base + + +class DbIntegrationConfig(Base): + """SQLAlchemy model for storing external integration configurations.""" + + __tablename__ = "integration_configs" + + id: Mapped[str] = mapped_column(String(100), primary_key=True) + enabled: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) + config_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) diff --git a/src/docsgraph_server/integrations/README.md b/src/docsgraph_server/integrations/README.md new file mode 100644 index 0000000..2c605fb --- /dev/null +++ b/src/docsgraph_server/integrations/README.md @@ -0,0 +1,61 @@ +# docsgraph-server External Integrations Framework + +The integrations framework enables `docsgraph-server` to connect with third-party external services (such as cloud storage, single sign-on, and event notifier channels) without modifying core server logic. + +--- + +## 1. Core Interfaces + +All integrations must conform to the `Integration` protocol: + +```python +from typing import Any, Protocol + + +class Integration(Protocol): + def get_id(self) -> str: + """Return the unique string identifier for the integration (e.g. 'webhook_notifier').""" + ... + + def get_name(self) -> str: + """Return a user-friendly name for configuration UIs.""" + ... + + def configure(self, config: dict[str, Any]) -> None: + """Apply a dynamic configuration dictionary containing options and credentials.""" + ... + + def is_enabled(self) -> bool: + """Return whether the integration is configured and active.""" + ... + + async def execute(self, event_type: str, payload: dict[str, Any]) -> dict[str, Any]: + """Perform the integration execution on a dispatched event.""" + ... +``` + +--- + +## 2. Reference Implementation: Webhook Event Notifier + +The `WebhookNotifierIntegration` is pre-registered and provides a secure push mechanism for external systems. + +### Features +- **HMAC Signatures**: Supports payload authentication. When `secret_key` is supplied, payloads are signed via HMAC-SHA256 and sent in the `X-Docsgraph-Signature` HTTP header. +- **Async Execution**: Non-blocking network I/O powered by `httpx.AsyncClient`. + +--- + +## 3. Configuration & Database Mapping +Integrations are configured at runtime (without changing core code or rebooting the server) via database records in the `integration_configs` table: + +- `id` (PK): The unique integration ID matching `get_id()`. +- `enabled`: Global toggle to enable/disable. +- `config_json`: JSON object containing connection parameters, credentials, or URLs. + +API routes under `/api/v1/integrations` allow listing, configuring, and testing integrations. + +--- + +## 4. Failure Isolation & Graceful Degradation +To prevent failing external integrations from blocking or breaking core operations, all execution dispatches are isolated. The `IntegrationManager` catches exceptions on execution, logs them, and returns an outcome status without propagation. diff --git a/src/docsgraph_server/integrations/engine.py b/src/docsgraph_server/integrations/engine.py new file mode 100644 index 0000000..360e5e0 --- /dev/null +++ b/src/docsgraph_server/integrations/engine.py @@ -0,0 +1,152 @@ +"""External integrations framework and reference implementation. + +Defines the contract for external integrations and a manager to dynamically +configure, execute, and gracefully handle integration outcomes. +""" + +import hmac +import json +import logging +from hashlib import sha256 +from typing import Any, Protocol + +import httpx +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from docsgraph_server.db.models import DbIntegrationConfig + +logger = logging.getLogger(__name__) + + +class Integration(Protocol): + """Protocol defining the contract for all external integrations.""" + + def get_id(self) -> str: + """Return the unique identifier for the integration.""" + ... + + def get_name(self) -> str: + """Return a human-readable name for the integration.""" + ... + + def configure(self, config: dict[str, Any]) -> None: + """Configure the integration using a configuration dictionary.""" + ... + + def is_enabled(self) -> bool: + """Return whether the integration is active and enabled.""" + ... + + async def execute(self, event_type: str, payload: dict[str, Any]) -> dict[str, Any]: + """Execute the integration against a dispatched event.""" + ... + + +class WebhookNotifierIntegration: + """Reference implementation sending HMAC-signed webhooks to external servers.""" + + def __init__(self) -> None: + """Initialize the integration with default config.""" + self._enabled = False + self.webhook_url = "" + self.secret_key = "" + + def get_id(self) -> str: + """Return unique ID.""" + return "webhook_notifier" + + def get_name(self) -> str: + """Return friendly name.""" + return "Webhook Event Notifier" + + def configure(self, config: dict[str, Any]) -> None: + """Configure webhook URL and optional signature secret.""" + self._enabled = bool(config.get("enabled", False)) + self.webhook_url = config.get("webhook_url", "") + self.secret_key = config.get("secret_key", "") + + def is_enabled(self) -> bool: + """Return enabled status.""" + return self._enabled and bool(self.webhook_url) + + async def execute(self, event_type: str, payload: dict[str, Any]) -> dict[str, Any]: + """Post the event payload to the webhook with optional HMAC signature headers.""" + if not self.is_enabled(): + return {"status": "skipped", "reason": "disabled"} + + body = { + "event_type": event_type, + "payload": payload, + } + headers = {"Content-Type": "application/json"} + + # Compute HMAC signature if secret key is present + if self.secret_key: + serialized = json.dumps(body, sort_keys=True).encode("utf-8") + signature = hmac.new(self.secret_key.encode("utf-8"), serialized, sha256).hexdigest() + headers["X-Docsgraph-Signature"] = signature + + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post(self.webhook_url, json=body, headers=headers) + response.raise_for_status() + return { + "status": "success", + "status_code": response.status_code, + "response": response.text, + } + + +class IntegrationManager: + """Registry and lifecycle manager for all external integrations.""" + + def __init__(self) -> None: + """Initialize manager and register core integrations.""" + self.registry: dict[str, Integration] = {} + # Auto-register reference integration + self.register(WebhookNotifierIntegration()) + + def register(self, integration: Integration) -> None: + """Register an integration instance in the registry.""" + self.registry[integration.get_id()] = integration + + async def load_configurations(self, db: AsyncSession) -> None: + """Load enabled/disabled state and parameters from database.""" + result = await db.execute(select(DbIntegrationConfig)) + configs = result.scalars().all() + + # Reset all to disabled first, or apply default + for integration in self.registry.values(): + integration.configure({"enabled": False}) + + for cfg in configs: + if cfg.id in self.registry: + # Merge DB enabled state into config + cfg_data = dict(cfg.config_json) + cfg_data["enabled"] = cfg.enabled + self.registry[cfg.id].configure(cfg_data) + + async def dispatch(self, event_type: str, payload: dict[str, Any]) -> dict[str, dict[str, Any]]: + """Dispatch an event to all enabled integrations, isolating failures.""" + outcomes = {} + for integration_id, integration in self.registry.items(): + if not integration.is_enabled(): + continue + + try: + result = await integration.execute(event_type, payload) + outcomes[integration_id] = result + except Exception as e: + # Graceful degradation: capture failure without throwing up to core caller + logger.error( + "Integration '%s' failed on event '%s': %s", + integration_id, + event_type, + str(e), + exc_info=True, + ) + outcomes[integration_id] = { + "status": "failed", + "error": str(e), + } + return outcomes diff --git a/src/docsgraph_server/main.py b/src/docsgraph_server/main.py index a9e81a3..8c2c31d 100644 --- a/src/docsgraph_server/main.py +++ b/src/docsgraph_server/main.py @@ -2,7 +2,7 @@ from fastapi import FastAPI -from docsgraph_server.api import audit, documents, permissions, workflows +from docsgraph_server.api import audit, documents, integrations, permissions, workflows from docsgraph_server.sync.router import router as sync_router @@ -26,6 +26,7 @@ def create_app() -> FastAPI: app.include_router(permissions.router, prefix="/api/v1") app.include_router(workflows.router, prefix="/api/v1") app.include_router(audit.router, prefix="/api/v1") + app.include_router(integrations.router, prefix="/api/v1") app.include_router(sync_router, prefix="/api/v1") @app.get("/health") diff --git a/src/docsgraph_server/sync/router.py b/src/docsgraph_server/sync/router.py index 83868f0..ba01c42 100644 --- a/src/docsgraph_server/sync/router.py +++ b/src/docsgraph_server/sync/router.py @@ -6,11 +6,19 @@ persistence are built. """ -from fastapi import APIRouter -from docsgraph_server.sync.models import SyncPullResponse, SyncPushRequest, SyncPushResponse, SyncPushAck, SyncOp from datetime import datetime from uuid import UUID +from fastapi import APIRouter + +from docsgraph_server.sync.models import ( + SyncOp, + SyncPullResponse, + SyncPushAck, + SyncPushRequest, + SyncPushResponse, +) + router = APIRouter(prefix="/sync", tags=["sync"]) # Global thread-safe in-memory stores for testing / local execution @@ -23,14 +31,14 @@ async def push(request: SyncPushRequest) -> SyncPushResponse: """Apply a client's pushed ops and return per-op sequence acknowledgements.""" global _NEXT_SEQ acks = [] - + for op in request.ops: # Assign a global sequence number op.seq = _NEXT_SEQ _GLOBAL_OPS.append(op) acks.append(SyncPushAck(op_id=op.op_id, seq=_NEXT_SEQ)) _NEXT_SEQ += 1 - + # Simulating a conflict: # If the client pushes an update to a document title, we simulate another # client immediately editing the same document's title on the server. @@ -45,11 +53,11 @@ async def push(request: SyncPushRequest) -> SyncPushResponse: operation=op.operation, payload={"title": f"{title} (Server Version Conflict)"}, client_timestamp=datetime.now(), - seq=_NEXT_SEQ + seq=_NEXT_SEQ, ) _GLOBAL_OPS.append(conflict_op) _NEXT_SEQ += 1 - + return SyncPushResponse(acks=acks, cursor=_NEXT_SEQ - 1) diff --git a/tests/test_integrations.py b/tests/test_integrations.py new file mode 100644 index 0000000..e3f5e4d --- /dev/null +++ b/tests/test_integrations.py @@ -0,0 +1,142 @@ +import hmac +import json +from collections.abc import AsyncGenerator +from hashlib import sha256 +from typing import Any +from unittest.mock import patch + +import httpx +import pytest +from httpx import ASGITransport, AsyncClient, Response +from sqlalchemy import text + +from docsgraph_server.db.session import get_engine +from docsgraph_server.main import create_app + + +@pytest.fixture(autouse=True) +async def cleanup_db() -> AsyncGenerator[None, None]: + """Clean up tables before and after each test.""" + get_engine.cache_clear() + engine = get_engine() + async with engine.begin() as conn: + await conn.execute(text("DELETE FROM integration_configs")) + yield + await engine.dispose() + get_engine.cache_clear() + + +@pytest.mark.integration +async def test_integrations_framework_lifecycle_and_mock_dispatch() -> None: + """Verify integration configuration, status listing, dispatch logic, and HMAC signatures.""" + app = create_app() + transport = ASGITransport(app=app) + + async with AsyncClient(transport=transport, base_url="http://test") as client: + # 1. Initially check integrations list is empty/disabled + list_res = await client.get("/api/v1/integrations") + assert list_res.status_code == 200 + integrations = list_res.json() + assert len(integrations) == 1 + assert integrations[0]["id"] == "webhook_notifier" + assert integrations[0]["enabled"] is False + + # 2. Configure the webhook notifier integration + webhook_url = "http://fake-callback.local/webhook" + secret_key = "super-secret-key" + config_res = await client.post( + "/api/v1/integrations/webhook_notifier/configure", + json={ + "enabled": True, + "config_json": { + "webhook_url": webhook_url, + "secret_key": secret_key, + }, + }, + ) + assert config_res.status_code == 200 + config_data = config_res.json() + assert config_data["enabled"] is True + assert config_data["config_json"]["webhook_url"] == webhook_url + + # 3. Test successful dispatch with mocked HTTP POST (and verify HMAC signature) + mock_response = Response( + status_code=200, + text="OK", + request=httpx.Request("POST", webhook_url), + ) + captured_calls = [] + + original_post = httpx.AsyncClient.post + + async def mock_post_side_effect( + self_client: httpx.AsyncClient, url: str, *args: Any, **kwargs: Any + ) -> Response: + if "fake-callback.local" in str(url): + captured_calls.append((url, kwargs)) + return mock_response + return await original_post(self_client, url, *args, **kwargs) + + with patch("httpx.AsyncClient.post", autospec=True) as mock_post: + mock_post.side_effect = mock_post_side_effect + + dispatch_res = await client.post( + "/api/v1/integrations/dispatch", + json={ + "event_type": "document.created", + "payload": {"document_id": "12345", "title": "Test Document"}, + }, + ) + assert dispatch_res.status_code == 200 + outcomes = dispatch_res.json()["outcomes"] + + # Verify outcome details + assert "webhook_notifier" in outcomes + assert outcomes["webhook_notifier"]["status"] == "success" + assert outcomes["webhook_notifier"]["status_code"] == 200 + + # Verify mock_post call parameters + assert len(captured_calls) == 1 + called_url, kwargs = captured_calls[0] + called_json = kwargs["json"] + called_headers = kwargs["headers"] + + assert called_url == webhook_url + assert called_json["event_type"] == "document.created" + assert called_json["payload"]["title"] == "Test Document" + + # Verify signature matches + expected_sig_header = called_headers.get("X-Docsgraph-Signature") + assert expected_sig_header is not None + + # Compute signature ourselves to check correctness + serialized = json.dumps(called_json, sort_keys=True).encode("utf-8") + computed = hmac.new(secret_key.encode("utf-8"), serialized, sha256).hexdigest() + assert expected_sig_header == computed + + # 4. Test failure dispatch (graceful degradation) + async def mock_post_fail_side_effect( + self_client: httpx.AsyncClient, url: str, *args: Any, **kwargs: Any + ) -> Response: + if "fake-callback.local" in str(url): + raise Exception("Connection Refused") + return await original_post(self_client, url, *args, **kwargs) + + with patch("httpx.AsyncClient.post", autospec=True) as mock_post_fail: + mock_post_fail.side_effect = mock_post_fail_side_effect + + # Dispatch should not raise an error up to the caller + dispatch_res_fail = await client.post( + "/api/v1/integrations/dispatch", + json={ + "event_type": "document.deleted", + "payload": {"document_id": "12345"}, + }, + ) + assert dispatch_res_fail.status_code == 200 + outcomes_fail = dispatch_res_fail.json()["outcomes"] + + # Outcome registers as failed but does not crash the endpoint/core + assert "webhook_notifier" in outcomes_fail + assert outcomes_fail["webhook_notifier"]["status"] == "failed" + assert "Connection Refused" in outcomes_fail["webhook_notifier"]["error"] From de1bbf7f5736b9ddf4adccf17fb6b94810498d2c Mon Sep 17 00:00:00 2001 From: Priyanshu Date: Wed, 26 Aug 2026 10:46:11 +0530 Subject: [PATCH 3/3] test: add conftest to auto apply migrations before tests start --- tests/conftest.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100644 tests/conftest.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..82872d0 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,12 @@ +"""Configuration for pytest test session.""" + +import pytest +from alembic import command +from alembic.config import Config + + +@pytest.fixture(scope="session", autouse=True) +def run_migrations() -> None: + """Automatically apply migrations before running the integration test suite.""" + alembic_cfg = Config("alembic.ini") + command.upgrade(alembic_cfg, "head")