Add monitor API tests (B8)
- test_monitor.py: 12 test cases covering: - API key generation (success, 404, admin-only) - heartbeat-v2 endpoint (valid key, invalid key, missing key) - API key revocation and validation - ServerState data persistence (agents_json, cpu_pct, etc.) - Disabled server rejection - conftest.py: import app.models.monitor for test database setup All tests passing (pytest -v).
This commit is contained in:
322
tests/test_monitor.py
Normal file
322
tests/test_monitor.py
Normal file
@@ -0,0 +1,322 @@
|
||||
"""Tests for monitor API — API Key and heartbeat-v2 endpoints."""
|
||||
import json
|
||||
import pytest
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture()
|
||||
def make_server(db):
|
||||
"""Factory to create a MonitoredServer row."""
|
||||
from app.models.monitor import MonitoredServer
|
||||
|
||||
_counter = [0]
|
||||
|
||||
def _make(identifier=None, display_name=None, is_enabled=True, api_key=None):
|
||||
_counter[0] += 1
|
||||
n = _counter[0]
|
||||
s = MonitoredServer(
|
||||
identifier=identifier or f"test-server-{n}",
|
||||
display_name=display_name or f"Test Server {n}",
|
||||
is_enabled=is_enabled,
|
||||
api_key=api_key,
|
||||
)
|
||||
db.add(s)
|
||||
db.commit()
|
||||
db.refresh(s)
|
||||
return s
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def admin_auth(make_user, auth_header):
|
||||
"""Create admin user and return auth header."""
|
||||
user = make_user(is_admin=True)
|
||||
return auth_header(user), user
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# API Key Generation Tests (POST /admin/servers/{id}/api-key)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_generate_api_key_success(client, db, make_server, admin_auth):
|
||||
"""B8-1: api-key 生成成功返回 200 + api_key"""
|
||||
headers, _ = admin_auth
|
||||
server = make_server()
|
||||
|
||||
response = client.post(f"/monitor/admin/servers/{server.id}/api-key", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "api_key" in data
|
||||
assert data["server_id"] == server.id
|
||||
assert len(data["api_key"]) > 20 # token_urlsafe(32) produces ~43 chars
|
||||
|
||||
# Verify key is stored in DB
|
||||
db.refresh(server)
|
||||
assert server.api_key == data["api_key"]
|
||||
|
||||
|
||||
def test_generate_api_key_not_found(client, db, admin_auth):
|
||||
"""B8-2: api-key 生成对不存在 server 返回 404"""
|
||||
headers, _ = admin_auth
|
||||
|
||||
response = client.post("/monitor/admin/servers/99999/api-key", headers=headers)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert "not found" in response.json()["detail"].lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Heartbeat v2 Tests (POST /server/heartbeat-v2)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_heartbeat_v2_valid_key(client, db, make_server):
|
||||
"""B8-3: heartbeat-v2 带有效 key 返回 200"""
|
||||
api_key = "test-api-key-valid-123"
|
||||
server = make_server(api_key=api_key)
|
||||
|
||||
payload = {
|
||||
"identifier": server.identifier,
|
||||
"openclaw_version": "1.0.0",
|
||||
"agents": [{"id": "agent1", "status": "idle"}],
|
||||
"cpu_pct": 45.5,
|
||||
"mem_pct": 60.0,
|
||||
"disk_pct": 30.0,
|
||||
"swap_pct": 10.0,
|
||||
"load_avg": [0.5, 0.6, 0.7],
|
||||
"uptime_seconds": 3600,
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/monitor/server/heartbeat-v2",
|
||||
json=payload,
|
||||
headers={"X-API-Key": api_key},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["ok"] is True
|
||||
assert data["server_id"] == server.id
|
||||
assert data["identifier"] == server.identifier
|
||||
|
||||
|
||||
def test_heartbeat_v2_invalid_key(client, db, make_server):
|
||||
"""B8-4: heartbeat-v2 带无效 key 返回 401"""
|
||||
server = make_server(api_key="valid-key")
|
||||
|
||||
payload = {
|
||||
"identifier": server.identifier,
|
||||
"cpu_pct": 45.5,
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/monitor/server/heartbeat-v2",
|
||||
json=payload,
|
||||
headers={"X-API-Key": "invalid-key"},
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "invalid" in response.json()["detail"].lower() or "api key" in response.json()["detail"].lower()
|
||||
|
||||
|
||||
def test_heartbeat_v2_missing_key(client, db, make_server):
|
||||
"""B8-5: heartbeat-v2 不带 key 返回 401"""
|
||||
server = make_server(api_key="valid-key")
|
||||
|
||||
payload = {"identifier": server.identifier}
|
||||
|
||||
response = client.post("/monitor/server/heartbeat-v2", json=payload)
|
||||
|
||||
# FastAPI will reject missing required header with 422 (validation error)
|
||||
# or we may expect 401 depending on implementation
|
||||
assert response.status_code in [401, 422]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Revoke API Key Tests (DELETE /admin/servers/{id}/api-key)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_revoke_api_key_makes_key_invalid(client, db, make_server, admin_auth):
|
||||
"""B8-6: 测试 revoke 后 key 失效"""
|
||||
headers, _ = admin_auth
|
||||
api_key = "test-key-to-revoke"
|
||||
server = make_server(api_key=api_key)
|
||||
|
||||
# First, verify key works
|
||||
payload = {"identifier": server.identifier, "cpu_pct": 50.0}
|
||||
response1 = client.post(
|
||||
"/monitor/server/heartbeat-v2",
|
||||
json=payload,
|
||||
headers={"X-API-Key": api_key},
|
||||
)
|
||||
assert response1.status_code == 200
|
||||
|
||||
# Revoke the key
|
||||
response2 = client.delete(f"/monitor/admin/servers/{server.id}/api-key", headers=headers)
|
||||
assert response2.status_code == 204
|
||||
|
||||
# Verify key no longer works
|
||||
response3 = client.post(
|
||||
"/monitor/server/heartbeat-v2",
|
||||
json=payload,
|
||||
headers={"X-API-Key": api_key},
|
||||
)
|
||||
assert response3.status_code == 401
|
||||
|
||||
# Verify DB
|
||||
db.refresh(server)
|
||||
assert server.api_key is None
|
||||
|
||||
|
||||
def test_revoke_api_key_not_found(client, db, admin_auth):
|
||||
"""Revoke API key for non-existent server returns 404"""
|
||||
headers, _ = admin_auth
|
||||
|
||||
response = client.delete("/monitor/admin/servers/99999/api-key", headers=headers)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Data Storage Tests (ServerState)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_heartbeat_v2_stores_data_correctly(client, db, make_server):
|
||||
"""B8-7: 测试数据是否正确写入 ServerState（agents_json, cpu_pct, etc.）"""
|
||||
from app.models.monitor import ServerState
|
||||
|
||||
api_key = "test-api-key-storage"
|
||||
server = make_server(api_key=api_key)
|
||||
|
||||
agents = [{"id": "agent1", "name": "TestAgent", "status": "busy"}]
|
||||
payload = {
|
||||
"identifier": server.identifier,
|
||||
"openclaw_version": "2.5.1",
|
||||
"agents": agents,
|
||||
"cpu_pct": 75.5,
|
||||
"mem_pct": 82.3,
|
||||
"disk_pct": 45.0,
|
||||
"swap_pct": 5.5,
|
||||
"load_avg": [1.2, 1.5, 1.8],
|
||||
"uptime_seconds": 7200,
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/monitor/server/heartbeat-v2",
|
||||
json=payload,
|
||||
headers={"X-API-Key": api_key},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify ServerState in DB
|
||||
state = db.query(ServerState).filter(ServerState.server_id == server.id).first()
|
||||
assert state is not None
|
||||
assert state.openclaw_version == "2.5.1"
|
||||
assert state.cpu_pct == 75.5
|
||||
assert state.mem_pct == 82.3
|
||||
assert state.disk_pct == 45.0
|
||||
assert state.swap_pct == 5.5
|
||||
assert state.last_seen_at is not None
|
||||
|
||||
# Verify agents_json
|
||||
stored_agents = json.loads(state.agents_json)
|
||||
assert stored_agents == agents
|
||||
assert stored_agents[0]["id"] == "agent1"
|
||||
assert stored_agents[0]["status"] == "busy"
|
||||
|
||||
|
||||
def test_heartbeat_v2_updates_existing_state(client, db, make_server):
|
||||
"""Test that subsequent heartbeats update the existing ServerState"""
|
||||
from app.models.monitor import ServerState
|
||||
|
||||
api_key = "test-api-key-update"
|
||||
server = make_server(api_key=api_key)
|
||||
|
||||
# First heartbeat
|
||||
response1 = client.post(
|
||||
"/monitor/server/heartbeat-v2",
|
||||
json={
|
||||
"identifier": server.identifier,
|
||||
"cpu_pct": 30.0,
|
||||
"mem_pct": 40.0,
|
||||
"agents": [{"id": "a1"}],
|
||||
},
|
||||
headers={"X-API-Key": api_key},
|
||||
)
|
||||
assert response1.status_code == 200
|
||||
|
||||
state1 = db.query(ServerState).filter(ServerState.server_id == server.id).first()
|
||||
first_seen = state1.last_seen_at
|
||||
|
||||
# Second heartbeat with different data
|
||||
response2 = client.post(
|
||||
"/monitor/server/heartbeat-v2",
|
||||
json={
|
||||
"identifier": server.identifier,
|
||||
"cpu_pct": 60.0,
|
||||
"mem_pct": 70.0,
|
||||
"agents": [{"id": "a1"}, {"id": "a2"}],
|
||||
},
|
||||
headers={"X-API-Key": api_key},
|
||||
)
|
||||
assert response2.status_code == 200
|
||||
|
||||
# Verify update
|
||||
db.refresh(state1)
|
||||
assert state1.cpu_pct == 60.0
|
||||
assert state1.mem_pct == 70.0
|
||||
assert len(json.loads(state1.agents_json)) == 2
|
||||
# last_seen_at should be updated (newer timestamp)
|
||||
assert state1.last_seen_at >= first_seen
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Authorization Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_generate_api_key_requires_admin(client, make_user, auth_header, make_server):
|
||||
"""Only admin can generate API keys"""
|
||||
non_admin = make_user(is_admin=False)
|
||||
server = make_server()
|
||||
|
||||
response = client.post(
|
||||
f"/monitor/admin/servers/{server.id}/api-key",
|
||||
headers=auth_header(non_admin),
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_revoke_api_key_requires_admin(client, make_user, auth_header, make_server):
|
||||
"""Only admin can revoke API keys"""
|
||||
non_admin = make_user(is_admin=False)
|
||||
server = make_server(api_key="some-key")
|
||||
|
||||
response = client.delete(
|
||||
f"/monitor/admin/servers/{server.id}/api-key",
|
||||
headers=auth_header(non_admin),
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Disabled Server Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_heartbeat_v2_disabled_server(client, db, make_server):
|
||||
"""Heartbeat-v2 should reject disabled servers even with valid key"""
|
||||
api_key = "test-key-disabled"
|
||||
server = make_server(api_key=api_key, is_enabled=False)
|
||||
|
||||
response = client.post(
|
||||
"/monitor/server/heartbeat-v2",
|
||||
json={"identifier": server.identifier, "cpu_pct": 50.0},
|
||||
headers={"X-API-Key": api_key},
|
||||
)
|
||||
|
||||
assert response.status_code == 401 # Should be treated as invalid key
|
||||
Reference in New Issue
Block a user