feat: v6.5.1 — 7 tests webhooks_v2 dé-skipés (0 skip, 749 verts) + roadmap rattrapée (sync.py Bearer coché)
This commit is contained in:
+221
-33
@@ -1,18 +1,26 @@
|
||||
"""FlowDeck — v6.4.0 Webhooks v2 tests (HMAC, retries, +20 events)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from app.db import get_conn, init_db
|
||||
from app.services import webhook_outbound
|
||||
from app.services.webhook_outbound import (
|
||||
EVENTS,
|
||||
MAX_ATTEMPTS,
|
||||
RETRY_DELAYS,
|
||||
_deliver_once,
|
||||
_event_matches,
|
||||
init_webhook_tables,
|
||||
deliver_to_sub,
|
||||
fire_event,
|
||||
retry_due_deliveries,
|
||||
sign_payload,
|
||||
verify_signature,
|
||||
)
|
||||
@@ -21,9 +29,38 @@ from app.services.webhook_outbound import (
|
||||
|
||||
@pytest.fixture
|
||||
def db():
|
||||
"""Ensure webhook tables exist."""
|
||||
init_webhook_tables()
|
||||
return True
|
||||
"""Fresh isolated SQLite database with the full FlowDeck schema.
|
||||
|
||||
Creates ``webhook_subscriptions`` + ``webhook_deliveries`` (versioned
|
||||
migrations) so delivery journal rows can be asserted, then restores the
|
||||
previous settings — safe under ``pytest -n auto``.
|
||||
"""
|
||||
tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
||||
db_path = tmp.name
|
||||
tmp.close()
|
||||
|
||||
prev_env = os.environ.get("DATABASE_URL")
|
||||
os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
|
||||
|
||||
import app.config
|
||||
settings_obj = app.config.settings
|
||||
prev_url = settings_obj.database_url
|
||||
settings_obj.database_url = f"sqlite:///{db_path}"
|
||||
|
||||
init_db()
|
||||
|
||||
yield
|
||||
|
||||
settings_obj.database_url = prev_url
|
||||
if prev_env is None:
|
||||
os.environ.pop("DATABASE_URL", None)
|
||||
else:
|
||||
os.environ["DATABASE_URL"] = prev_env
|
||||
for suffix in ("", "-wal", "-shm"):
|
||||
try:
|
||||
os.unlink(db_path + suffix)
|
||||
except (PermissionError, OSError):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -280,53 +317,204 @@ async def test_deliver_once_request_error():
|
||||
assert "Request error" in error
|
||||
|
||||
|
||||
# ── Integration Tests (Skipped if dependencies are missing) ────────────────────
|
||||
# ── Integration Tests (isolated DB + mocked HTTP) ───────────────────────────
|
||||
|
||||
|
||||
def _add_sub(url: str = "https://receiver.example/hook", event: str = "*",
|
||||
secret: str = "test_secret") -> int:
|
||||
"""Insert an active subscription and return its id."""
|
||||
with get_conn() as conn:
|
||||
cur = conn.execute(
|
||||
"INSERT INTO webhook_subscriptions (url, event, secret, active) VALUES (?, ?, ?, 1)",
|
||||
(url, event, secret),
|
||||
)
|
||||
conn.commit()
|
||||
return cur.lastrowid
|
||||
|
||||
|
||||
def _journal() -> list[dict]:
|
||||
"""All delivery-journal rows, oldest first."""
|
||||
with get_conn() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM webhook_deliveries ORDER BY id"
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def _patch_async_client(monkeypatch, handler) -> None:
|
||||
"""Route every ``httpx.AsyncClient`` built by webhook_outbound to a MockTransport."""
|
||||
real_client = httpx.AsyncClient
|
||||
transport = httpx.MockTransport(handler)
|
||||
|
||||
def factory(*args, **kwargs):
|
||||
kwargs.pop("transport", None)
|
||||
return real_client(transport=transport, **kwargs)
|
||||
|
||||
from types import SimpleNamespace
|
||||
stub = SimpleNamespace(
|
||||
AsyncClient=factory,
|
||||
TimeoutException=httpx.TimeoutException,
|
||||
RequestError=httpx.RequestError,
|
||||
)
|
||||
monkeypatch.setattr(webhook_outbound, "httpx", stub)
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config")
|
||||
@pytest.mark.asyncio
|
||||
async def test_deliver_to_sub_retries_on_failure():
|
||||
"""Test that deliver_to_sub retries on failure."""
|
||||
pass
|
||||
async def test_deliver_to_sub_retries_on_failure(db, monkeypatch):
|
||||
"""Two failures then success → 3 attempts, journal retrying→retrying→delivered."""
|
||||
monkeypatch.setattr(webhook_outbound, "RETRY_DELAYS", (0, 0, 0)) # no real sleeps
|
||||
sub_id = _add_sub(url="https://receiver.example/retry", event="page.created")
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.side_effect = [
|
||||
MagicMock(status_code=500),
|
||||
MagicMock(status_code=500),
|
||||
MagicMock(status_code=200),
|
||||
]
|
||||
|
||||
ok = await deliver_to_sub(
|
||||
sub_id, "https://receiver.example/retry", "page.created",
|
||||
{"page_id": 1}, "test_secret", _client=mock_client,
|
||||
)
|
||||
|
||||
assert ok is True
|
||||
assert mock_client.post.call_count == 3
|
||||
assert [r["status"] for r in _journal()] == ["retrying", "retrying", "delivered"]
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config")
|
||||
@pytest.mark.asyncio
|
||||
async def test_deliver_to_sub_fails_after_max_attempts():
|
||||
"""Test that deliver_to_sub fails after MAX_ATTEMPTS."""
|
||||
pass
|
||||
async def test_deliver_to_sub_fails_after_max_attempts(db, monkeypatch):
|
||||
"""Persistent 500 → exactly MAX_ATTEMPTS attempts, final journal row 'failed'."""
|
||||
monkeypatch.setattr(webhook_outbound, "RETRY_DELAYS", (0, 0, 0))
|
||||
sub_id = _add_sub(url="https://receiver.example/down")
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.return_value = MagicMock(status_code=500)
|
||||
|
||||
ok = await deliver_to_sub(
|
||||
sub_id, "https://receiver.example/down", "page.created",
|
||||
{}, "test_secret", _client=mock_client,
|
||||
)
|
||||
|
||||
assert ok is False
|
||||
assert mock_client.post.call_count == MAX_ATTEMPTS
|
||||
journal = _journal()
|
||||
assert len(journal) == MAX_ATTEMPTS # 3 'retrying' + 1 'failed'
|
||||
assert journal[-1]["status"] == "failed"
|
||||
assert journal[-1]["http_code"] == 500
|
||||
assert journal[-1]["attempt"] == MAX_ATTEMPTS - 1
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config")
|
||||
@pytest.mark.asyncio
|
||||
async def test_fire_event_ignores_unknown_events():
|
||||
"""Test that fire_event ignores unknown events."""
|
||||
pass
|
||||
async def test_fire_event_ignores_unknown_events(db, monkeypatch):
|
||||
"""fire_event returns before touching subscribers/HTTP for unknown events."""
|
||||
delivered = AsyncMock()
|
||||
monkeypatch.setattr(webhook_outbound, "deliver_to_sub", delivered)
|
||||
_add_sub(url="https://receiver.example/any", event="*")
|
||||
|
||||
await fire_event("nope.not_an_event", {"x": 1})
|
||||
|
||||
delivered.assert_not_awaited()
|
||||
assert _journal() == []
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config")
|
||||
@pytest.mark.asyncio
|
||||
async def test_fire_event_no_subscribers():
|
||||
"""Test that fire_event does nothing if no subscribers."""
|
||||
pass
|
||||
async def test_fire_event_no_subscribers(db, monkeypatch):
|
||||
"""Valid event but empty webhook_subscriptions → no delivery attempt."""
|
||||
delivered = AsyncMock()
|
||||
monkeypatch.setattr(webhook_outbound, "deliver_to_sub", delivered)
|
||||
|
||||
await fire_event("page.created", {"page_id": 1})
|
||||
|
||||
delivered.assert_not_awaited()
|
||||
assert _journal() == []
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config")
|
||||
@pytest.mark.asyncio
|
||||
async def test_fire_event_delivers_to_subscribers():
|
||||
"""Test that fire_event delivers to matching subscribers."""
|
||||
pass
|
||||
async def test_fire_event_delivers_to_subscribers(db, monkeypatch):
|
||||
"""Wildcard fan-out: only the subscription matching page.* receives the event."""
|
||||
matching = _add_sub(url="https://receiver.example/pages", event="page.*")
|
||||
_add_sub(url="https://receiver.example/collections", event="collection.created")
|
||||
delivered = AsyncMock()
|
||||
monkeypatch.setattr(webhook_outbound, "deliver_to_sub", delivered)
|
||||
|
||||
await fire_event("page.created", {"page_id": 42})
|
||||
|
||||
delivered.assert_awaited_once()
|
||||
args = delivered.await_args.args
|
||||
assert args[0] == matching
|
||||
assert args[1] == "https://receiver.example/pages"
|
||||
assert args[2] == "page.created"
|
||||
assert args[3] == {"page_id": 42}
|
||||
assert args[4] == "test_secret"
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config")
|
||||
@pytest.mark.asyncio
|
||||
async def test_full_webhook_flow():
|
||||
"""Test a full webhook flow: event fired → delivered to subscriber."""
|
||||
pass
|
||||
async def test_full_webhook_flow(db, monkeypatch):
|
||||
"""End-to-end: subscription → fire_event → signed HTTP POST → 'delivered' journal."""
|
||||
secret = "flow_secret_abc"
|
||||
sub_id = _add_sub(url="https://receiver.example/hook", event="page.*", secret=secret)
|
||||
|
||||
seen: dict = {}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
seen["url"] = str(request.url)
|
||||
seen["headers"] = dict(request.headers)
|
||||
seen["body"] = bytes(request.content)
|
||||
return httpx.Response(200, json={"ok": True})
|
||||
|
||||
_patch_async_client(monkeypatch, handler)
|
||||
|
||||
await fire_event("page.created", {"page_id": 7, "title": "Hello"})
|
||||
|
||||
# HTTP assertions: right URL, event header, verifiable HMAC over the raw body
|
||||
assert seen["url"] == "https://receiver.example/hook"
|
||||
assert seen["headers"]["x-flowdeck-event"] == "page.created"
|
||||
assert verify_signature(secret, seen["body"],
|
||||
seen["headers"]["x-flowdeck-signature"]) is True
|
||||
payload = json.loads(seen["body"])
|
||||
assert payload["event"] == "page.created"
|
||||
assert payload["page_id"] == 7
|
||||
assert payload["title"] == "Hello"
|
||||
|
||||
# Journal assertions
|
||||
journal = _journal()
|
||||
assert len(journal) == 1
|
||||
assert journal[0]["webhook_id"] == sub_id
|
||||
assert journal[0]["status"] == "delivered"
|
||||
assert journal[0]["http_code"] == 200
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config")
|
||||
@pytest.mark.asyncio
|
||||
async def test_retry_due_deliveries():
|
||||
"""Test retry_due_deliveries retries failed deliveries."""
|
||||
pass
|
||||
async def test_retry_due_deliveries(db, monkeypatch):
|
||||
"""A 'retrying' row past next_retry_at is re-fired, delivered, then superseded."""
|
||||
monkeypatch.setattr(webhook_outbound, "RETRY_DELAYS", (0, 0, 0))
|
||||
sub_id = _add_sub(url="https://receiver.example/retry-due", event="page.*")
|
||||
with get_conn() as conn:
|
||||
conn.execute(
|
||||
"""INSERT INTO webhook_deliveries
|
||||
(webhook_id, event, status, http_code, error, duration_ms,
|
||||
attempt, payload, next_retry_at)
|
||||
VALUES (?, 'page.created', 'retrying', 500, 'HTTP 500', 12, 0, ?, ?)""",
|
||||
(sub_id, json.dumps({"page_id": 9}), time.time() - 1),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
hits: list = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
hits.append(request)
|
||||
return httpx.Response(200)
|
||||
|
||||
_patch_async_client(monkeypatch, handler)
|
||||
|
||||
retried = await retry_due_deliveries(now=time.time())
|
||||
|
||||
assert retried == 1
|
||||
assert len(hits) == 1
|
||||
assert hits[0].headers["x-flowdeck-event"] == "page.created"
|
||||
journal = _journal()
|
||||
assert len(journal) == 2
|
||||
assert journal[0]["status"] == "superseded" # original due row
|
||||
assert journal[1]["status"] == "delivered" # new attempt
|
||||
assert journal[1]["http_code"] == 200
|
||||
|
||||
Reference in New Issue
Block a user