"""Tests for the srs-gate Hermes plugin. Standard library only.

Run from the repo root:
    python -m unittest discover -s hermes-plugin/srs-gate -v

All phone digits, ids and the secret below are placeholders, not real values.
"""
from __future__ import annotations

import asyncio
import enum
import importlib.util
import inspect
import json
import logging
import os
import socket
import tempfile
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from types import SimpleNamespace
from unittest import mock

HERE = Path(__file__).resolve().parent


def _load_plugin():
    spec = importlib.util.spec_from_file_location("srs_gate", HERE / "__init__.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


gate = _load_plugin()

SKIP = {"action": "skip", "reason": "srs-gate"}
SECRET = "test-only-secret-not-real-0123456789"
PHONE = "910000000001"
SENDER = PHONE + "@s.whatsapp.net"
CHAT = "100000000000077@lid"
MID = "3EB0TESTMSGID01"
TEXT = "Shop Zq7\n0012-5\n4567 x 10\ncall " + PHONE
SENSITIVE = [SECRET, PHONE, SENDER, CHAT, "100000000000077", "Zq7", "4567 x 10"]


class MessageType(enum.Enum):
    TEXT = "text"
    LOCATION = "location"
    PHOTO = "photo"
    VIDEO = "video"
    AUDIO = "audio"
    VOICE = "voice"
    DOCUMENT = "document"
    STICKER = "sticker"
    COMMAND = "command"


def make_event(text=TEXT):
    source = SimpleNamespace(
        platform="whatsapp",
        chat_id=CHAT,
        chat_type="dm",
        user_id=SENDER,
        user_id_alt=None,
        chat_id_alt=None,
        is_bot=False,
        message_id=MID,
    )
    return SimpleNamespace(
        text=text,
        message_type=MessageType.TEXT,
        message_id=MID,
        source=source,
        media_urls=[],
        media_types=[],
        internal=False,
        edit_revision=None,
        metadata={},
        user_id=None,
    )


class ExplodingEvent:
    @property
    def source(self):
        raise RuntimeError("boom")

    @property
    def internal(self):
        raise RuntimeError("boom")


# ---------------------------------------------------------------- stub srs-api


class _Server(ThreadingHTTPServer):
    daemon_threads = True
    block_on_close = False

    def handle_error(self, request, client_address):
        pass  


class Stub:
    """Stands in for srs-api. `script` is a list of actions, one per request; after it runs out, 200.

    Actions: ("status", code, delay_s) or ("garbage",).
    """

    def __init__(self, script=None):
        self.requests = []
        self.script = list(script or [])
        self._lock = threading.Lock()
        stub = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *args):
                pass

            def do_POST(self):
                length = int(self.headers.get("content-length") or 0)
                body = self.rfile.read(length)
                with stub._lock:
                    stub.requests.append(
                        {"method": "POST", "path": self.path, "headers": dict(self.headers), "body": body}
                    )
                    action = stub.script.pop(0) if stub.script else ("status", 200, 0)
                if action[0] == "garbage":
                    self.wfile.write(b"this is not http\r\n\r\n")
                    self.close_connection = True
                    return
                _, code, delay = action
                if delay:
                    time.sleep(delay)
                payload = b'{"handled":true,"duplicate":false}'
                self.send_response(code)
                if 300 <= code < 400:
                    self.send_header("location", "/elsewhere")
                self.send_header("content-type", "application/json")
                self.send_header("content-length", str(len(payload)))
                self.end_headers()
                self.wfile.write(payload)

            do_GET = do_POST

        self.server = _Server(("127.0.0.1", 0), Handler)
        self.url = "http://127.0.0.1:%d" % self.server.server_address[1]
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()

    def stop(self):
        self.server.shutdown()
        self.server.server_close()


def closed_port_url():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return "http://127.0.0.1:%d" % port


class ListHandler(logging.Handler):
    def __init__(self):
        super().__init__(level=logging.DEBUG)
        self.records = []

    def emit(self, record):
        self.records.append(record)


