"""Tests for Baota get_config SSH detect caching and bootstrap lock."""

import asyncio
from unittest.mock import AsyncMock, patch

import pytest

from server.application.services.btpanel_service import BtPanelService
from server.domain.models import Server
from server.infrastructure.btpanel.bootstrap_lock import bootstrap_lock


class _FakeRedis:
    def __init__(self) -> None:
        self.data: dict[str, str] = {}

    async def set(self, key: str, value: str, ex: int | None = None, nx: bool = False):
        if nx and key in self.data:
            return False
        self.data[key] = value
        return True

    async def eval(self, _script: str, _numkeys: int, key: str, token: str):
        if self.data.get(key) == token:
            del self.data[key]
            return 1
        return 0


@pytest.mark.asyncio
async def test_resolve_bt_installed_uses_cache_when_disabled():
    server = Server(
        id=1, name="s", domain="1.2.3.4",
        extra_attrs={"bt_panel": {
            "bootstrap_state": "disabled",
            "bt_installed": True,
        }},
    )

    class _Repo:
        async def get_by_id(self, _sid):
            return server

    svc = BtPanelService(db=None)  # type: ignore[arg-type]
    svc.servers = _Repo()  # type: ignore[assignment]

    with patch(
        "server.application.services.btpanel_service.detect_bt_panel_installed",
        new_callable=AsyncMock,
    ) as detect:
        result = await svc._resolve_bt_installed(server, refresh=False)
    assert result is True
    detect.assert_not_called()


@pytest.mark.asyncio
async def test_resolve_bt_installed_ssh_when_pending():
    server = Server(
        id=2, name="s", domain="1.2.3.4",
        extra_attrs={"bt_panel": {"bootstrap_state": "pending", "auto_bootstrap": True}},
    )

    class _Repo:
        async def update(self, s):
            return s

    svc = BtPanelService(db=None)  # type: ignore[arg-type]
    svc.servers = _Repo()  # type: ignore[assignment]

    with patch(
        "server.application.services.btpanel_service.detect_bt_panel_installed",
        new_callable=AsyncMock,
        return_value=False,
    ) as detect:
        result = await svc._resolve_bt_installed(server, refresh=False)
    assert result is False
    detect.assert_awaited_once()


@pytest.mark.asyncio
async def test_bootstrap_lock_serializes_same_server(monkeypatch):
    redis = _FakeRedis()
    monkeypatch.setattr("server.infrastructure.btpanel.bootstrap_lock.get_redis", lambda: redis)
    order: list[str] = []

    async def worker(tag: str) -> None:
        async with bootstrap_lock(99):
            order.append(f"{tag}-in")
            await asyncio.sleep(0.05)
            order.append(f"{tag}-out")

    await asyncio.gather(worker("a"), worker("b"))
    # One worker fully completes before the other enters
    assert order.index("a-in") < order.index("a-out")
    assert order.index("b-in") < order.index("b-out")
    assert (order[1] == "a-out" and order[2] == "b-in") or (order[1] == "b-out" and order[2] == "a-in")