from __future__ import annotations

import hashlib
import json
import threading
import unittest
from http.client import HTTPConnection
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Any
from urllib.parse import quote

from agent_call.cell_agent import (
    AsteriskConfigApplier,
    CellAgentError,
    CellAgentService,
    make_server,
)
from agent_call.sip_management import (
    SipManagementError,
    SipManagementService,
    _validate_trunk,
)


class FakeApplier:
    def __init__(self) -> None:
        self.calls: list[tuple[str, dict[str, Any]]] = []
        self.fail = False

    def apply(self, trunk_id: str, config: dict[str, Any]) -> None:
        if self.fail:
            raise CellAgentError(502, "ASTERISK_RELOAD_FAILED", "reload failed")
        self.calls.append((trunk_id, config))


class CellAgentTests(unittest.TestCase):
    def test_revision_hash_idempotency_and_gap_protection(self) -> None:
        with TemporaryDirectory() as directory:
            applier = FakeApplier()
            service = CellAgentService(
                Path(directory) / "cell.sqlite3", "cell-a", applier
            )
            try:
                body = self._publication("cell-a", "trunk:a", 1)
                first = service.apply(body, "request-1")
                second = service.apply(body, "request-1-retry")
                self.assertFalse(first["idempotent"])
                self.assertTrue(second["idempotent"])
                self.assertEqual(len(applier.calls), 1)
                with self.assertRaisesRegex(CellAgentError, "next revision"):
                    service.apply(
                        self._publication("cell-a", "trunk:a", 3), "request-3"
                    )
                bad = self._publication("cell-a", "trunk:a", 2)
                bad["config_sha256"] = "0" * 64
                with self.assertRaisesRegex(CellAgentError, "does not match"):
                    service.apply(bad, "request-bad")
            finally:
                service.close()

    def test_http_endpoint_unquotes_path_and_rejects_mismatch(self) -> None:
        with TemporaryDirectory() as directory:
            service = CellAgentService(
                Path(directory) / "cell.sqlite3", "cell-a", FakeApplier()
            )
            server = make_server(service, "127.0.0.1", 0)
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()
            try:
                body = self._publication("cell-a", "trunk:a", 1)
                response = self._request(
                    server.server_port,
                    f"/v1/sip/trunks/{quote('trunk:a', safe='')}/apply",
                    body,
                )
                self.assertEqual(response.status, 200)
                mismatch = self._publication("cell-a", "other", 2)
                response = self._request(
                    server.server_port,
                    "/v1/sip/trunks/trunk%3Aa/apply",
                    mismatch,
                )
                self.assertEqual(response.status, 400)
            finally:
                server.shutdown()
                server.server_close()
                thread.join(timeout=2)
                service.close()

    def test_asterisk_applier_writes_codec_and_restores_on_reload_failure(self) -> None:
        with TemporaryDirectory() as directory:

            class RecordingApplier(AsteriskConfigApplier):
                def __init__(self, config_dir: str) -> None:
                    super().__init__(config_dir)
                    self.fail = False

                def _reload(self) -> None:
                    if self.fail:
                        raise CellAgentError(
                            502, "ASTERISK_RELOAD_FAILED", "reload failed"
                        )

            applier = RecordingApplier(directory)
            config = self._config()
            applier.apply("trunk:a", config)
            target = next(Path(directory).glob("*.conf"))
            original = target.read_text(encoding="utf-8")
            self.assertIn("allow=alaw", original)
            self.assertIn("from_user=BD93205882", original)
            applier.fail = True
            with self.assertRaises(CellAgentError):
                applier.apply(
                    "trunk:a",
                    {
                        **config,
                        "codec_profile": {"allowed": ["PCMU"], "preferred": "PCMU"},
                    },
                )
            self.assertEqual(target.read_text(encoding="utf-8"), original)

    def test_real_management_activates_only_after_cell_ack(self) -> None:
        class Publisher:
            def __init__(self) -> None:
                self.fail = False
                self.calls: list[int] = []

            def publish(
                self,
                cell: dict[str, Any],
                trunk_id: str,
                revision: int,
                config: dict[str, Any],
                request_id: str,
            ) -> dict[str, Any]:
                self.calls.append(revision)
                if self.fail:
                    raise OSError("offline")
                digest = hashlib.sha256(
                    json.dumps(
                        config,
                        ensure_ascii=False,
                        sort_keys=True,
                        separators=(",", ":"),
                    ).encode()
                ).hexdigest()
                return {
                    "mode": "real",
                    "cell_id": cell["cell_id"],
                    "trunk_id": trunk_id,
                    "revision": revision,
                    "config_sha256": digest,
                    "status": "applied",
                }

        with TemporaryDirectory() as directory:
            publisher = Publisher()
            service = SipManagementService(
                Path(directory) / "sip.sqlite3", mode="real", publisher=publisher
            )
            try:
                service.upsert_cell(
                    "cell-a",
                    {
                        "egress_pool_id": "egress-a",
                        "codec_capabilities": ["PCMA"],
                        "status": "healthy",
                        "max_concurrency": 10,
                        "management_url": "https://cell-a.internal:9443",
                    },
                    expected_revision=0,
                    actor="ops",
                    request_id="cell-1",
                )
                service.upsert_trunk(
                    "trunk:a",
                    self._config(),
                    expected_revision=0,
                    actor="ops",
                    request_id="trunk-1",
                )
                published = service.publish_trunk(
                    "trunk:a", expected_revision=1, actor="ops", request_id="pub-1"
                )
                self.assertEqual(published["active_revision"], 1)
                self.assertEqual(
                    service.list_publications("trunk:a")[0]["status"], "applied"
                )
                publisher.fail = True
                service.upsert_trunk(
                    "trunk:a",
                    {**self._config(), "display_name": "Provider B"},
                    expected_revision=1,
                    actor="ops",
                    request_id="trunk-2",
                )
                with self.assertRaisesRegex(SipManagementError, "every Cell"):
                    service.publish_trunk(
                        "trunk:a", expected_revision=2, actor="ops", request_id="pub-2"
                    )
                self.assertEqual(service.get_trunk("trunk:a")["active_revision"], 1)
                self.assertEqual(
                    service.list_publications("trunk:a")[0]["status"], "applied"
                )
                self.assertEqual(
                    service.list_publications("trunk:a")[1]["status"], "failed"
                )
            finally:
                service.close()

    def test_real_publication_resumes_after_process_crash(self) -> None:
        class CrashOncePublisher:
            def __init__(self) -> None:
                self.crashed = False

            def publish(
                self,
                cell: dict[str, Any],
                trunk_id: str,
                revision: int,
                config: dict[str, Any],
                request_id: str,
            ) -> dict[str, Any]:
                if not self.crashed:
                    self.crashed = True
                    raise KeyboardInterrupt
                digest = hashlib.sha256(
                    json.dumps(
                        config,
                        ensure_ascii=False,
                        sort_keys=True,
                        separators=(",", ":"),
                    ).encode()
                ).hexdigest()
                return {
                    "mode": "real",
                    "cell_id": cell["cell_id"],
                    "trunk_id": trunk_id,
                    "revision": revision,
                    "config_sha256": digest,
                    "status": "applied",
                }

        with TemporaryDirectory() as directory:
            publisher = CrashOncePublisher()
            service = SipManagementService(
                Path(directory) / "sip.sqlite3", mode="real", publisher=publisher
            )
            try:
                service.upsert_cell(
                    "cell-a",
                    {
                        "egress_pool_id": "egress-a",
                        "codec_capabilities": ["PCMA"],
                        "status": "healthy",
                        "max_concurrency": 10,
                        "management_url": "https://cell-a.internal:9443",
                    },
                    expected_revision=0,
                    actor="ops",
                    request_id="cell-1",
                )
                service.upsert_trunk(
                    "trunk:a",
                    self._config(),
                    expected_revision=0,
                    actor="ops",
                    request_id="trunk-1",
                )
                with self.assertRaises(KeyboardInterrupt):
                    service.publish_trunk(
                        "trunk:a", expected_revision=1, actor="ops", request_id="pub-1"
                    )
                self.assertEqual(service.get_trunk("trunk:a")["active_revision"], 0)
                resumed = service.publish_trunk(
                    "trunk:a", expected_revision=1, actor="ops", request_id="pub-2"
                )
                self.assertEqual(resumed["active_revision"], 1)
                self.assertEqual(
                    service.list_publications("trunk:a")[0]["status"], "applied"
                )
            finally:
                service.close()

    @staticmethod
    def _config() -> dict[str, Any]:
        return {
            "display_name": "Provider A",
            "enabled": True,
            "sip": {
                "host": "61.132.228.221",
                "port": 5060,
                "transport": "udp",
                "auth_mode": "ip",
                "register": False,
            },
            "codec_profile": {"allowed": ["PCMA"], "preferred": "PCMA"},
            "caller_ids": ["BD93205882"],
            "dial_prefix": "7089",
            "egress_pool_id": "egress-a",
            "max_concurrency": 100,
            "max_cps": 10,
        }

    @classmethod
    def _publication(cls, cell_id: str, trunk_id: str, revision: int) -> dict[str, Any]:
        config = _validate_trunk(trunk_id, cls._config())
        digest = hashlib.sha256(
            json.dumps(
                config, ensure_ascii=False, sort_keys=True, separators=(",", ":")
            ).encode()
        ).hexdigest()
        return {
            "mode": "real",
            "cell_id": cell_id,
            "trunk_id": trunk_id,
            "revision": revision,
            "config": config,
            "config_sha256": digest,
        }

    @staticmethod
    def _request(port: int, path: str, body: dict[str, Any]):
        encoded = json.dumps(body, ensure_ascii=False).encode()
        connection = HTTPConnection("127.0.0.1", port, timeout=3)
        connection.request(
            "POST",
            path,
            body=encoded,
            headers={
                "Content-Type": "application/json",
                "Content-Length": str(len(encoded)),
                "X-Request-ID": "http-test",
            },
        )
        response = connection.getresponse()
        response.read()
        connection.close()
        return response


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