flowdeck/tests/test_webhooks_v2.py

"""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,
    deliver_to_sub,
    fire_event,
    retry_due_deliveries,
    sign_payload,
    verify_signature,
)

# ── Fixtures ────────────────────────────────────────────────────────────────

@pytest.fixture
def db():
    """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
def sample_payload():
    return {"page_id": 123, "title": "Test Page", "user_id": 456}


@pytest.fixture
def sample_secret():
    return "test_webhook_secret_12345"


# ── HMAC Signature Tests ────────────────────────────────────────────────────


def test_sign_payload_returns_sha256_hex():
    """Test that sign_payload returns a valid HMAC-SHA256 signature."""
    secret = "my_secret"
    body = b'{"event": "page.created", "data": {}}'
    signature = sign_payload(secret, body)

    assert signature.startswith("sha256=")
    hex_part = signature[7:]  # Remove 'sha256=' prefix
    assert len(hex_part) == 64  # SHA-256 hex is 64 chars


def test_sign_payload_consistency():
    """Test that the same input produces the same signature."""
    secret = "my_secret"
    body = b'{"event": "page.created"}'

    sig1 = sign_payload(secret, body)
    sig2 = sign_payload(secret, body)

    assert sig1 == sig2


def test_verify_signature_valid():
    """Test that verify_signature returns True for valid signatures."""
    secret = "my_secret"
    body = b'{"event": "page.created", "data": {}}'
    signature = sign_payload(secret, body)

    assert verify_signature(secret, body, signature) is True


def test_verify_signature_invalid_secret():
    """Test that verify_signature returns False for wrong secret."""
    secret = "my_secret"
    wrong_secret = "wrong_secret"
    body = b'{"event": "page.created"}'
    signature = sign_payload(secret, body)

    assert verify_signature(wrong_secret, body, signature) is False


def test_verify_signature_invalid_signature():
    """Test that verify_signature returns False for tampered signature."""
    secret = "my_secret"
    body = b'{"event": "page.created"}'
    signature = "sha256=invalid_hex_value"

    assert verify_signature(secret, body, signature) is False


def test_verify_signature_empty_secret():
    """Test that verify_signature returns False for empty secret."""
    body = b'{"event": "page.created"}'
    signature = "sha256=somehash"

    assert verify_signature("", body, signature) is False


def test_verify_signature_empty_signature():
    """Test that verify_signature returns False for empty signature."""
    secret = "my_secret"
    body = b'{"event": "page.created"}'

    assert verify_signature(secret, body, "") is False


def test_verify_signature_none_values():
    """Test that verify_signature handles None values."""
    assert verify_signature(None, b'{}', "sha256=abc") is False
    assert verify_signature("secret", None, "sha256=abc") is False
    assert verify_signature("secret", b'{}', None) is False


# ── Event Matching Tests ────────────────────────────────────────────────────


def test_event_matches_exact():
    """Test exact event matching."""
    assert _event_matches("page.created", "page.created") is True
    assert _event_matches("page.created", "page.updated") is False


def test_event_matches_wildcard_all():
    """Test wildcard '*' matches all events."""
    assert _event_matches("*", "page.created") is True
    assert _event_matches("*", "collection.deleted") is True
    assert _event_matches("all", "agent.run.finished") is True


def test_event_matches_wildcard_prefix():
    """Test wildcard prefix matching (e.g., 'page.*')."""
    assert _event_matches("page.*", "page.created") is True
    assert _event_matches("page.*", "page.updated") is True
    assert _event_matches("page.*", "page.deleted") is True
    assert _event_matches("page.*", "collection.created") is False


def test_event_matches_wildcard_suffix():
    """Test that suffix wildcards are not supported (only prefix)."""
    assert _event_matches("*.created", "page.created") is False
    assert _event_matches("*.created", "collection.created") is False


# ── Event Catalogue Tests ────────────────────────────────────────────────────


def test_events_list_has_50_plus_events():
    """Test that EVENTS list has at least 50 events."""
    assert len(EVENTS) >= 50, f"Expected at least 50 events, got {len(EVENTS)}"