# ---------------------------------------------------------------- base


class GateCase(unittest.IsolatedAsyncioTestCase):
    """Fresh stub, config file, fast timings and a log capture for every test.

    Every test also checks, on teardown, that no log record anywhere holds the text, the secret or a number.
    """

    def setUp(self):
        self.stub = Stub()
        self.addCleanup(self.stub.stop)
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        env = mock.patch.dict(os.environ, {"HERMES_HOME": self.tmp.name})
        env.start()
        self.addCleanup(env.stop)
        self.write_config(url=self.stub.url)
        for name, value in (("TRIES", 3), ("TRY_TIMEOUT_S", 1.0), ("BACKOFF_S", 0.01), ("OVERALL_TIMEOUT_S", 5.0)):
            p = mock.patch.object(gate, name, value)
            p.start()
            self.addCleanup(p.stop)
        gate._config_warned = False
        self.logs = ListHandler()
        root = logging.getLogger()
        self._old_level = root.level
        root.addHandler(self.logs)
        root.setLevel(logging.DEBUG)
        self.addCleanup(self._restore_logging)

    def _restore_logging(self):
        root = logging.getLogger()
        root.removeHandler(self.logs)
        root.setLevel(self._old_level)

    def tearDown(self):
        self.assert_no_sensitive_logs()

    # helpers

    def write_config(self, url=None, secret=SECRET, extra=""):
        lines = ["# srs-gate test config"]
        if url is not None:
            lines.append("SRS_URL=" + url)
        if secret is not None:
            lines.append("SRS_SECRET=" + secret)
        Path(self.tmp.name, "srs-gate.env").write_text("\n".join(lines) + "\n" + extra, encoding="utf-8")

    def config_path(self):
        return Path(self.tmp.name, "srs-gate.env")

    async def call(self, event):
        return await gate.on_inbound(event=event, gateway=object(), session_store=object())

    def log_text(self):
        out = []
        for r in self.logs.records:
            out.append(r.getMessage())
            out.append(repr(r.args))
            if r.exc_info or r.exc_text:
                out.append(logging.Formatter().formatException(r.exc_info) if r.exc_info else r.exc_text)
        return "\n".join(out)

    def assert_no_sensitive_logs(self):
        text = self.log_text()
        for needle in SENSITIVE:
            self.assertNotIn(needle, text, "sensitive value in logs")

    def gate_messages(self):
        return [r.getMessage() for r in self.logs.records if r.name == "srs_gate"]


# ---------------------------------------------------------------- payload


class PayloadTests(GateCase):
    def test_exact_payload_for_direct_text(self):
        self.assertEqual(
            gate._payload(make_event()),
            {"kind": "text", "message_id": MID, "sender_jid": SENDER, "chat_id": CHAT, "text": TEXT},
        )

    def test_plain_string_message_type_is_accepted(self):
        ev = make_event()
        ev.message_type = "text"
        self.assertIsNotNone(gate._payload(ev))

    async def test_text_arrives_byte_for_byte(self):
        texts = {
            "zeros_and_breaks": "0012\n0099\r\n  0001  \n\n0000 - 5\t",
            "malayalam": "ഷോപ്പ് 0012 - 5",
            "hundred_lines": "\n".join("%04d - %d" % (i, i % 7 + 1) for i in range(100)),
        }
        for name, text in texts.items():
            with self.subTest(name):
                before = len(self.stub.requests)
                self.assertEqual(await self.call(make_event(text)), SKIP)
                self.assertEqual(len(self.stub.requests), before + 1)
                body = json.loads(self.stub.requests[-1]["body"].decode("utf-8"))
                self.assertEqual(body["text"], text)
                self.assertEqual(body["kind"], "text")


# ---------------------------------------------------------------- filters


def _set(path, value):
    def apply(ev):
        target = ev
        parts = path.split(".")
        for p in parts[:-1]:
            target = getattr(target, p)
        setattr(target, parts[-1], value)

    return apply


