from __future__ import annotations

import hashlib
import hmac
from types import SimpleNamespace
import unittest
from unittest.mock import patch

from services.webhooks.delivery import (
    canonical_json,
    should_retry,
    webhook_signature,
)
from services.webhooks.events import analysis_completed_payload
from services.webhooks.security import UnsafeWebhookUrl, validate_webhook_url


class WebhookSecurityTest(unittest.IsolatedAsyncioTestCase):
    async def test_requires_https(self):
        with self.assertRaises(UnsafeWebhookUrl):
            await validate_webhook_url("http://hooks.example.com/events")

    async def test_rejects_loopback_ip(self):
        with self.assertRaises(UnsafeWebhookUrl):
            await validate_webhook_url("https://127.0.0.1/events")

    async def test_rejects_hostname_resolving_to_private_ip(self):
        records = [(2, 1, 6, "", ("10.0.0.8", 443))]
        with patch("services.webhooks.security.socket.getaddrinfo", return_value=records):
            with self.assertRaises(UnsafeWebhookUrl):
                await validate_webhook_url("https://hooks.example.com/events")

    async def test_accepts_hostname_resolving_to_public_ip(self):
        records = [(2, 1, 6, "", ("8.8.8.8", 443))]
        with patch("services.webhooks.security.socket.getaddrinfo", return_value=records):
            result = await validate_webhook_url("https://hooks.example.com/events")
        self.assertEqual(result, "https://hooks.example.com/events")


class WebhookDeliveryTest(unittest.TestCase):
    def test_signature_uses_timestamp_dot_raw_body(self):
        body = canonical_json({"b": 2, "a": "中文"})
        expected = hmac.new(
            b"secret",
            b"1700000000." + body,
            hashlib.sha256,
        ).hexdigest()
        self.assertEqual(
            webhook_signature("secret", "1700000000", body),
            f"v1={expected}",
        )

    def test_only_transient_responses_are_retried(self):
        for status in (None, 408, 429, 500, 503):
            self.assertTrue(should_retry(status))
        for status in (200, 301, 400, 401, 404):
            self.assertFalse(should_retry(status))

    def test_analysis_event_has_stable_envelope(self):
        conversation = SimpleNamespace(
            id="conv_001",
            assistant_id="asst_001",
            assistant_name="销售助手",
        )
        from datetime import UTC, datetime

        payload = analysis_completed_payload(
            conversation,  # type: ignore[arg-type]
            {"customer_intent": "high"},
            event_id="evt_001",
            timestamp=datetime(2026, 8, 7, tzinfo=UTC),
        )
        self.assertEqual(payload["id"], "evt_001")
        self.assertEqual(payload["event"], "call.analysis.completed")
        self.assertEqual(payload["data"]["conversationId"], "conv_001")
        self.assertEqual(
            payload["data"]["analysis"], {"customer_intent": "high"}
        )


if __name__ == "__main__":
    unittest.main()