"""Быстрые тесты `src/core/nats.py` без сервера NATS: аутентификация, supervisor первого подключения, publish.

Клиент nats-py подменяется фабрикой (`client_factory`), паузы backoff — инжектируемым `sleep`
(spec recognition-job-exchange, «Соединение с NATS по шаблону, без JetStream»; design D-6).
"""

import ast
import asyncio
import os
from collections.abc import Iterator
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import nkeys
import pytest
from loguru import logger

from src.contract.subjects import NATS_MSG_ID_HEADER, RESULT_SUBJECT
from src.core.config import NATSSettings
from src.core.nats import (
    RESULT_QUEUE_GROUP,
    NatsConfigError,
    NatsConnection,
    NatsNotConnectedError,
    NatsPublishError,
    build_nats_auth,
    used_subjects,
)

SRC_DIR = Path(__file__).resolve().parent.parent / "src"


def _nats_settings(seed: str = "", jwt: str = "") -> NATSSettings:
    return NATSSettings(NATS_URL="nats://localhost:4222", NATS_SEED=seed, NATS_JWT=jwt)


def _valid_seed() -> str:
    return nkeys.encode_seed(os.urandom(32), nkeys.PREFIX_BYTE_USER).decode("utf-8")


@pytest.fixture
def log_records() -> Iterator[list[Any]]:
    """Записи loguru, выпущенные за время теста."""
    records: list[Any] = []
    handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG")
    yield records
    logger.remove(handler_id)


# --- build_nats_auth ----------------------------------------------------------------------------
def test_build_nats_auth_empty_credentials_gives_no_kwargs():
    assert build_nats_auth(_nats_settings()) == {}


@pytest.mark.parametrize(("seed", "jwt"), [("SU-only-seed", ""), ("", "only-jwt")])
def test_build_nats_auth_partial_credentials_is_config_error(seed: str, jwt: str):
    with pytest.raises(NatsConfigError, match="NATS_SEED"):
        build_nats_auth(_nats_settings(seed, jwt))


def test_build_nats_auth_invalid_seed_is_config_error():
    with pytest.raises(NatsConfigError, match="NATS_SEED"):
        build_nats_auth(_nats_settings("not-a-seed", "header.payload.signature"))


def test_nats_config_error_is_runtime_error():
    assert issubclass(NatsConfigError, RuntimeError)


def test_build_nats_auth_valid_credentials_give_callbacks():
    auth = build_nats_auth(_nats_settings(_valid_seed(), "header.payload.signature"))

    assert set(auth) == {"user_jwt_cb", "signature_cb"}
    assert auth["user_jwt_cb"]() == b"header.payload.signature"
    signature = auth["signature_cb"]("nonce")
    assert isinstance(signature, bytes)
    assert signature


# --- поддельный клиент nats-py ------------------------------------------------------------------
class _FakeClient:
    """Минимальный двойник `nats.aio.client.Client`: подключение, подписка, публикация, закрытие."""

    def __init__(self, world: "_FakeWorld") -> None:
        self.world = world
        self.is_connected = False
        self.is_closed = False
        self.connect_options: dict[str, Any] = {}
        self.subscriptions: list[dict[str, Any]] = []
        self.published: list[dict[str, Any]] = []
        self.flush_error: Exception | None = None
        self.drained = False

    async def connect(self, servers: str, **options: Any) -> None:
        self.world.connect_calls += 1
        self.connect_options = {"servers": servers, **options}
        if self.world.connect_calls <= self.world.fail_times:
            raise OSError("Connection refused")
        self.is_connected = True

    async def subscribe(self, subject: str, queue: str = "", cb: Any = None) -> SimpleNamespace:
        self.subscriptions.append({"subject": subject, "queue": queue, "cb": cb})
        return SimpleNamespace(subject=subject, queue=queue)

    async def publish(self, subject: str, payload: bytes = b"", headers: dict[str, str] | None = None) -> None:
        self.published.append({"subject": subject, "payload": payload, "headers": headers})

    async def flush(self, timeout: int = 10) -> None:
        if self.flush_error is not None:
            raise self.flush_error

    async def drain(self) -> None:
        self.drained = True
        await self.close()

    async def close(self) -> None:
        self.is_connected = False
        self.is_closed = True


class _FakeWorld:
    """Фабрика клиентов: первые `fail_times` подключений падают, затем успех."""

    def __init__(self, fail_times: int = 0) -> None:
        self.fail_times = fail_times
        self.connect_calls = 0
        self.clients: list[_FakeClient] = []

    def __call__(self) -> _FakeClient:
        client = _FakeClient(self)
        self.clients.append(client)
        return client


class _RecordingSleep:
    """Подмена `asyncio.sleep`: запоминает интервалы backoff и не ждёт."""

    def __init__(self) -> None:
        self.delays: list[float] = []

    async def __call__(self, delay: float) -> None:
        self.delays.append(delay)
        await asyncio.sleep(0)


