Nexus Agent
b46922fe92
宝塔式自选子机监测：Pin CRUD、全指标 SSH/Redis 探针、探针落库、 /ws/watch 推送、监测历史与进程抽屉；Agent 可选附带 net/disk IO。 Co-authored-by: Cursor <cursoragent@cursor.com>
88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
"""Watch pin API integration tests."""
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock
|
|
|
|
import bcrypt
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from server.domain.models import Admin, Server
|
|
from server.utils.watch_metrics import WATCH_MAX_SLOTS
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_watch_redis(monkeypatch):
|
|
mock_redis = AsyncMock()
|
|
mock_redis.lindex = AsyncMock(return_value=None)
|
|
mock_redis.lrange = AsyncMock(return_value=[])
|
|
mock_redis.get = AsyncMock(return_value=None)
|
|
monkeypatch.setattr("server.application.services.watch_service.get_redis", lambda: mock_redis)
|
|
return mock_redis
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_watch_pin_crud(db_session, test_admin, mock_watch_redis):
|
|
from server.application.services.watch_service import SlotsFullError, WatchService
|
|
|
|
for i in range(WATCH_MAX_SLOTS):
|
|
s = Server(name=f"watch-{i}", domain=f"w{i}.example.com", port=22, agent_version="1.0")
|
|
db_session.add(s)
|
|
await db_session.commit()
|
|
servers = (await db_session.execute(select(Server))).scalars().all()
|
|
|
|
svc = WatchService(db_session)
|
|
admin = test_admin["admin"]
|
|
|
|
r0 = await svc.pin_server(admin=admin, server_id=servers[0].id)
|
|
assert len(r0["slots"]) == 4
|
|
assert r0["slots"][0]["empty"] is False
|
|
|
|
await svc.pin_server(admin=admin, server_id=servers[0].id)
|
|
|
|
for s in servers[1:]:
|
|
await svc.pin_server(admin=admin, server_id=s.id)
|
|
|
|
with pytest.raises(SlotsFullError):
|
|
extra = Server(name="extra", domain="extra.example.com", port=22, agent_version="1.0")
|
|
db_session.add(extra)
|
|
await db_session.flush()
|
|
await svc.pin_server(admin=admin, server_id=extra.id)
|
|
|
|
await svc.remove_slot(admin=admin, slot_index=0)
|
|
listing = await svc.list_slots(admin.id)
|
|
assert listing["slots"][0]["empty"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_watch_probe_record_on_failure(db_session):
|
|
from server.infrastructure.database.watch_repo import WatchRepositoryImpl
|
|
from server.utils.watch_metrics import failure_sample
|
|
|
|
repo = WatchRepositoryImpl(db_session)
|
|
admin = Admin(username="watch_probe_admin", password_hash=bcrypt.hashpw(b"x", bcrypt.gensalt()).decode(), is_active=True)
|
|
db_session.add(admin)
|
|
server = Server(name="probe-fail", domain="pf.example.com", port=22)
|
|
db_session.add(server)
|
|
await db_session.flush()
|
|
|
|
created = await repo.create_pin(admin_id=admin.id, server_id=server.id, slot_index=0)
|
|
sample = failure_sample(status="ssh_timeout", source="ssh", error="timeout")
|
|
row = await repo.insert_probe_record(
|
|
session_id=created["session_id"],
|
|
server_id=server.id,
|
|
admin_id=admin.id,
|
|
recorded_at=datetime.now(timezone.utc).replace(tzinfo=None),
|
|
source=sample.source,
|
|
probe_status=sample.probe_status,
|
|
duration_ms=8000,
|
|
is_online=False,
|
|
cpu_pct=None,
|
|
mem_pct=None,
|
|
disk_pct=None,
|
|
error=sample.error,
|
|
)
|
|
await db_session.commit()
|
|
assert row.probe_status == "ssh_timeout"
|
|
assert row.cpu_pct is None
|