mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-25 04:28:00 +09:00
Juan Denis
0e29978e7b
Telemetry: ensure sanitized payloads bigger than the 32KiB threshold are replaced with a {_truncated: True} marker before storage, and add a unit test covering that behavior.
Backend API: adjust imports in secrets_webhooks to use WorkspaceService and include User model.
Frontend: make Sidebar resilient to malformed active_workspace JSON in localStorage (try/catch), and ensure socketService.disconnect() is called when user becomes unauthenticated in NotificationsProvider to clear notifications and unread count.
Files: backend/services/telemetry_service.py, backend/tests/test_telemetry.py, backend/app/api/secrets_webhooks.py, frontend/src/components/Sidebar.jsx, frontend/src/contexts/NotificationsContext.jsx
214 lines
9.1 KiB
Python
214 lines
9.1 KiB
Python
"""Tests for the unified telemetry / system event stream."""
|
|
from datetime import datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from app import db
|
|
from app.models.system_event import SystemEvent
|
|
from app.services.telemetry_service import TelemetryService, generate_correlation_id
|
|
|
|
|
|
class TestTelemetryService:
|
|
def test_emit_event_creates_row(self, app):
|
|
event = TelemetryService.emit(
|
|
source='test',
|
|
event_type='test.event',
|
|
message='Hello telemetry',
|
|
severity='info',
|
|
resource_type='app',
|
|
resource_id=42,
|
|
payload={'hello': 'world'},
|
|
commit=True,
|
|
)
|
|
assert event is not None
|
|
assert event.id is not None
|
|
assert event.source == 'test'
|
|
assert event.event_type == 'test.event'
|
|
assert event.message == 'Hello telemetry'
|
|
assert event.resource_type == 'app'
|
|
assert event.resource_id == '42'
|
|
assert event.get_payload() == {'hello': 'world'}
|
|
|
|
def test_emit_redacts_sensitive_payload_keys(self, app):
|
|
event = TelemetryService.emit(
|
|
source='test',
|
|
event_type='test.secret',
|
|
payload={
|
|
'username': 'admin',
|
|
'api_key': 'super-secret',
|
|
'nested': {'password': 'hunter2'},
|
|
},
|
|
commit=True,
|
|
)
|
|
payload = event.get_payload()
|
|
assert payload['username'] == 'admin'
|
|
assert payload['api_key'] == '[redacted]'
|
|
assert payload['nested']['password'] == '[redacted]'
|
|
|
|
def test_emit_invalid_severity_defaults_to_info(self, app):
|
|
event = TelemetryService.emit(
|
|
source='test',
|
|
event_type='test.event',
|
|
severity='not-a-severity',
|
|
commit=True,
|
|
)
|
|
assert event.severity == 'info'
|
|
|
|
def test_emit_truncates_oversized_payload(self, app):
|
|
# After sanitization each string is capped at MAX_STRING_LENGTH and each
|
|
# dict at MAX_DICT_ITEMS, so many large entries still exceed 32 KiB.
|
|
large_payload = {f'key_{i}': 'x' * 2000 for i in range(50)}
|
|
event = TelemetryService.emit(
|
|
source='test',
|
|
event_type='test.large',
|
|
payload=large_payload,
|
|
commit=True,
|
|
)
|
|
payload = event.get_payload()
|
|
assert payload == {'_truncated': True}
|
|
|
|
def test_emit_does_not_raise_on_failure(self, app, monkeypatch):
|
|
monkeypatch.setattr(db.session, 'commit', lambda: (_ for _ in ()).throw(Exception('boom')))
|
|
event = TelemetryService.emit(
|
|
source='test',
|
|
event_type='test.event',
|
|
commit=True,
|
|
)
|
|
assert event is None
|
|
|
|
def test_get_events_filter_by_source_and_severity(self, app):
|
|
TelemetryService.emit(source='audit', event_type='audit.login', severity='info', commit=True)
|
|
TelemetryService.emit(source='backup', event_type='backup.failed', severity='error', commit=True)
|
|
TelemetryService.emit(source='backup', event_type='backup.completed', severity='info', commit=True)
|
|
|
|
result = TelemetryService.get_events(source='backup', severity='error')
|
|
assert result.total == 1
|
|
assert result.items[0].event_type == 'backup.failed'
|
|
|
|
def test_get_events_by_correlation_id(self, app):
|
|
correlation_id = generate_correlation_id()
|
|
TelemetryService.emit(source='backup', event_type='backup.started', correlation_id=correlation_id, commit=True)
|
|
TelemetryService.emit(source='backup', event_type='backup.completed', correlation_id=correlation_id, commit=True)
|
|
TelemetryService.emit(source='backup', event_type='backup.started', commit=True)
|
|
|
|
events = TelemetryService.get_events_by_correlation(correlation_id)
|
|
assert len(events) == 2
|
|
|
|
def test_get_stats(self, app):
|
|
TelemetryService.emit(source='backup', event_type='backup.completed', severity='info', commit=True)
|
|
TelemetryService.emit(source='backup', event_type='backup.failed', severity='error', commit=True)
|
|
TelemetryService.emit(source='monitoring', event_type='monitoring.alert', severity='warning', commit=True)
|
|
|
|
stats = TelemetryService.get_stats(hours=24)
|
|
assert stats['total'] == 3
|
|
assert stats['by_severity']['error'] == 1
|
|
assert stats['by_severity']['warning'] == 1
|
|
assert stats['by_severity']['info'] == 1
|
|
assert stats['by_source']['backup'] == 2
|
|
assert stats['by_source']['monitoring'] == 1
|
|
|
|
def test_cleanup_old_events(self, app):
|
|
old = TelemetryService.emit(source='test', event_type='test.old', commit=True)
|
|
old.timestamp = datetime.utcnow() - timedelta(days=100)
|
|
db.session.commit()
|
|
|
|
TelemetryService.emit(source='test', event_type='test.new', commit=True)
|
|
|
|
deleted = TelemetryService.cleanup_old_events(days=90)
|
|
assert deleted == 1
|
|
assert SystemEvent.query.filter_by(event_type='test.old').first() is None
|
|
assert SystemEvent.query.filter_by(event_type='test.new').first() is not None
|
|
|
|
|
|
class TestTelemetryAPI:
|
|
def test_list_events_requires_auth(self, client):
|
|
response = client.get('/api/v1/telemetry/events')
|
|
assert response.status_code == 401
|
|
|
|
def test_list_events(self, app, client, auth_headers):
|
|
TelemetryService.emit(source='audit', event_type='audit.login', severity='info', commit=True)
|
|
TelemetryService.emit(source='backup', event_type='backup.failed', severity='error', commit=True)
|
|
|
|
response = client.get('/api/v1/telemetry/events', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['total'] == 2
|
|
assert len(data['events']) == 2
|
|
|
|
def test_list_events_filter_by_source(self, app, client, auth_headers):
|
|
TelemetryService.emit(source='audit', event_type='audit.login', commit=True)
|
|
TelemetryService.emit(source='backup', event_type='backup.failed', commit=True)
|
|
|
|
response = client.get('/api/v1/telemetry/events?source=backup', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['total'] == 1
|
|
assert data['events'][0]['source'] == 'backup'
|
|
|
|
def test_get_event(self, app, client, auth_headers):
|
|
event = TelemetryService.emit(source='test', event_type='test.event', commit=True)
|
|
response = client.get(f'/api/v1/telemetry/events/{event.id}', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['id'] == event.id
|
|
assert data['source'] == 'test'
|
|
|
|
def test_get_event_not_found(self, client, auth_headers):
|
|
response = client.get('/api/v1/telemetry/events/99999', headers=auth_headers)
|
|
assert response.status_code == 404
|
|
|
|
def test_get_events_by_correlation(self, app, client, auth_headers):
|
|
correlation_id = generate_correlation_id()
|
|
TelemetryService.emit(source='backup', event_type='backup.started', correlation_id=correlation_id, commit=True)
|
|
TelemetryService.emit(source='backup', event_type='backup.completed', correlation_id=correlation_id, commit=True)
|
|
|
|
response = client.get(f'/api/v1/telemetry/events/by-correlation/{correlation_id}', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert len(data['events']) == 2
|
|
|
|
def test_get_stats_endpoint(self, app, client, auth_headers):
|
|
TelemetryService.emit(source='backup', event_type='backup.completed', commit=True)
|
|
response = client.get('/api/v1/telemetry/stats', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['total'] >= 1
|
|
assert 'by_severity' in data
|
|
assert 'by_source' in data
|
|
|
|
def test_cleanup_events_admin_only(self, app, client, auth_headers):
|
|
response = client.delete('/api/v1/telemetry/events?days=1', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
def test_emit_test_event_admin_only(self, app, client, auth_headers):
|
|
response = client.post('/api/v1/telemetry/events/test', json={'message': 'test'}, headers=auth_headers)
|
|
assert response.status_code == 201
|
|
data = response.get_json()
|
|
assert data['event_type'] == 'telemetry.test'
|
|
|
|
|
|
class TestTelemetryIntegrations:
|
|
def test_audit_service_emits_telemetry_event(self, app):
|
|
from app.services.audit_service import AuditService
|
|
from app.models import User
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
user = User(email='audit@test.local', username='audituser', password_hash=generate_password_hash('x'))
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
|
|
AuditService.log(
|
|
action='app.create',
|
|
user_id=user.id,
|
|
target_type='app',
|
|
target_id=123,
|
|
details={'name': 'my-app'},
|
|
commit=True,
|
|
)
|
|
|
|
event = SystemEvent.query.filter_by(source='audit').first()
|
|
assert event is not None
|
|
assert event.event_type == 'audit.action_logged'
|
|
assert event.actor_user_id == user.id
|
|
assert event.get_payload()['details']['name'] == 'my-app'
|