async def _handler(msg: Any) -> None:
    return None


# --- supervisor первого подключения -------------------------------------------------------------
async def test_supervisor_retries_with_backoff_then_connects_and_subscribes_once(log_records: list[Any]):
    world = _FakeWorld(fail_times=7)
    sleep = _RecordingSleep()
    subscribed: list[NatsConnection] = []

    async def on_connected(conn: NatsConnection) -> None:
        subscribed.append(conn)
        await conn.subscribe_results(_handler)

    conn = NatsConnection(_nats_settings(), on_connected=on_connected, client_factory=world, sleep=sleep)
    assert not conn.is_connected

    await conn.start()
    assert await conn.wait_ready(timeout=2)

    assert conn.is_connected
    assert world.connect_calls == 8  # noqa: PLR2004
    assert sleep.delays == [1, 2, 4, 8, 16, 30, 30]
    assert subscribed == [conn]
    connected_client = world.clients[-1]
    assert [(s["subject"], s["queue"]) for s in connected_client.subscriptions] == [
        (RESULT_SUBJECT, RESULT_QUEUE_GROUP)
    ]
    options = connected_client.connect_options
    assert options["servers"] == "nats://localhost:4222"
    assert options["max_reconnect_attempts"] == -1
    assert options["reconnect_time_wait"] == 2  # noqa: PLR2004
    for callback in ("error_cb", "disconnected_cb", "reconnected_cb", "closed_cb"):
        assert callable(options[callback])

    # Первая ошибка — warning с причиной, остальные в пределах минуты подавлены
    warnings = [r for r in log_records if r["level"].name == "WARNING"]
    assert len(warnings) == 1
    assert "error_category=nats_error" in warnings[0]["message"]
    assert "Connection refused" in warnings[0]["message"]

    await conn.stop()
    assert connected_client.is_closed


async def test_supervisor_connects_first_time_without_sleep():
    world = _FakeWorld()
    sleep = _RecordingSleep()
    conn = NatsConnection(_nats_settings(), client_factory=world, sleep=sleep)

    await conn.start()
    assert await conn.wait_ready(timeout=2)

    assert sleep.delays == []
    assert world.connect_calls == 1
    await conn.stop()


async def test_start_does_not_block_and_stop_cancels_supervisor():
    world = _FakeWorld(fail_times=10**6)
    never = asyncio.Event()

    async def blocking_sleep(delay: float) -> None:
        await never.wait()

    conn = NatsConnection(_nats_settings(), client_factory=world, sleep=blocking_sleep)
    await conn.start()
    await asyncio.sleep(0.01)

    assert not conn.is_connected
    assert world.connect_calls == 1

    await conn.stop()  # без исключений наружу
    assert not conn.is_connected
    assert not await conn.wait_ready(timeout=0.01)


async def test_config_error_logged_as_config_error_and_not_retried(log_records: list[Any]):
    world = _FakeWorld()
    sleep = _RecordingSleep()
    conn = NatsConnection(_nats_settings(seed="SU-only-seed"), client_factory=world, sleep=sleep)

    await conn.start()
    assert not await conn.wait_ready(timeout=0.05)

    assert world.connect_calls == 0
    assert sleep.delays == []
    assert not conn.is_connected
    errors = [r for r in log_records if r["level"].name == "ERROR"]
    assert len(errors) == 1
    assert "error_category=config_error" in errors[0]["message"]
    assert "nats_error" not in errors[0]["message"]
    await conn.stop()


async def test_on_connected_failure_is_retried(log_records: list[Any]):
    world = _FakeWorld()
    sleep = _RecordingSleep()
    calls = 0

    async def flaky_on_connected(conn: NatsConnection) -> None:
        nonlocal calls
        calls += 1
        if calls < 3:  # noqa: PLR2004
            raise RuntimeError("подписка не удалась")

    conn = NatsConnection(_nats_settings(), on_connected=flaky_on_connected, client_factory=world, sleep=sleep)
    await conn.start()
    assert await conn.wait_ready(timeout=2)

    assert calls == 3  # noqa: PLR2004
    assert sleep.delays == [1, 2]
    await conn.stop()


async def test_stop_without_start_is_noop():
    conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld())
    await conn.stop()
    assert not conn.is_connected


