zhi
cf8a43d5b2
- Add api_key field to MonitoredServer model
- Add migration to create api_key column with unique index
- Add /admin/servers/{id}/api-key endpoint for key generation
- Add /admin/servers/{id}/api-key DELETE endpoint for revocation
- Add /server/heartbeat-v2 endpoint with X-API-Key header authentication
- Add TelemetryPayload model with extended fields (load_avg, uptime_seconds)
- Add basic unit tests for API key functionality
100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
"""Tests for Monitor API Key authentication and heartbeat v2"""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
# Test helper functions
|
|
def test_api_key_generation():
|
|
"""Test that API key is generated with correct format"""
|
|
import secrets
|
|
api_key = secrets.token_urlsafe(32)
|
|
# Should be around 43 chars (base64 encoded 32 bytes)
|
|
assert len(api_key) >= 40
|
|
assert len(api_key) <= 50
|
|
|
|
|
|
def test_monitored_server_model_has_api_key():
|
|
"""Test that MonitoredServer model has api_key field"""
|
|
from app.models.monitor import MonitoredServer
|
|
# Check the model has the api_key column defined
|
|
columns = [c.name for c in MonitoredServer.__table__.columns]
|
|
assert 'api_key' in columns
|
|
|
|
|
|
def test_api_key_endpoint_exists():
|
|
"""Test that /admin/servers/{id}/api-key endpoint is defined"""
|
|
from app.api.routers.monitor import router
|
|
# Check endpoint is in routes
|
|
paths = [r.path for r in router.routes]
|
|
assert any('/admin/servers/{server_id}/api-key' in p for p in paths)
|
|
|
|
|
|
def test_heartbeat_v2_endpoint_exists():
|
|
"""Test that /server/heartbeat-v2 endpoint is defined"""
|
|
from app.api.routers.monitor import router
|
|
paths = [r.path for r in router.routes]
|
|
assert any('/server/heartbeat-v2' in p for p in paths)
|
|
|
|
|
|
def test_telemetry_payload_model():
|
|
"""Test TelemetryPayload model fields"""
|
|
from app.api.routers.monitor import TelemetryPayload
|
|
payload = TelemetryPayload(
|
|
identifier='test-server',
|
|
openclaw_version='1.0.0',
|
|
agents=[{'name': 'test', 'status': 'active'}],
|
|
cpu_pct=50.0,
|
|
mem_pct=60.0,
|
|
disk_pct=70.0,
|
|
load_avg=[1.0, 2.0, 3.0],
|
|
uptime_seconds=3600
|
|
)
|
|
assert payload.identifier == 'test-server'
|
|
assert payload.openclaw_version == '1.0.0'
|
|
assert len(payload.agents) == 1
|
|
assert payload.load_avg == [1.0, 2.0, 3.0]
|
|
|
|
|
|
def test_heartbeat_v2_requires_api_key_header():
|
|
"""Test that heartbeat-v2 requires X-API-Key header"""
|
|
from app.api.routers.monitor import server_heartbeat_v2
|
|
# The function signature should require x_api_key parameter
|
|
import inspect
|
|
sig = inspect.signature(server_heartbeat_v2)
|
|
params = list(sig.parameters.keys())
|
|
assert 'x_api_key' in params
|
|
|
|
|
|
# Integration test placeholders (require running database)
|
|
class TestAPIKeyIntegration:
|
|
"""Integration tests - require database and running server"""
|
|
|
|
@pytest.mark.skip(reason="Requires running server")
|
|
def test_generate_api_key(self):
|
|
"""Generate API key for a server"""
|
|
# POST /admin/servers/{id}/api-key
|
|
# Should return api_key
|
|
pass
|
|
|
|
@pytest.mark.skip(reason="Requires running server")
|
|
def test_heartbeat_v2_with_valid_key(self):
|
|
"""Send heartbeat with valid API key"""
|
|
# POST /server/heartbeat-v2 with X-API-Key header
|
|
# Should return ok: true
|
|
pass
|
|
|
|
@pytest.mark.skip(reason="Requires running server")
|
|
def test_heartbeat_v2_with_invalid_key(self):
|
|
"""Send heartbeat with invalid API key"""
|
|
# POST /server/heartbeat-v2 with invalid X-API-Key
|
|
# Should return 401
|
|
pass
|
|
|
|
@pytest.mark.skip(reason="Requires running server")
|
|
def test_revoke_api_key(self):
|
|
"""Revoke API key"""
|
|
# DELETE /admin/servers/{id}/api-key
|
|
# Should return 204
|
|
pass |