Add webhooks & reformat a bit

tests/test_fanout.py

@@ -56,6 +56,26 @@ class TestScopeMatchesMessage:
         scope = {"messages": {"contacts": ["pk1"]}}
         assert not _scope_matches_message(scope, {"type": "PRIV", "conversation_key": "pk2"})
 
+    def test_dict_channels_except_excludes_listed(self):
+        scope = {"messages": {"channels": {"except": ["ch1"]}, "contacts": "all"}}
+        assert not _scope_matches_message(scope, {"type": "CHAN", "conversation_key": "ch1"})
+
+    def test_dict_channels_except_includes_unlisted(self):
+        scope = {"messages": {"channels": {"except": ["ch1"]}, "contacts": "all"}}
+        assert _scope_matches_message(scope, {"type": "CHAN", "conversation_key": "ch2"})
+
+    def test_dict_contacts_except_excludes_listed(self):
+        scope = {"messages": {"channels": "all", "contacts": {"except": ["pk1"]}}}
+        assert not _scope_matches_message(scope, {"type": "PRIV", "conversation_key": "pk1"})
+
+    def test_dict_contacts_except_includes_unlisted(self):
+        scope = {"messages": {"channels": "all", "contacts": {"except": ["pk1"]}}}
+        assert _scope_matches_message(scope, {"type": "PRIV", "conversation_key": "pk2"})
+
+    def test_dict_channels_except_empty_matches_all(self):
+        scope = {"messages": {"channels": {"except": []}}}
+        assert _scope_matches_message(scope, {"type": "CHAN", "conversation_key": "ch1"})
+
 
 class TestScopeMatchesRaw:
     def test_all_matches(self):
@@ -626,3 +646,106 @@ class TestMigration037:
             assert row[0] == 0
         finally:
             await db.disconnect()
+
+
+# ---------------------------------------------------------------------------
+# Webhook module unit tests
+# ---------------------------------------------------------------------------
+
+
+class TestWebhookModule:
+    @pytest.mark.asyncio
+    async def test_status_disconnected_when_no_url(self):
+        from app.fanout.webhook import WebhookModule
+
+        mod = WebhookModule("test", {"url": ""})
+        await mod.start()
+        assert mod.status == "disconnected"
+        await mod.stop()
+
+    @pytest.mark.asyncio
+    async def test_status_connected_with_url(self):
+        from app.fanout.webhook import WebhookModule
+
+        mod = WebhookModule("test", {"url": "http://localhost:9999/hook"})
+        await mod.start()
+        assert mod.status == "connected"
+        await mod.stop()
+
+    @pytest.mark.asyncio
+    async def test_dispatch_with_matching_scope(self):
+        """WebhookModule dispatches through FanoutManager scope matching."""
+        manager = FanoutManager()
+        mod = StubModule()
+        scope = {"messages": {"channels": ["ch1"], "contacts": "none"}, "raw_packets": "none"}
+        manager._modules["test-webhook"] = (mod, scope)
+
+        await manager.broadcast_message({"type": "CHAN", "conversation_key": "ch1", "text": "yes"})
+        await manager.broadcast_message({"type": "CHAN", "conversation_key": "ch2", "text": "no"})
+        await manager.broadcast_message(
+            {"type": "PRIV", "conversation_key": "pk1", "text": "dm no"}
+        )
+
+        assert len(mod.message_calls) == 1
+        assert mod.message_calls[0]["text"] == "yes"
+
+
+# ---------------------------------------------------------------------------
+# Webhook router validation tests
+# ---------------------------------------------------------------------------
+
+
+class TestWebhookValidation:
+    def test_validate_webhook_config_requires_url(self):
+        from fastapi import HTTPException
+
+        from app.routers.fanout import _validate_webhook_config
+
+        with pytest.raises(HTTPException) as exc_info:
+            _validate_webhook_config({"url": ""})
+        assert exc_info.value.status_code == 400
+        assert "url is required" in exc_info.value.detail
+
+    def test_validate_webhook_config_requires_http_scheme(self):
+        from fastapi import HTTPException
+
+        from app.routers.fanout import _validate_webhook_config
+
+        with pytest.raises(HTTPException) as exc_info:
+            _validate_webhook_config({"url": "ftp://example.com"})
+        assert exc_info.value.status_code == 400
+
+    def test_validate_webhook_config_rejects_bad_method(self):
+        from fastapi import HTTPException
+
+        from app.routers.fanout import _validate_webhook_config
+
+        with pytest.raises(HTTPException) as exc_info:
+            _validate_webhook_config({"url": "https://example.com/hook", "method": "DELETE"})
+        assert exc_info.value.status_code == 400
+        assert "method" in exc_info.value.detail
+
+    def test_validate_webhook_config_accepts_valid(self):
+        from app.routers.fanout import _validate_webhook_config
+
+        # Should not raise
+        _validate_webhook_config(
+            {"url": "https://example.com/hook", "method": "POST", "headers": {}}
+        )
+
+    def test_enforce_scope_webhook_strips_raw_packets(self):
+        from app.routers.fanout import _enforce_scope
+
+        scope = _enforce_scope("webhook", {"messages": "all", "raw_packets": "all"})
+        assert scope["raw_packets"] == "none"
+        assert scope["messages"] == "all"
+
+    def test_enforce_scope_webhook_preserves_selective(self):
+        from app.routers.fanout import _enforce_scope
+
+        scope = _enforce_scope(
+            "webhook",
+            {"messages": {"channels": ["ch1"], "contacts": "none"}, "raw_packets": "all"},
+        )
+        assert scope["raw_packets"] == "none"
+        assert scope["messages"] == {"channels": ["ch1"], "contacts": "none"}