async def test_error_callback_logging_is_throttled_and_reset_on_reconnect(log_records: list[Any]):
    world = _FakeWorld()
    conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep())
    await conn.start()
    assert await conn.wait_ready(timeout=2)
    options = world.clients[-1].connect_options

    await options["disconnected_cb"]()
    for _ in range(5):
        await options["error_cb"](OSError("Connection refused"))
    await options["reconnected_cb"]()
    await options["error_cb"](OSError("Connection reset"))

    warnings = [r["message"] for r in log_records if r["level"].name == "WARNING"]
    errors_logged = [m for m in warnings if "error_category=nats_error" in m]
    assert len(errors_logged) == 2  # noqa: PLR2004 - первая ошибка серии и первая после переподключения
    assert any("Connection reset" in m for m in errors_logged)
    await conn.stop()


# --- publish / subscribe ------------------------------------------------------------------------
async def test_publish_without_connection_raises_publish_error():
    conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld())

    with pytest.raises(NatsPublishError, match="NATS"):
        await conn.publish("files.tabular_uploaded", b"{}", "job:1")


async def test_publish_sends_msg_id_header_and_flushes():
    world = _FakeWorld()
    conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep())
    await conn.start()
    assert await conn.wait_ready(timeout=2)

    await conn.publish("files.tabular_uploaded", b'{"a": 1}', "job:1")

    assert world.clients[-1].published == [
        {"subject": "files.tabular_uploaded", "payload": b'{"a": 1}', "headers": {NATS_MSG_ID_HEADER: "job:1"}}
    ]
    await conn.stop()


async def test_publish_flush_failure_raises_publish_error():
    world = _FakeWorld()
    conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep())
    await conn.start()
    assert await conn.wait_ready(timeout=2)
    world.clients[-1].flush_error = TimeoutError()

    with pytest.raises(NatsPublishError):
        await conn.publish("files.tabular_uploaded", b"{}", "job:1")
    await conn.stop()


async def test_subscribe_results_handler_errors_are_logged_not_raised(log_records: list[Any]):
    world = _FakeWorld()
    conn = NatsConnection(
        _nats_settings(), client_factory=world, sleep=_RecordingSleep(), result_subject="test.subject", queue="g"
    )
    await conn.start()
    assert await conn.wait_ready(timeout=2)

    async def failing_handler(msg: Any) -> None:
        raise ValueError("сломанный обработчик")

    await conn.subscribe_results(failing_handler)
    subscription = world.clients[-1].subscriptions[0]
    assert (subscription["subject"], subscription["queue"]) == ("test.subject", "g")

    await subscription["cb"](SimpleNamespace(subject="test.subject", data=b"xyz", headers={}))

    errors = [r for r in log_records if r["level"].name == "ERROR"]
    assert len(errors) == 1
    assert "error_category=handler_error" in errors[0]["message"]
    await conn.stop()


async def test_subscribe_results_without_connection_raises():
    conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld())
    with pytest.raises(NatsNotConnectedError):
        await conn.subscribe_results(_handler)


async def test_stop_drains_connected_client():
    world = _FakeWorld()
    conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep())
    await conn.start()
    assert await conn.wait_ready(timeout=2)

    await conn.stop()

    assert world.clients[-1].drained
    assert not conn.is_connected


# --- без JetStream -------------------------------------------------------------------------------
def test_src_does_not_use_jetstream():
    """JetStream на платформе нет: в src/ нет вызовов `.jetstream()` и импортов `nats.js`."""
    offenders: list[str] = []
    for path in SRC_DIR.rglob("*.py"):
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            if isinstance(node, ast.Attribute) and node.attr.lower() == "jetstream":
                offenders.append(f"{path}:{node.lineno} .{node.attr}")
            elif isinstance(node, ast.ImportFrom) and (node.module or "").startswith("nats.js"):
                offenders.append(f"{path}:{node.lineno} from {node.module}")
            elif isinstance(node, ast.Import):
                offenders.extend(
                    f"{path}:{node.lineno} import {alias.name}"
                    for alias in node.names
                    if alias.name.startswith("nats.js")
                )
    assert offenders == []


def test_used_subjects_core() -> None:
    nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
    assert used_subjects(nats) == {
        "tabular_request": "files.tabular_uploaded",
        "image_request": "files.img_uploaded",
        "result": "test.files.tabular_parsed",
    }


def test_connection_subscribes_to_configured_result_subject() -> None:
    nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
    assert NatsConnection(nats).result_subject == "test.files.tabular_parsed"
    assert NatsConnection(nats, result_subject="override").result_subject == "override"


async def test_connection_subscribes_with_configured_result_subject() -> None:
    """T3.2: subscribe вызывается с subject'ом из настроек (без override) и группой очереди."""
    nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
    world = _FakeWorld()
    conn = NatsConnection(nats, client_factory=world, sleep=_RecordingSleep(), queue="g")
    await conn.start()
    assert await conn.wait_ready(timeout=2)

    await conn.subscribe_results(_handler)

    subscription = world.clients[-1].subscriptions[0]
    assert (subscription["subject"], subscription["queue"]) == ("test.files.tabular_parsed", "g")
    await conn.stop()