Nexus Agent
08a0157c95
调度页执行周期可视化/单次执行/分类选机/推送源对齐；登录 IP 门控与无缝导航； 服务器批量后台任务、执行记录、凭据合并、各页激活刷新与错误提示修复。 Co-authored-by: Cursor <cursoragent@cursor.com>
51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
"""Auto-generate per-server agent_api_key when missing (batch/single install)."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from server.application.server_batch_common import ensure_agent_api_key, generate_agent_api_key_value
|
|
from server.domain.models import Server
|
|
|
|
|
|
def test_generate_agent_api_key_format():
|
|
key = generate_agent_api_key_value()
|
|
assert key.startswith("nxs-")
|
|
assert len(key) > 20
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_agent_api_key_returns_existing():
|
|
server = Server(id=1, name="s1", agent_api_key="nxs-existing")
|
|
assert await ensure_agent_api_key(server) == "nxs-existing"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_agent_api_key_generates_and_persists():
|
|
server = Server(id=42, name="机器人", agent_api_key=None)
|
|
row = Server(id=42, name="机器人", agent_api_key=None)
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_by_id = AsyncMock(return_value=row)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.commit = AsyncMock()
|
|
|
|
mock_ctx = MagicMock()
|
|
mock_ctx.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch(
|
|
"server.infrastructure.database.session.AsyncSessionLocal",
|
|
return_value=mock_ctx,
|
|
), patch(
|
|
"server.infrastructure.database.server_repo.ServerRepositoryImpl",
|
|
return_value=mock_repo,
|
|
):
|
|
key = await ensure_agent_api_key(server)
|
|
|
|
assert key.startswith("nxs-")
|
|
assert row.agent_api_key == key
|
|
assert server.agent_api_key == key
|
|
mock_session.commit.assert_awaited_once()
|