tests/test_fanout_integration.py

@@ -401,3 +401,437 @@ class TestFanoutMqttIntegration:
 
         assert len(mqtt_broker.published) == 1
         assert "raw/" in mqtt_broker.published[0][0]
+
+
+# ---------------------------------------------------------------------------
+# Webhook capture HTTP server
+# ---------------------------------------------------------------------------
+
+
+class WebhookCaptureServer:
+    """Tiny HTTP server that captures POST requests for webhook testing."""
+
+    def __init__(self):
+        self.received: list[dict] = []
+        self._server: asyncio.Server | None = None
+        self.port: int = 0
+
+    async def start(self) -> int:
+        self._server = await asyncio.start_server(self._handle, "127.0.0.1", 0)
+        self.port = self._server.sockets[0].getsockname()[1]
+        return self.port
+
+    async def stop(self):
+        if self._server:
+            self._server.close()
+            await self._server.wait_closed()
+
+    async def wait_for(self, count: int, timeout: float = 5.0) -> list[dict]:
+        deadline = asyncio.get_event_loop().time() + timeout
+        while len(self.received) < count:
+            if asyncio.get_event_loop().time() >= deadline:
+                break
+            await asyncio.sleep(0.02)
+        return list(self.received)
+
+    async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
+        try:
+            # Read HTTP request line
+            request_line = await reader.readline()
+            if not request_line:
+                return
+
+            # Read headers
+            headers: dict[str, str] = {}
+            while True:
+                line = await reader.readline()
+                if line in (b"\r\n", b"\n", b""):
+                    break
+                decoded = line.decode("utf-8", errors="replace").strip()
+                if ":" in decoded:
+                    key, val = decoded.split(":", 1)
+                    headers[key.strip().lower()] = val.strip()
+
+            # Read body
+            content_length = int(headers.get("content-length", "0"))
+            body = b""
+            if content_length > 0:
+                body = await reader.readexactly(content_length)
+
+            payload: dict = {}
+            if body:
+                try:
+                    payload = json.loads(body)
+                except Exception:
+                    payload = {"_raw": body.decode("utf-8", errors="replace")}
+
+            self.received.append(
+                {
+                    "method": request_line.decode().split()[0],
+                    "headers": headers,
+                    "body": payload,
+                }
+            )
+
+            # Send 200 OK
+            response = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"
+            writer.write(response)
+            await writer.drain()
+        except (asyncio.IncompleteReadError, ConnectionError, OSError):
+            pass
+        finally:
+            writer.close()
+
+
+@pytest.fixture
+async def webhook_server():
+    server = WebhookCaptureServer()
+    await server.start()
+    yield server
+    await server.stop()
+
+
+def _webhook_config(port: int, secret: str = "") -> dict:
+    return {
+        "url": f"http://127.0.0.1:{port}/hook",
+        "method": "POST",
+        "headers": {},
+        "secret": secret,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Webhook integration tests
+# ---------------------------------------------------------------------------
+
+
+class TestFanoutWebhookIntegration:
+    """End-to-end: real HTTP capture server <-> real WebhookModule."""
+
+    @pytest.mark.asyncio
+    async def test_webhook_receives_message(self, webhook_server, integration_db):
+        """An enabled webhook receives message data via HTTP POST."""
+        cfg = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Test Hook",
+            config=_webhook_config(webhook_server.port),
+            scope={"messages": "all", "raw_packets": "none"},
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg["id"])
+
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk1", "text": "hello webhook"}
+            )
+
+            results = await webhook_server.wait_for(1)
+        finally:
+            await manager.stop_all()
+
+        assert len(results) == 1
+        assert results[0]["body"]["text"] == "hello webhook"
+        assert results[0]["body"]["conversation_key"] == "pk1"
+        assert results[0]["headers"].get("x-webhook-event") == "message"
+
+    @pytest.mark.asyncio
+    async def test_webhook_sends_secret_header(self, webhook_server, integration_db):
+        """Webhook sends X-Webhook-Secret when configured."""
+        cfg = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Secret Hook",
+            config=_webhook_config(webhook_server.port, secret="my-secret-123"),
+            scope={"messages": "all", "raw_packets": "none"},
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg["id"])
+
+            await manager.broadcast_message(
+                {"type": "CHAN", "conversation_key": "ch1", "text": "secret test"}
+            )
+
+            results = await webhook_server.wait_for(1)
+        finally:
+            await manager.stop_all()
+
+        assert len(results) == 1
+        assert results[0]["headers"].get("x-webhook-secret") == "my-secret-123"
+
+    @pytest.mark.asyncio
+    async def test_webhook_disabled_no_delivery(self, webhook_server, integration_db):
+        """Disabled webhook should not deliver any messages."""
+        await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Disabled Hook",
+            config=_webhook_config(webhook_server.port),
+            scope={"messages": "all", "raw_packets": "none"},
+            enabled=False,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            assert len(manager._modules) == 0
+
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk1", "text": "nope"}
+            )
+            await asyncio.sleep(0.3)
+        finally:
+            await manager.stop_all()
+
+        assert len(webhook_server.received) == 0
+
+    @pytest.mark.asyncio
+    async def test_webhook_scope_selective_channels(self, webhook_server, integration_db):
+        """Webhook with selective scope only fires for matching channels."""
+        cfg = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Selective Hook",
+            config=_webhook_config(webhook_server.port),
+            scope={"messages": {"channels": ["ch-yes"], "contacts": "none"}, "raw_packets": "none"},
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg["id"])
+
+            # Matching channel — should deliver
+            await manager.broadcast_message(
+                {"type": "CHAN", "conversation_key": "ch-yes", "text": "included"}
+            )
+            # Non-matching channel — should NOT deliver
+            await manager.broadcast_message(
+                {"type": "CHAN", "conversation_key": "ch-no", "text": "excluded"}
+            )
+            # DM — contacts is "none", should NOT deliver
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk1", "text": "dm excluded"}
+            )
+
+            await webhook_server.wait_for(1)
+            await asyncio.sleep(0.3)  # wait for any stragglers
+        finally:
+            await manager.stop_all()
+
+        assert len(webhook_server.received) == 1
+        assert webhook_server.received[0]["body"]["text"] == "included"
+
+    @pytest.mark.asyncio
+    async def test_webhook_scope_selective_contacts(self, webhook_server, integration_db):
+        """Webhook with selective scope only fires for matching contacts."""
+        cfg = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Contact Hook",
+            config=_webhook_config(webhook_server.port),
+            scope={
+                "messages": {"channels": "none", "contacts": ["pk-yes"]},
+                "raw_packets": "none",
+            },
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg["id"])
+
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk-yes", "text": "dm included"}
+            )
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk-no", "text": "dm excluded"}
+            )
+
+            await webhook_server.wait_for(1)
+            await asyncio.sleep(0.3)
+        finally:
+            await manager.stop_all()
+
+        assert len(webhook_server.received) == 1
+        assert webhook_server.received[0]["body"]["text"] == "dm included"
+
+    @pytest.mark.asyncio
+    async def test_webhook_scope_all_receives_everything(self, webhook_server, integration_db):
+        """Webhook with scope messages='all' receives DMs and channel messages."""
+        cfg = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="All Hook",
+            config=_webhook_config(webhook_server.port),
+            scope={"messages": "all", "raw_packets": "none"},
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg["id"])
+
+            await manager.broadcast_message(
+                {"type": "CHAN", "conversation_key": "ch1", "text": "channel msg"}
+            )
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk1", "text": "dm msg"}
+            )
+
+            results = await webhook_server.wait_for(2)
+        finally:
+            await manager.stop_all()
+
+        assert len(results) == 2
+        texts = {r["body"]["text"] for r in results}
+        assert "channel msg" in texts
+        assert "dm msg" in texts
+
+    @pytest.mark.asyncio
+    async def test_webhook_scope_none_receives_nothing(self, webhook_server, integration_db):
+        """Webhook with scope messages='none' receives nothing."""
+        cfg = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="None Hook",
+            config=_webhook_config(webhook_server.port),
+            scope={"messages": "none", "raw_packets": "none"},
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg["id"])
+
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk1", "text": "should not arrive"}
+            )
+            await asyncio.sleep(0.3)
+        finally:
+            await manager.stop_all()
+
+        assert len(webhook_server.received) == 0
+
+    @pytest.mark.asyncio
+    async def test_two_webhooks_both_receive(self, webhook_server, integration_db):
+        """Two enabled webhooks both receive the same message."""
+        cfg_a = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Hook A",
+            config=_webhook_config(webhook_server.port, secret="a"),
+            scope={"messages": "all", "raw_packets": "none"},
+            enabled=True,
+        )
+        cfg_b = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Hook B",
+            config=_webhook_config(webhook_server.port, secret="b"),
+            scope={"messages": "all", "raw_packets": "none"},
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg_a["id"])
+            await _wait_connected(manager, cfg_b["id"])
+
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk1", "text": "multi"}
+            )
+
+            results = await webhook_server.wait_for(2)
+        finally:
+            await manager.stop_all()
+
+        assert len(results) == 2
+        secrets = {r["headers"].get("x-webhook-secret") for r in results}
+        assert "a" in secrets
+        assert "b" in secrets
+
+    @pytest.mark.asyncio
+    async def test_webhook_disable_stops_delivery(self, webhook_server, integration_db):
+        """Disabling a webhook stops delivery immediately."""
+        cfg = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Toggle Hook",
+            config=_webhook_config(webhook_server.port),
+            scope={"messages": "all", "raw_packets": "none"},
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg["id"])
+
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk1", "text": "before disable"}
+            )
+            await webhook_server.wait_for(1)
+            assert len(webhook_server.received) == 1
+
+            # Disable
+            await FanoutConfigRepository.update(cfg["id"], enabled=False)
+            await manager.reload_config(cfg["id"])
+            assert cfg["id"] not in manager._modules
+
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk2", "text": "after disable"}
+            )
+            await asyncio.sleep(0.3)
+        finally:
+            await manager.stop_all()
+
+        assert len(webhook_server.received) == 1
+
+    @pytest.mark.asyncio
+    async def test_webhook_scope_except_channels(self, webhook_server, integration_db):
+        """Webhook with except-mode excludes listed channels, includes others."""
+        cfg = await FanoutConfigRepository.create(
+            config_type="webhook",
+            name="Except Hook",
+            config=_webhook_config(webhook_server.port),
+            scope={
+                "messages": {
+                    "channels": {"except": ["ch-excluded"]},
+                    "contacts": {"except": []},
+                },
+                "raw_packets": "none",
+            },
+            enabled=True,
+        )
+
+        manager = FanoutManager()
+        try:
+            await manager.load_from_db()
+            await _wait_connected(manager, cfg["id"])
+
+            # Excluded channel — should NOT deliver
+            await manager.broadcast_message(
+                {"type": "CHAN", "conversation_key": "ch-excluded", "text": "nope"}
+            )
+            # Non-excluded channel — should deliver
+            await manager.broadcast_message(
+                {"type": "CHAN", "conversation_key": "ch-other", "text": "yes"}
+            )
+            # DM with empty except list — should deliver
+            await manager.broadcast_message(
+                {"type": "PRIV", "conversation_key": "pk1", "text": "dm yes"}
+            )
+
+            await webhook_server.wait_for(2)
+            await asyncio.sleep(0.3)
+        finally:
+            await manager.stop_all()
+
+        assert len(webhook_server.received) == 2
+        texts = {r["body"]["text"] for r in webhook_server.received}
+        assert "yes" in texts
+        assert "dm yes" in texts
+        assert "nope" not in texts