def test_events_list_contains_new_events():
    """Test that new events are included in the catalogue."""
    new_events = [
        "page.renamed", "page.duplicated", "page.archived",
        "comment.updated", "comment.deleted",
        "mention.resolved",
        "collection.renamed", "collection.duplicated",
        "collection.page.moved",
        "collection.view.updated", "collection.view.deleted",
        "collection.property.created", "collection.property.updated", "collection.property.deleted",
        "sprint.deleted", "sprint.started", "sprint.completed", "sprint.canceled",
        "share.created", "share.updated", "share.revoked",
        "automation.failed", "automation.retrying",
        "agent.run.started", "agent.run.failed",
        "workspace.created", "workspace.updated", "workspace.deleted",
        "workspace.member.added", "workspace.member.removed", "workspace.member.role_changed",
        "file.uploaded", "file.deleted",
        "import.started", "import.completed", "import.failed",
    ]
    for event in new_events:
        assert event in EVENTS, f"Event '{event}' not found in EVENTS list"


def test_events_list_contains_legacy_events():
    """Test that legacy events are still present."""
    legacy_events = [
        "page.created", "page.updated", "page.deleted", "page.moved",
        "comment.added", "comment.resolved",
        "collection.created", "collection.updated", "collection.deleted",
        "sprint.created", "sprint.updated",
        "favorite.added", "favorite.removed",
        "automation.fired", "agent.run.finished",
        "ping",
    ]
    for event in legacy_events:
        assert event in EVENTS, f"Legacy event '{event}' missing from EVENTS list"


# ── Retry Schedule Tests ────────────────────────────────────────────────────


def test_retry_delays_are_2_10_60():
    """Test that retry delays are 2s, 10s, 60s."""
    assert RETRY_DELAYS == (2, 10, 60)


def test_max_attempts_is_4():
    """Test that MAX_ATTEMPTS is 4 (1 initial + 3 retries)."""
    assert MAX_ATTEMPTS == 4


# ── Delivery Tests (Mocked) ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_deliver_once_success():
    """Test successful delivery in _deliver_once."""
    mock_client = AsyncMock()
    mock_response = MagicMock()
    mock_response.status_code = 200
    mock_client.post.return_value = mock_response

    url = "https://example.com/webhook"
    event = "page.created"
    payload = {"page_id": 123}
    secret = "test_secret"

    code, error = await _deliver_once(mock_client, url, event, payload, secret)

    assert code == 200
    assert error == ""
    mock_client.post.assert_called_once()
    call_args = mock_client.post.call_args
    assert call_args.kwargs["headers"]["X-FlowDeck-Event"] == event
    assert call_args.kwargs["headers"]["X-FlowDeck-Signature"].startswith("sha256=")


@pytest.mark.asyncio
async def test_deliver_once_http_error():
    """Test _deliver_once handles HTTP errors."""
    mock_client = AsyncMock()
    mock_response = MagicMock()
    mock_response.status_code = 500
    mock_client.post.return_value = mock_response

    url = "https://example.com/webhook"
    event = "page.created"
    payload = {}
    secret = "test_secret"

    code, error = await _deliver_once(mock_client, url, event, payload, secret)

    assert code == 500
    assert "HTTP 500" in error


@pytest.mark.asyncio
async def test_deliver_once_timeout():
    """Test _deliver_once handles timeout errors."""
    mock_client = AsyncMock()
    mock_client.post.side_effect = httpx.TimeoutException("Timeout after 30s")

    url = "https://example.com/webhook"
    event = "page.created"
    payload = {}
    secret = "test_secret"

    code, error = await _deliver_once(mock_client, url, event, payload, secret)

    assert code is None
    assert "Timeout" in error


@pytest.mark.asyncio
async def test_deliver_once_request_error():
    """Test _deliver_once handles request errors."""
    mock_client = AsyncMock()
    mock_client.post.side_effect = httpx.RequestError("Connection refused")

    url = "https://example.com/webhook"
    event = "page.created"
    payload = {}
    secret = "test_secret"

    code, error = await _deliver_once(mock_client, url, event, payload, secret)

    assert code is None
    assert "Request error" in error


# ── 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)
    # A42 : la fabrique partagée crée les clients — elle doit voir le stub
    # (boucle neuve par test → aucun cache à purger, on purge par sécurité).
    from app.services import http_client as _hc

    monkeypatch.setattr(_hc, "httpx", stub)
    _hc._clients.clear()


@pytest.mark.asyncio
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.asyncio
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.asyncio
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.asyncio
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.asyncio
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.asyncio
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.asyncio
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