r
82426b1941
新增侧栏「宝塔面板」菜单、代理 API 与连接配置页；通过 SSH 读写子机 api.json 自动写入凭据，5 分钟后台重试，含审计修复、检测缓存与并发锁。 Co-authored-by: Cursor <cursoragent@cursor.com>
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""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
|
|
|
|
|
|
@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():
|
|
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")
|