def _delete(path):
    def apply(ev):
        target = ev
        parts = path.split(".")
        for p in parts[:-1]:
            target = getattr(target, p)
        delattr(target, parts[-1])

    return apply


NOT_FORWARDED = [
    ("group chat_type", _set("source.chat_type", "group")),
    ("channel chat_type", _set("source.chat_type", "channel")),
    ("thread chat_type", _set("source.chat_type", "thread")),
    ("unknown chat_type", _set("source.chat_type", "something-new")),
    ("chat_type wrong case", _set("source.chat_type", "DM")),
    ("chat_type None", _set("source.chat_type", None)),
    ("chat_type missing", _delete("source.chat_type")),
    ("group chat id", _set("source.chat_id", "120363000000000000@g.us")),
    ("internal", _set("internal", True)),
    ("is_bot", _set("source.is_bot", True)),
    ("edit_revision set", _set("edit_revision", "2")),
    ("edit_revision int", _set("edit_revision", 1)),
    ("blank text", _set("text", "")),
    ("whitespace text", _set("text", "  \n\t ")),
    ("text None", _set("text", None)),
    ("text not a string", _set("text", 1234)),
    ("missing message id", _set("message_id", None)),
    ("empty message id", _set("message_id", "")),
    ("missing sender", _set("source.user_id", None)),
    ("missing chat id", _set("source.chat_id", "")),
    ("no source", _set("source", None)),
    ("message_type missing", _set("message_type", None)),
] + [
    ("message_type " + t.value, _set("message_type", t)) for t in MessageType if t is not MessageType.TEXT
]


class FilterTests(GateCase):
    async def test_not_forwarded_but_still_skipped(self):
        for name, mutate in NOT_FORWARDED:
            with self.subTest(name):
                ev = make_event()
                mutate(ev)
                self.assertIsNone(gate._payload(ev))
                self.assertEqual(await self.call(ev), SKIP)
        self.assertEqual(self.stub.requests, [])

    async def test_photo_placeholder_text_is_not_forwarded(self):
        ev = make_event("[image received]")
        ev.message_type = MessageType.PHOTO
        self.assertEqual(await self.call(ev), SKIP)
        self.assertEqual(self.stub.requests, [])


# ---------------------------------------------------------------- request


class RequestTests(GateCase):
    async def test_headers_method_and_path(self):
        self.assertEqual(await self.call(make_event()), SKIP)
        self.assertEqual(len(self.stub.requests), 1)
        req = self.stub.requests[0]
        headers = {k.lower(): v for k, v in req["headers"].items()}
        self.assertEqual(req["method"], "POST")
        self.assertEqual(req["path"], "/ingest")
        self.assertEqual(headers["x-srs-secret"], SECRET)
        self.assertEqual(headers["content-type"], "application/json")

    async def test_trailing_slash_in_url(self):
        self.write_config(url=self.stub.url + "/")
        await self.call(make_event())
        self.assertEqual(self.stub.requests[0]["path"], "/ingest")

    async def test_quoted_config_values(self):
        self.write_config(url='"%s"' % self.stub.url, secret="'%s'" % SECRET)
        await self.call(make_event())
        headers = {k.lower(): v for k, v in self.stub.requests[0]["headers"].items()}
        self.assertEqual(headers["x-srs-secret"], SECRET)

    def test_proxy_environment_is_ignored(self):
        dead = closed_port_url()
        with mock.patch.dict(os.environ, {"http_proxy": dead, "https_proxy": dead, "no_proxy": ""}):
            fresh = _load_plugin()
            status, error = fresh._post(
                {"SRS_URL": self.stub.url, "SRS_SECRET": SECRET}, fresh._payload(make_event())
            )
        self.assertEqual((status, error), (200, None))
        self.assertEqual(len(self.stub.requests), 1)

    async def test_success_log_has_id_and_status(self):
        await self.call(make_event())
        msgs = self.gate_messages()
        self.assertEqual(len(msgs), 1)
        self.assertIn("id=" + MID, msgs[0])
        self.assertIn("status=200", msgs[0])


