tests/test_webhook_socket.py

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
    )