from __future__ import annotations import asyncio import hashlib import hmac import json import os import sys import time from typing import AsyncGenerator import pytest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from app.services.webhook_socket_service import ChannelManager, sign_payload, verify_signature API_KEY = "changeme" AUTH_HEADER = {"Authorization": f"Bearer {API_KEY}"} BASE = "http://localhost:7860/api/v1" class TestChannelManager: def test_create_channel(self): mgr = ChannelManager() ch = mgr.create_channel() assert ch.channel_id is not None assert len(ch.channel_id) == 16 assert ch.buffer_size == 0 assert ch.secret is None def test_create_channel_custom_id(self): mgr = ChannelManager() ch = mgr.create_channel(channel_id="my-channel") assert ch.channel_id == "my-channel" def test_create_channel_with_secret_and_buffer(self): mgr = ChannelManager() ch = mgr.create_channel(channel_id="secure", secret="s3cret", buffer_size=10) assert ch.secret == "s3cret" assert ch.buffer_size == 10 def test_get_channel(self): mgr = ChannelManager() mgr.create_channel(channel_id="abc") assert mgr.get_channel("abc") is not None assert mgr.get_channel("nonexistent") is None def test_delete_channel(self): mgr = ChannelManager() mgr.create_channel(channel_id="del-me") assert mgr.delete_channel("del-me") is True assert mgr.get_channel("del-me") is None assert mgr.delete_channel("del-me") is False def test_create_duplicate_id(self): mgr = ChannelManager() mgr.create_channel(channel_id="dup") ch2 = mgr.create_channel(channel_id="dup") assert mgr.get_channel("dup") is ch2 def test_default_buffer_from_manager(self): mgr = ChannelManager(default_buffer=25) ch = mgr.create_channel() assert ch.buffer_size == 25 def test_publish_nonexistent_channel(self): mgr = ChannelManager() result = asyncio.run(mgr.publish("no-such-channel", {"hello": "world"})) assert result == -1 @pytest.mark.asyncio async def test_publish_and_buffer(self): mgr = ChannelManager() ch = mgr.create_channel(channel_id="buf-test", buffer_size=3) await mgr.publish("buf-test", {"n": 1}) await mgr.publish("buf-test", {"n": 2}) await mgr.publish("buf-test", {"n": 3}) await mgr.publish("buf-test", {"n": 4}) assert len(ch.history) == 3 assert ch.history[0]["payload"]["n"] == 2 assert ch.history[2]["payload"]["n"] == 4 assert ch.message_count == 4 def test_stats(self): mgr = ChannelManager() mgr.create_channel(channel_id="a") mgr.create_channel(channel_id="b") stats = mgr.stats() assert stats["channels"] == 2 assert stats["total_subscribers"] == 0 def test_sign_and_verify(self): secret = "my-secret" body = b'{"hello":"world"}' sig = sign_payload(secret, body) assert sig.startswith("sha256=") assert verify_signature(secret, body, sig) is True assert verify_signature(secret, body, "sha256=bad") is False assert verify_signature("wrong-secret", body, sig) is False def test_sign_constant_result(self): secret = "test" body = b"data" sig1 = sign_payload(secret, body) sig2 = sign_payload(secret, body) assert sig1 == sig2 def test_channel_info_fields(self): mgr = ChannelManager() mgr.create_channel(channel_id="info-test", buffer_size=5) ch = mgr.get_channel("info-test") assert ch.channel_id == "info-test" assert ch.buffer_size == 5 assert ch.message_count == 0 assert ch.created_at > 0 assert ch.last_activity > 0 def test_publish_with_subscriber(self): mgr = ChannelManager() mgr.create_channel(channel_id="sub-test") ch = mgr.get_channel("sub-test") async def dummy(): return ch.channel_id q = asyncio.Queue() ch.subscribers[dummy] = q result = asyncio.run(mgr.publish("sub-test", {"msg": "hello"})) assert result == 1 assert ch.message_count == 1 def test_multiple_publishes(self): mgr = ChannelManager() ch = mgr.create_channel(channel_id="multi-pub", buffer_size=10) for i in range(5): asyncio.run(mgr.publish("multi-pub", {"n": i})) assert ch.message_count == 5 assert len(ch.history) == 5 def test_channel_manager_defaults(self): mgr = ChannelManager() assert mgr.default_buffer == 0 assert mgr.channels == {} pytestmark_integration = pytest.mark.skipif( not os.environ.get("RUN_INTEGRATION_TESTS"), reason="Set RUN_INTEGRATION_TESTS=1 to run integration tests", ) def _sign(body: bytes, secret: str) -> str: return "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() @pytest.mark.skipif( not os.environ.get("RUN_INTEGRATION_TESTS"), reason="Set RUN_INTEGRATION_TESTS=1 to run integration tests", ) class TestWebhookSocketIntegration: @pytest.fixture(autouse=True) async def _setup(self): import httpx async with httpx.AsyncClient(base_url=BASE) as client: self.client = client yield async def _create_channel(self, **kwargs) -> dict: resp = await self.client.post("/channels", json=kwargs, headers=AUTH_HEADER) assert resp.status_code == 201 return resp.json() async def _delete_channel(self, channel_id: str): resp = await self.client.delete(f"/channels/{channel_id}", headers=AUTH_HEADER) return resp.status_code == 200 async def test_health(self): resp = await self.client.get("/health") assert resp.status_code == 200 data = resp.json() assert data["success"] is True async def test_create_and_list_channels(self): ch = await self._create_channel(channel_id="test-list", buffer_size=5) assert ch["channel_id"] == "test-list" resp = await self.client.get("/channels", headers=AUTH_HEADER) assert resp.status_code == 200 data = resp.json() assert any(c["channel_id"] == "test-list" for c in data["channels"]) await self._delete_channel("test-list") async def test_create_channel_no_auth(self): resp = await self.client.post("/channels", json={}) assert resp.status_code in (401, 403) async def test_create_duplicate_channel(self): await self._create_channel(channel_id="dup-test") resp = await self.client.post("/channels", json={"channel_id": "dup-test"}, headers=AUTH_HEADER) assert resp.status_code == 409 await self._delete_channel("dup-test") async def test_channel_info(self): await self._create_channel(channel_id="info-test") resp = await self.client.get("/channels/info-test", headers=AUTH_HEADER) assert resp.status_code == 200 data = resp.json() assert data["channel_id"] == "info-test" assert data["subscribers"] == 0 assert data["messages"] == 0 await self._delete_channel("info-test") async def test_channel_info_not_found(self): resp = await self.client.get("/channels/no-such", headers=AUTH_HEADER) assert resp.status_code == 404 async def test_delete_channel(self): await self._create_channel(channel_id="delete-me") resp = await self.client.delete("/channels/delete-me", headers=AUTH_HEADER) assert resp.status_code == 200 assert resp.json()["deleted"] == "delete-me" resp = await self.client.get("/channels/delete-me", headers=AUTH_HEADER) assert resp.status_code == 404 async def test_delete_channel_not_found(self): resp = await self.client.delete("/channels/no-such", headers=AUTH_HEADER) assert resp.status_code == 404 async def test_webhook_delivers_to_ws(self): ch = await self._create_channel(channel_id="ws-deliver", buffer_size=5) cid = ch["channel_id"] import websockets ws_url = f"ws://localhost:7860/api/v1/ws/{cid}" async with websockets.connect(ws_url) as ws: connected = json.loads(await ws.recv()) assert connected["event"] == "connected" assert connected["channel"] == cid payload = {"msg": "hello from webhook", "num": 42} resp = await self.client.post(f"/webhook/{cid}", json=payload) assert resp.status_code == 200 wh_data = resp.json() assert wh_data["status"] == "delivered" assert wh_data["subscribers_notified"] == 1 received = json.loads(await ws.recv()) assert received["event"] == "message" assert received["channel"] == cid assert received["payload"] == payload await self._delete_channel(cid) async def test_webhook_no_subscribers(self): ch = await self._create_channel(channel_id="no-subs") resp = await self.client.post(f"/webhook/{ch['channel_id']}", json={"data": 1}) assert resp.status_code == 200 assert resp.json()["subscribers_notified"] == 0 await self._delete_channel(ch["channel_id"]) async def test_webhook_not_found(self): resp = await self.client.post("/webhook/no-such", json={"x": 1}) assert resp.status_code == 404 async def test_hook_alias(self): ch = await self._create_channel(channel_id="hook-alias") resp = await self.client.post(f"/hook/{ch['channel_id']}", json={"test": True}) assert resp.status_code == 200 assert resp.json()["status"] == "delivered" await self._delete_channel(ch["channel_id"]) async def test_hmac_signed_webhook(self): secret = "hmac-test-secret" ch = await self._create_channel(channel_id="hmac-test", secret=secret) cid = ch["channel_id"] payload = b'{"signed": "data"}' sig = _sign(payload, secret) resp = await self.client.post( f"/webhook/{cid}", content=payload, headers={"Content-Type": "application/json", "X-Signature-256": sig}, ) assert resp.status_code == 200, resp.text resp_bad = await self.client.post( f"/webhook/{cid}", content=payload, headers={"Content-Type": "application/json", "X-Signature-256": "sha256=bad"}, ) assert resp_bad.status_code == 401 resp_no_sig = await self.client.post( f"/webhook/{cid}", content=payload, headers={"Content-Type": "application/json"}, ) assert resp_no_sig.status_code == 401 await self._delete_channel(cid) async def test_webhook_form_urlencoded(self): ch = await self._create_channel(channel_id="form-test") cid = ch["channel_id"] resp = await self.client.post( f"/webhook/{cid}", data={"field1": "value1", "field2": "value2"}, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) assert resp.status_code == 200 assert resp.json()["status"] == "delivered" await self._delete_channel(cid) async def test_webhook_raw_text(self): ch = await self._create_channel(channel_id="raw-test") cid = ch["channel_id"] resp = await self.client.post( f"/webhook/{cid}", content="just some raw text", headers={"Content-Type": "text/plain"}, ) assert resp.status_code == 200 await self._delete_channel(cid) async def test_ws_replay_buffer(self): ch = await self._create_channel(channel_id="replay-test", buffer_size=3) cid = ch["channel_id"] await self.client.post(f"/webhook/{cid}", json={"n": 1}) await self.client.post(f"/webhook/{cid}", json={"n": 2}) await self.client.post(f"/webhook/{cid}", json={"n": 3}) import websockets ws_url = f"ws://localhost:7860/api/v1/ws/{cid}" async with websockets.connect(ws_url) as ws: connected = json.loads(await ws.recv()) assert connected["event"] == "connected" assert connected["buffered"] == 3 for expected_n in [1, 2, 3]: msg = json.loads(await ws.recv()) assert msg["payload"]["n"] == expected_n await self._delete_channel(cid) async def test_ws_auth_with_secret(self): ch = await self._create_channel(channel_id="ws-auth-test", secret="topsecret") cid = ch["channel_id"] import websockets ws_url = f"ws://localhost:7860/api/v1/ws/{cid}" async with websockets.connect(f"{ws_url}?secret=topsecret") as ws: msg = json.loads(await ws.recv()) assert msg["event"] == "connected" async with websockets.connect(ws_url) as ws: msg = json.loads(await ws.recv()) assert msg["event"] == "error" await self._delete_channel(cid) async def test_ws_channel_not_found(self): import websockets async with websockets.connect("ws://localhost:7860/api/v1/ws/does-not-exist") as ws: msg = json.loads(await ws.recv()) assert msg["event"] == "error" assert "channel not found" in msg["message"] async def test_stats_endpoint(self): resp = await self.client.get("/webhook-socket/stats", headers=AUTH_HEADER) assert resp.status_code == 200 data = resp.json() assert "channels" in data assert "total_messages" in data assert "total_subscribers" in data async def test_full_lifecycle(self): cid = "lifecycle-test" ch = await self._create_channel(channel_id=cid, buffer_size=10, secret="life-secret") assert ch["channel_id"] == cid info_resp = await self.client.get(f"/channels/{cid}", headers=AUTH_HEADER) assert info_resp.status_code == 200 assert info_resp.json()["messages"] == 0 import websockets ws_url = f"ws://localhost:7860/api/v1/ws/{cid}?secret=life-secret" async with websockets.connect(ws_url) as ws: connected = json.loads(await ws.recv()) assert connected["event"] == "connected" payload = {"event_type": "push", "data": {"ref": "main"}} sig = _sign(json.dumps(payload).encode(), "life-secret") resp = await self.client.post( f"/webhook/{cid}", json=payload, headers={"X-Signature-256": sig, "X-GitHub-Event": "push"}, ) assert resp.status_code == 200 wh = resp.json() assert wh["subscribers_notified"] == 1 received = json.loads(await ws.recv()) assert received["event"] == "message" assert received["payload"]["event_type"] == "push" assert received["headers"]["X-GitHub-Event"] == "push" stats_resp = await self.client.get(f"/channels/{cid}", headers=AUTH_HEADER) assert stats_resp.json()["messages"] == 1 await self._delete_channel(cid) not_found = await self.client.get(f"/channels/{cid}", headers=AUTH_HEADER) assert not_found.status_code == 404 if __name__ == "__main__": import subprocess import sys as _sys os.environ["RUN_INTEGRATION_TESTS"] = "1" _sys.exit( subprocess.run( [_sys.executable, "-m", "pytest", __file__, "-v", "--tb=short"], cwd=os.path.join(os.path.dirname(__file__), ".."), ).returncode )