# ---------------------------------------------------------------- retries


class RetryTests(GateCase):
    async def test_500_then_200_succeeds_on_second_try(self):
        self.stub.script = [("status", 500, 0), ("status", 200, 0)]
        self.assertEqual(await self.call(make_event()), SKIP)
        self.assertEqual(len(self.stub.requests), 2)
        self.assertIn("status=200", self.gate_messages()[-1])

    async def test_500_three_times_gives_up(self):
        self.stub.script = [("status", 500, 0)] * 5
        self.assertEqual(await self.call(make_event()), SKIP)
        self.assertEqual(len(self.stub.requests), 3)
        self.assertIn("status=500", self.gate_messages()[-1])

    async def test_4xx_and_3xx_are_not_retried(self):
        for code in (400, 401, 404, 302):
            with self.subTest(code):
                self.stub.requests.clear()
                self.stub.script = [("status", code, 0), ("status", 200, 0)]
                self.assertEqual(await self.call(make_event()), SKIP)
                self.assertEqual(len(self.stub.requests), 1, "retried or followed a redirect")
                self.assertIn("status=%d" % code, self.gate_messages()[-1])

    def test_connection_refused_ends_after_three_tries(self):
        cfg = {"SRS_URL": closed_port_url(), "SRS_SECRET": SECRET}
        with mock.patch.object(gate, "TRY_TIMEOUT_S", 0.5), mock.patch.object(
            gate._OPENER, "open", wraps=gate._OPENER.open
        ) as opened:
            status, error = gate._post(cfg, gate._payload(make_event()))
        self.assertEqual(opened.call_count, 3)
        self.assertIsNone(status)
        self.assertIsNotNone(error)

    async def test_slow_server_hits_per_try_timeout_then_retries(self):
        self.stub.script = [("status", 200, 0.6), ("status", 200, 0)]
        with mock.patch.object(gate, "TRY_TIMEOUT_S", 0.2):
            self.assertEqual(await self.call(make_event()), SKIP)
        self.assertEqual(len(self.stub.requests), 2)
        self.assertIn("status=200", self.gate_messages()[-1])


# ---------------------------------------------------------------- fail closed


class FailClosedTests(GateCase):
    async def test_opener_raises(self):
        with mock.patch.object(gate._OPENER, "open", side_effect=RuntimeError("x")):
            self.assertEqual(await self.call(make_event()), SKIP)

    async def test_bad_events(self):
        for name, ev in (("None", None), ("object", object()), ("exploding", ExplodingEvent()), ("string", "hi")):
            with self.subTest(name):
                self.assertEqual(await self.call(ev), SKIP)
        self.assertEqual(await gate.on_inbound(), SKIP)  
        self.assertEqual(self.stub.requests, [])

    async def test_bad_config(self):
        cases = {
            "missing file": lambda: self.config_path().unlink(),
            "no secret": lambda: self.write_config(url=self.stub.url, secret=None),
            "empty secret": lambda: self.write_config(url=self.stub.url, secret=""),
            "no url": lambda: self.write_config(url=None),
            "file scheme": lambda: self.write_config(url="file:///etc/passwd"),
            "no scheme": lambda: self.write_config(url="127.0.0.1:8080"),
        }
        for name, breakit in cases.items():
            with self.subTest(name):
                breakit()
                self.assertEqual(await self.call(make_event()), SKIP)
        self.assertEqual(self.stub.requests, [])

    async def test_config_error_logged_once_until_fixed(self):
        self.write_config(url=self.stub.url, secret=None)
        await self.call(make_event())
        await self.call(make_event())
        self.assertEqual(len(self.gate_messages()), 1)
        self.write_config(url=self.stub.url)
        await self.call(make_event())  
        self.write_config(url=self.stub.url, secret=None)
        await self.call(make_event())
        config_warnings = [m for m in self.gate_messages() if "config" in m]
        self.assertEqual(len(config_warnings), 2)

    async def test_garbage_reply(self):
        self.stub.script = [("garbage",)] * 3
        self.assertEqual(await self.call(make_event()), SKIP)
        self.assertGreaterEqual(len(self.stub.requests), 1)

    async def test_overall_timeout(self):
        self.stub.script = [("status", 200, 0.5)]
        with mock.patch.object(gate, "OVERALL_TIMEOUT_S", 0.15), mock.patch.object(
            gate, "TRIES", 1
        ), mock.patch.object(gate, "TRY_TIMEOUT_S", 1.0):
            started = time.monotonic()
            self.assertEqual(await self.call(make_event()), SKIP)
            self.assertLess(time.monotonic() - started, 0.45)
        self.assertIn("TimeoutError", self.gate_messages()[-1])

    def test_post_never_raises_even_systemexit(self):
        cfg = {"SRS_URL": self.stub.url, "SRS_SECRET": SECRET}
        with mock.patch.object(gate._OPENER, "open", side_effect=SystemExit(3)):
            status, error = gate._post(cfg, gate._payload(make_event()))
        self.assertIsNone(status)
        self.assertEqual(error, "SystemExit")

    async def test_systemexit_in_http_call(self):
        with mock.patch.object(gate._OPENER, "open", side_effect=SystemExit(3)):
            self.assertEqual(await self.call(make_event()), SKIP)

    async def test_backstop_catches_systemexit_outside_the_thread(self):
        with mock.patch.object(gate, "_load_config", side_effect=SystemExit(3)):
            self.assertEqual(await self.call(make_event()), SKIP)

    async def test_backstop_catches_unexpected_error(self):
        with mock.patch.object(gate, "_payload", side_effect=ValueError("x")):
            self.assertEqual(await self.call(make_event()), SKIP)


# ---------------------------------------------------------------- loop behaviour


class LoopTests(GateCase):
    async def test_slow_srs_api_does_not_block_the_loop(self):
        self.stub.script = [("status", 200, 0.8)]
        ticks = 0
        done = asyncio.Event()

        async def ticker():
            nonlocal ticks
            while not done.is_set():
                await asyncio.sleep(0.02)
                ticks += 1

        async def run():
            try:
                return await self.call(make_event())
            finally:
                done.set()

        result, _ = await asyncio.gather(run(), ticker())
        self.assertEqual(result, SKIP)
        self.assertGreaterEqual(ticks, 15, "event loop was blocked while waiting for srs-api")

    async def test_cancellation_is_re_raised(self):
        self.stub.script = [("status", 200, 0.6)]
        with mock.patch.object(gate, "TRIES", 1):
            task = asyncio.ensure_future(self.call(make_event()))
            await asyncio.sleep(0.15)
            task.cancel()
            with self.assertRaises(asyncio.CancelledError):
                await task


# ---------------------------------------------------------------- registration


class RegistrationTests(unittest.TestCase):
    def test_register_hooks_once(self):
        calls = []
        ctx = SimpleNamespace(register_hook=lambda name, cb: calls.append((name, cb)))
        gate.register(ctx)
        self.assertEqual(calls, [("pre_gateway_dispatch", gate.on_inbound)])

    def test_callback_is_async(self):
        self.assertTrue(inspect.iscoroutinefunction(gate.on_inbound))

    def test_manifest(self):
        text = (HERE / "plugin.yaml").read_text(encoding="utf-8")
        self.assertIn("name: srs-gate", text)
        self.assertIn("- pre_gateway_dispatch", text)

    def test_returns_a_fresh_skip_each_time(self):
        a = asyncio.run(gate.on_inbound(event=None))
        a["action"] = "allow"
        self.assertEqual(asyncio.run(gate.on_inbound(event=None)), SKIP)


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