roof
13074f56cb
Unit Tests / test (push) Successful in 12m34s
Root causes fixed:
- Dead LOG_LEVEL globals() lookup pinned root logger at INFO regardless of
PIC_LOG_LEVEL env or config; replaced with _resolve_root_log_level() +
apply_root_log_level() which sets both root logger and all attached handlers
at startup and on runtime re-apply.
- set_service_level() only set the named 'pic.<service>' logger; bare module
loggers (e.g. 'caddy_manager') were never reached, so per-service log files
stayed 0 bytes. Fixed via _SERVICE_MODULE_LOGGERS map covering all managers.
- Log viewer GET /api/logs had no level filter; added ?level= query param.
- Per-service log levels lived in an out-of-band config/api/log_levels.json
side-file with no validation; migrated into ConfigManager under a new
'logging' section ({python:{root,services}, containers:{caddy,coredns,
wireguard,mailserver,api}}) with get/set helpers, invalid-level rejection,
and one-time migration from the old file on first load.
New capabilities:
- Container log levels: Caddy (injects global log { level X } + hot reload),
CoreDNS (DEBUG enables log plugin, else errors-only), WireGuard/mailserver
via pending_restart path.
- PUT /api/logs/verbosity accepts {python, containers} dict; returns per-entry
applied:hot|pending_restart status.
- Webui Logs page gains two-section Verbosity tab (Python services + Container
services) with needs-restart badges.
- managers.py wires per-service loggers before manager instantiation and
re-applies persisted levels from ConfigManager; legacy log_levels.json read
removed.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1051 lines
45 KiB
Python
1051 lines
45 KiB
Python
"""
|
|
Tests for routes/services.py:
|
|
- /api/services/catalog (list, get, status, restart, reconfigure)
|
|
- /api/services/catalog/<id>/accounts (list, provision, deprovision, credentials)
|
|
- /api/services/bus/* (status, events, start, stop, restart)
|
|
- /api/logs/* endpoints
|
|
"""
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
|
|
|
|
import app as app_module
|
|
from app import app
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
app.config['TESTING'] = True
|
|
with app.test_client() as c:
|
|
yield c
|
|
|
|
|
|
def _make_registry(services):
|
|
reg = MagicMock()
|
|
reg.list_all = MagicMock(return_value=services)
|
|
reg.list_active = MagicMock(return_value=services)
|
|
reg.get = MagicMock(side_effect=lambda sid: next(
|
|
(s for s in services if s['id'] == sid), None))
|
|
return reg
|
|
|
|
|
|
def _make_service(sid, kind='builtin'):
|
|
return {'id': sid, 'name': sid.title(), 'kind': kind,
|
|
'subdomain': sid, 'capabilities': {}}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/catalog
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetServicesCatalog:
|
|
def test_returns_200(self, client):
|
|
reg = _make_registry([])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_services_list(self, client):
|
|
reg = _make_registry([_make_service('email')])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog')
|
|
data = json.loads(resp.data)
|
|
assert 'services' in data
|
|
assert len(data['services']) == 1
|
|
|
|
def test_returns_empty_when_no_services(self, client):
|
|
reg = _make_registry([])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog')
|
|
data = json.loads(resp.data)
|
|
assert data['services'] == []
|
|
|
|
def test_500_on_exception(self, client):
|
|
reg = MagicMock()
|
|
reg.list_all.side_effect = Exception('registry error')
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/catalog/<service_id>
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetServiceCatalogEntry:
|
|
def test_returns_200_for_known_service(self, client):
|
|
reg = _make_registry([_make_service('email')])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog/email')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_service_data(self, client):
|
|
reg = _make_registry([_make_service('email')])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog/email')
|
|
data = json.loads(resp.data)
|
|
assert data['id'] == 'email'
|
|
|
|
def test_returns_404_for_unknown_service(self, client):
|
|
reg = _make_registry([])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog/nonexistent')
|
|
assert resp.status_code == 404
|
|
|
|
def test_404_includes_error_message(self, client):
|
|
reg = _make_registry([])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog/nonexistent')
|
|
data = json.loads(resp.data)
|
|
assert 'error' in data
|
|
|
|
def test_500_on_exception(self, client):
|
|
reg = MagicMock()
|
|
reg.get.side_effect = Exception('registry error')
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/catalog/email')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/catalog/<service_id>/status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetServiceContainerStatus:
|
|
def test_returns_200(self, client):
|
|
svc = _make_service('email')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
composer.status_service.return_value = {'running': True, 'containers': []}
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.get('/api/services/catalog/email/status')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_404_for_unknown_service(self, client):
|
|
reg = _make_registry([])
|
|
composer = MagicMock()
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.get('/api/services/catalog/nonexistent/status')
|
|
assert resp.status_code == 404
|
|
|
|
def test_400_on_value_error(self, client):
|
|
svc = _make_service('email')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
composer.status_service.side_effect = ValueError('bad service')
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.get('/api/services/catalog/email/status')
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/services/catalog/<service_id>/restart
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRestartServiceContainers:
|
|
def test_returns_200_on_success(self, client):
|
|
svc = _make_service('email')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
composer.restart_service.return_value = {'ok': True, 'stdout': ''}
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/email/restart')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_404_for_unknown_service(self, client):
|
|
reg = _make_registry([])
|
|
composer = MagicMock()
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/nonexistent/restart')
|
|
assert resp.status_code == 404
|
|
|
|
def test_returns_500_when_restart_fails(self, client):
|
|
svc = _make_service('email')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
composer.restart_service.return_value = {'ok': False, 'stderr': 'container error'}
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/email/restart')
|
|
assert resp.status_code == 500
|
|
|
|
def test_response_includes_message_on_success(self, client):
|
|
svc = _make_service('email')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
composer.restart_service.return_value = {'ok': True}
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/email/restart')
|
|
data = json.loads(resp.data)
|
|
assert 'message' in data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/services/catalog/<service_id>/reconfigure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestReconfigureService:
|
|
def test_404_for_unknown_service(self, client):
|
|
reg = _make_registry([])
|
|
composer = MagicMock()
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/nonexistent/reconfigure')
|
|
assert resp.status_code == 404
|
|
|
|
def test_400_for_builtin_service(self, client):
|
|
svc = _make_service('wireguard', kind='builtin')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/wireguard/reconfigure')
|
|
assert resp.status_code == 400
|
|
data = json.loads(resp.data)
|
|
assert 'Builtins' in data['error']
|
|
|
|
def test_400_when_no_compose_file(self, client):
|
|
svc = _make_service('myapp', kind='store')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
composer.has_compose_file.return_value = False
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/myapp/reconfigure')
|
|
assert resp.status_code == 400
|
|
|
|
def test_200_on_success(self, client):
|
|
svc = _make_service('myapp', kind='store')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
composer.has_compose_file.return_value = True
|
|
composer.up.return_value = {'ok': True, 'stdout': ''}
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/myapp/reconfigure')
|
|
assert resp.status_code == 200
|
|
|
|
def test_500_when_up_fails(self, client):
|
|
svc = _make_service('myapp', kind='store')
|
|
reg = _make_registry([svc])
|
|
composer = MagicMock()
|
|
composer.has_compose_file.return_value = True
|
|
composer.up.return_value = {'ok': False, 'stderr': 'error'}
|
|
with patch.object(app_module, 'service_registry', reg), \
|
|
patch.object(app_module, 'service_composer', composer):
|
|
resp = client.post('/api/services/catalog/myapp/reconfigure')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/catalog/<id>/accounts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListServiceAccounts:
|
|
def test_returns_200(self, client):
|
|
am = MagicMock()
|
|
am.list_accounts.return_value = ['alice', 'bob']
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.get('/api/services/catalog/email/accounts')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_accounts_list(self, client):
|
|
am = MagicMock()
|
|
am.list_accounts.return_value = ['alice']
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.get('/api/services/catalog/email/accounts')
|
|
data = json.loads(resp.data)
|
|
assert data['accounts'] == ['alice']
|
|
assert data['service_id'] == 'email'
|
|
|
|
def test_500_on_exception(self, client):
|
|
am = MagicMock()
|
|
am.list_accounts.side_effect = Exception('fail')
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.get('/api/services/catalog/email/accounts')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/services/catalog/<id>/accounts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestProvisionServiceAccount:
|
|
def test_returns_201_on_success(self, client):
|
|
am = MagicMock()
|
|
am.provision.return_value = None
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.post('/api/services/catalog/email/accounts',
|
|
data=json.dumps({'username': 'alice'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 201
|
|
|
|
def test_400_when_username_missing(self, client):
|
|
am = MagicMock()
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.post('/api/services/catalog/email/accounts',
|
|
data=json.dumps({}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 400
|
|
|
|
def test_400_on_value_error(self, client):
|
|
am = MagicMock()
|
|
am.provision.side_effect = ValueError('user already exists')
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.post('/api/services/catalog/email/accounts',
|
|
data=json.dumps({'username': 'alice'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 400
|
|
|
|
def test_500_on_runtime_error(self, client):
|
|
am = MagicMock()
|
|
am.provision.side_effect = RuntimeError('docker exec failed')
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.post('/api/services/catalog/email/accounts',
|
|
data=json.dumps({'username': 'alice'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 500
|
|
|
|
def test_response_shape(self, client):
|
|
am = MagicMock()
|
|
am.provision.return_value = None
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.post('/api/services/catalog/email/accounts',
|
|
data=json.dumps({'username': 'alice'}),
|
|
content_type='application/json')
|
|
data = json.loads(resp.data)
|
|
assert data['service_id'] == 'email'
|
|
assert data['username'] == 'alice'
|
|
assert data['provisioned'] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DELETE /api/services/catalog/<id>/accounts/<username>
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDeprovisionServiceAccount:
|
|
def test_returns_200_on_success(self, client):
|
|
am = MagicMock()
|
|
am.deprovision.return_value = True
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.delete('/api/services/catalog/email/accounts/alice')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_500_when_deprovision_fails(self, client):
|
|
am = MagicMock()
|
|
am.deprovision.return_value = False
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.delete('/api/services/catalog/email/accounts/alice')
|
|
assert resp.status_code == 500
|
|
|
|
def test_400_on_value_error(self, client):
|
|
am = MagicMock()
|
|
am.deprovision.side_effect = ValueError('not found')
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.delete('/api/services/catalog/email/accounts/alice')
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/catalog/<id>/accounts/<username>/credentials
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetServiceAccountCredentials:
|
|
def test_returns_200_with_creds(self, client):
|
|
am = MagicMock()
|
|
am.get_credentials.return_value = {'password': 'secret'}
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.get('/api/services/catalog/email/accounts/alice/credentials')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_404_when_not_provisioned(self, client):
|
|
am = MagicMock()
|
|
am.get_credentials.return_value = None
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.get('/api/services/catalog/email/accounts/alice/credentials')
|
|
assert resp.status_code == 404
|
|
|
|
def test_returns_creds_in_response(self, client):
|
|
am = MagicMock()
|
|
am.get_credentials.return_value = {'password': 'mypass'}
|
|
with patch.object(app_module, 'account_manager', am):
|
|
resp = client.get('/api/services/catalog/email/accounts/alice/credentials')
|
|
data = json.loads(resp.data)
|
|
assert data['service_id'] == 'email'
|
|
assert data['username'] == 'alice'
|
|
assert 'password' in data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/bus/status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServiceBusStatus:
|
|
def test_returns_200(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.get_service_status_summary.return_value = {}
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.get('/api/services/bus/status')
|
|
assert resp.status_code == 200
|
|
|
|
def test_500_on_exception(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.get_service_status_summary.side_effect = Exception('fail')
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.get('/api/services/bus/status')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/services/bus/services/<name>/start
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServiceBusStart:
|
|
def test_returns_200_on_success(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.orchestrate_service_start.return_value = True
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.post('/api/services/bus/services/network/start')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_500_on_failure(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.orchestrate_service_start.return_value = False
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.post('/api/services/bus/services/network/start')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/services/bus/services/<name>/stop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServiceBusStop:
|
|
def test_returns_200_on_success(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.orchestrate_service_stop.return_value = True
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.post('/api/services/bus/services/network/stop')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_500_on_failure(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.orchestrate_service_stop.return_value = False
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.post('/api/services/bus/services/network/stop')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/services/bus/services/<name>/restart
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServiceBusRestart:
|
|
def test_returns_200_on_success(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.orchestrate_service_restart.return_value = True
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.post('/api/services/bus/services/email/restart')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_500_on_failure(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.orchestrate_service_restart.return_value = False
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.post('/api/services/bus/services/email/restart')
|
|
assert resp.status_code == 500
|
|
|
|
def test_500_on_exception(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.orchestrate_service_restart.side_effect = Exception('bus error')
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.post('/api/services/bus/services/email/restart')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/bus/events
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServiceBusEvents:
|
|
def test_returns_200(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.get_event_history.return_value = []
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.get('/api/services/bus/events')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_list(self, client):
|
|
mock_sb = MagicMock()
|
|
mock_sb.get_event_history.return_value = []
|
|
with patch.object(app_module, 'service_bus', mock_sb):
|
|
resp = client.get('/api/services/bus/events')
|
|
data = json.loads(resp.data)
|
|
assert isinstance(data, list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/logs/services/<service>
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetServiceLogs:
|
|
def test_returns_200(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.get_service_logs.return_value = ['line1', 'line2']
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.get('/api/logs/services/network')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_log_lines(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.get_service_logs.return_value = ['line1']
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.get('/api/logs/services/network')
|
|
data = json.loads(resp.data)
|
|
assert 'logs' in data
|
|
assert data['service'] == 'network'
|
|
|
|
def test_500_on_exception(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.get_service_logs.side_effect = Exception('fail')
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.get('/api/logs/services/network')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/logs/search
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSearchLogs:
|
|
def test_returns_200(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.search_logs.return_value = []
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.post('/api/logs/search',
|
|
data=json.dumps({'query': 'error'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_results(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.search_logs.return_value = [{'message': 'error occurred'}]
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.post('/api/logs/search',
|
|
data=json.dumps({'query': 'error'}),
|
|
content_type='application/json')
|
|
data = json.loads(resp.data)
|
|
assert 'results' in data
|
|
assert 'count' in data
|
|
assert data['count'] == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/logs/statistics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestLogStatistics:
|
|
def test_returns_200(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.get_log_statistics.return_value = {'total': 100}
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.get('/api/logs/statistics')
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/logs/rotate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRotateLogs:
|
|
def test_returns_200(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.rotate_logs.return_value = None
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.post('/api/logs/rotate',
|
|
data=json.dumps({}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/logs/files
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetLogFiles:
|
|
def test_returns_200(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.get_all_log_file_infos.return_value = []
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.get('/api/logs/files')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_file_list(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.get_all_log_file_infos.return_value = [
|
|
{'file': 'network.log', 'size': 1024, 'backup': False}
|
|
]
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.get('/api/logs/files')
|
|
data = json.loads(resp.data)
|
|
assert len(data) == 1
|
|
assert data[0]['file'] == 'network.log'
|
|
|
|
def test_500_on_exception(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.get_all_log_file_infos.side_effect = Exception('disk error')
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.get('/api/logs/files')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/logs/verbosity
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetLogVerbosity:
|
|
def test_returns_200(self, client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_logging_config.return_value = {
|
|
'python': {'root': 'INFO', 'services': {'network': 'INFO'}},
|
|
'containers': {'caddy': 'INFO'},
|
|
}
|
|
with patch.object(app_module, 'config_manager', mock_cm):
|
|
resp = client.get('/api/logs/verbosity')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_python_and_container_sections(self, client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_logging_config.return_value = {
|
|
'python': {'root': 'INFO', 'services': {'network': 'DEBUG', 'email': 'INFO'}},
|
|
'containers': {'caddy': 'WARNING'},
|
|
}
|
|
with patch.object(app_module, 'config_manager', mock_cm):
|
|
resp = client.get('/api/logs/verbosity')
|
|
data = json.loads(resp.data)
|
|
assert data['python']['services']['network'] == 'DEBUG'
|
|
assert data['containers']['caddy'] == 'WARNING'
|
|
|
|
def test_500_on_exception(self, client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_logging_config.side_effect = Exception('fail')
|
|
with patch.object(app_module, 'config_manager', mock_cm):
|
|
resp = client.get('/api/logs/verbosity')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PUT /api/logs/verbosity
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSetLogVerbosity:
|
|
def test_returns_200(self, client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_logging_config.return_value = {
|
|
'python': {'root': 'INFO', 'services': {'network': 'DEBUG'}},
|
|
'containers': {},
|
|
}
|
|
with patch.object(app_module, 'config_manager', mock_cm), \
|
|
patch.object(app_module, 'log_manager', MagicMock()):
|
|
resp = client.put('/api/logs/verbosity',
|
|
data=json.dumps({'python': {'services': {'network': 'DEBUG'}}}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 200
|
|
|
|
def test_persists_via_config_manager_and_applies_hot(self, client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_logging_config.return_value = {'python': {}, 'containers': {}}
|
|
mock_lm = MagicMock()
|
|
with patch.object(app_module, 'config_manager', mock_cm), \
|
|
patch.object(app_module, 'log_manager', mock_lm):
|
|
client.put('/api/logs/verbosity',
|
|
data=json.dumps({'python': {'services': {'network': 'DEBUG'}}}),
|
|
content_type='application/json')
|
|
mock_cm.set_python_log_level.assert_called_with('network', 'DEBUG')
|
|
mock_lm.set_service_level.assert_called_with('network', 'DEBUG')
|
|
|
|
def test_500_on_exception(self, client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.set_python_log_level.side_effect = Exception('fail')
|
|
with patch.object(app_module, 'config_manager', mock_cm), \
|
|
patch.object(app_module, 'log_manager', MagicMock()):
|
|
resp = client.put('/api/logs/verbosity',
|
|
data=json.dumps({'python': {'services': {'network': 'DEBUG'}}}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/active
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetActiveServices:
|
|
def test_returns_200(self, client):
|
|
reg = _make_registry([_make_service('email')])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/active')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_active_services_list(self, client):
|
|
reg = _make_registry([
|
|
{'id': 'email', 'name': 'Email', 'subdomain': 'mail', 'capabilities': {'smtp': True}},
|
|
])
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/active')
|
|
data = json.loads(resp.data)
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
assert data[0]['id'] == 'email'
|
|
|
|
def test_500_on_exception(self, client):
|
|
reg = MagicMock()
|
|
reg.list_active.side_effect = Exception('error')
|
|
with patch.object(app_module, 'service_registry', reg):
|
|
resp = client.get('/api/services/active')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetAllServicesStatus:
|
|
def test_returns_200(self, client):
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.list_services.return_value = []
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/status')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_status_for_each_service(self, client):
|
|
mock_svc = MagicMock()
|
|
mock_svc.get_status.return_value = {
|
|
'status': 'online', 'running': True, 'timestamp': '2026-01-01T00:00:00'
|
|
}
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.list_services.return_value = ['network']
|
|
mock_sbus.get_service.return_value = mock_svc
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/status')
|
|
data = json.loads(resp.data)
|
|
assert 'network' in data
|
|
|
|
def test_service_exception_sets_error_in_status(self, client):
|
|
mock_svc = MagicMock()
|
|
mock_svc.get_status.side_effect = Exception('service down')
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.list_services.return_value = ['email']
|
|
mock_sbus.get_service.return_value = mock_svc
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/status')
|
|
data = json.loads(resp.data)
|
|
assert 'email' in data
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/connectivity
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTestAllServicesConnectivity:
|
|
def test_returns_200(self, client):
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.list_services.return_value = []
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/connectivity')
|
|
assert resp.status_code == 200
|
|
|
|
def test_includes_connectivity_result(self, client):
|
|
mock_svc = MagicMock()
|
|
mock_svc.test_connectivity.return_value = {'success': True}
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.list_services.return_value = ['network']
|
|
mock_sbus.get_service.return_value = mock_svc
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/connectivity')
|
|
data = json.loads(resp.data)
|
|
assert 'network' in data
|
|
|
|
def test_500_on_bus_exception(self, client):
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.list_services.side_effect = Exception('bus fail')
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/connectivity')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/logs (backend logs endpoint)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetBackendLogs:
|
|
def test_returns_404_when_log_file_missing(self, client):
|
|
# The default state: picell.log does not exist in the api/ directory
|
|
import os
|
|
import routes.services as svc_routes
|
|
# If the log file doesn't exist, it should return 404
|
|
original = os.path.exists
|
|
def fake_exists(path):
|
|
if 'picell.log' in path:
|
|
return False
|
|
return original(path)
|
|
with patch('routes.services.os.path.exists', side_effect=fake_exists):
|
|
resp = client.get('/api/logs')
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/logs/export
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestExportLogsEndpoint:
|
|
def test_returns_200(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.export_logs.return_value = '[]'
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.post('/api/logs/export',
|
|
data=json.dumps({'format': 'json'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_logs_and_format(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.export_logs.return_value = '[{"level":"INFO"}]'
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.post('/api/logs/export',
|
|
data=json.dumps({'format': 'json'}),
|
|
content_type='application/json')
|
|
data = json.loads(resp.data)
|
|
assert 'logs' in data
|
|
assert data['format'] == 'json'
|
|
|
|
def test_500_on_exception(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.export_logs.side_effect = Exception('export failed')
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.post('/api/logs/export',
|
|
data=json.dumps({'format': 'json'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/services/status — service-specific branches (wireguard, email, etc.)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetAllServicesStatusServiceBranches:
|
|
"""Cover the elif branches in get_all_services_status for each service type."""
|
|
|
|
def _status_for_service(self, client, service_name, raw_status):
|
|
mock_svc = MagicMock()
|
|
mock_svc.get_status.return_value = dict(
|
|
{'status': 'online', 'running': True, 'timestamp': '2026-01-01T00:00:00'},
|
|
**raw_status
|
|
)
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.list_services.return_value = [service_name]
|
|
mock_sbus.get_service.return_value = mock_svc
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/status')
|
|
return json.loads(resp.data)
|
|
|
|
def test_wireguard_branch_includes_peers_count(self, client):
|
|
data = self._status_for_service(client, 'wireguard',
|
|
{'peers_count': 3, 'interface': 'wg0'})
|
|
assert data['wireguard'].get('peers_count') == 3
|
|
assert data['wireguard'].get('interface') == 'wg0'
|
|
|
|
def test_email_branch_includes_users_count(self, client):
|
|
data = self._status_for_service(client, 'email',
|
|
{'users_count': 5, 'domain': 'cell.local'})
|
|
assert data['email'].get('users_count') == 5
|
|
assert data['email'].get('domain') == 'cell.local'
|
|
|
|
def test_calendar_branch_includes_calendars_count(self, client):
|
|
data = self._status_for_service(client, 'calendar',
|
|
{'users_count': 2, 'calendars_count': 4})
|
|
assert data['calendar'].get('calendars_count') == 4
|
|
|
|
def test_files_branch_includes_storage_used(self, client):
|
|
data = self._status_for_service(client, 'files',
|
|
{'users_count': 1, 'total_storage_used': {'used': 100}})
|
|
assert data['files'].get('storage_used') == {'used': 100}
|
|
|
|
def test_routing_branch_includes_nat_rules_count(self, client):
|
|
data = self._status_for_service(client, 'routing',
|
|
{'nat_rules_count': 2, 'peer_routes_count': 1,
|
|
'firewall_rules_count': 3})
|
|
assert data['routing'].get('nat_rules_count') == 2
|
|
|
|
def test_vault_branch_includes_certificates_count(self, client):
|
|
data = self._status_for_service(client, 'vault',
|
|
{'certificates_count': 5, 'trusted_keys_count': 2})
|
|
assert data['vault'].get('certificates_count') == 5
|
|
|
|
def test_non_dict_status_uses_string_fallback(self, client):
|
|
"""When get_status returns a non-dict, it gets stored as a string+bool."""
|
|
mock_svc = MagicMock()
|
|
mock_svc.get_status.return_value = 'running'
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.list_services.return_value = ['network']
|
|
mock_sbus.get_service.return_value = mock_svc
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/status')
|
|
data = json.loads(resp.data)
|
|
assert 'network' in data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Error paths in catalog service management endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServiceCatalogErrorPaths:
|
|
"""Cover ValueError/Exception branches in catalog management endpoints."""
|
|
|
|
def test_get_service_container_status_value_error_returns_400(self, client):
|
|
mock_reg = MagicMock()
|
|
mock_reg.get.return_value = {'id': 'email', 'kind': 'builtin'}
|
|
mock_composer = MagicMock()
|
|
mock_composer.status_service.side_effect = ValueError('bad request')
|
|
with patch.object(app_module, 'service_registry', mock_reg), \
|
|
patch.object(app_module, 'service_composer', mock_composer):
|
|
resp = client.get('/api/services/catalog/email/status')
|
|
assert resp.status_code == 400
|
|
|
|
def test_restart_service_containers_value_error_returns_400(self, client):
|
|
mock_reg = MagicMock()
|
|
mock_reg.get.return_value = {'id': 'email', 'kind': 'builtin'}
|
|
mock_composer = MagicMock()
|
|
mock_composer.restart_service.side_effect = ValueError('bad service')
|
|
with patch.object(app_module, 'service_registry', mock_reg), \
|
|
patch.object(app_module, 'service_composer', mock_composer):
|
|
resp = client.post('/api/services/catalog/email/restart')
|
|
assert resp.status_code == 400
|
|
|
|
def test_restart_service_containers_exception_returns_500(self, client):
|
|
mock_reg = MagicMock()
|
|
mock_reg.get.return_value = {'id': 'email', 'kind': 'builtin'}
|
|
mock_composer = MagicMock()
|
|
mock_composer.restart_service.side_effect = Exception('docker down')
|
|
with patch.object(app_module, 'service_registry', mock_reg), \
|
|
patch.object(app_module, 'service_composer', mock_composer):
|
|
resp = client.post('/api/services/catalog/email/restart')
|
|
assert resp.status_code == 500
|
|
|
|
def test_reconfigure_service_value_error_returns_400(self, client):
|
|
mock_reg = MagicMock()
|
|
mock_reg.get.return_value = {'id': 'myapp', 'kind': 'store'}
|
|
mock_composer = MagicMock()
|
|
mock_composer.has_compose_file.return_value = True
|
|
mock_composer.up.side_effect = ValueError('invalid config')
|
|
with patch.object(app_module, 'service_registry', mock_reg), \
|
|
patch.object(app_module, 'service_composer', mock_composer):
|
|
resp = client.post('/api/services/catalog/myapp/reconfigure')
|
|
assert resp.status_code == 400
|
|
|
|
def test_reconfigure_service_exception_returns_500(self, client):
|
|
mock_reg = MagicMock()
|
|
mock_reg.get.return_value = {'id': 'myapp', 'kind': 'store'}
|
|
mock_composer = MagicMock()
|
|
mock_composer.has_compose_file.return_value = True
|
|
mock_composer.up.side_effect = Exception('compose fail')
|
|
with patch.object(app_module, 'service_registry', mock_reg), \
|
|
patch.object(app_module, 'service_composer', mock_composer):
|
|
resp = client.post('/api/services/catalog/myapp/reconfigure')
|
|
assert resp.status_code == 500
|
|
|
|
def test_provision_service_account_exception_returns_500(self, client):
|
|
mock_am = MagicMock()
|
|
mock_am.provision.side_effect = Exception('db failure')
|
|
with patch.object(app_module, 'account_manager', mock_am):
|
|
resp = client.post('/api/services/catalog/email/accounts',
|
|
data=json.dumps({'username': 'alice'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 500
|
|
|
|
def test_deprovision_service_account_value_error_returns_400(self, client):
|
|
mock_am = MagicMock()
|
|
mock_am.deprovision.side_effect = ValueError('user not found')
|
|
with patch.object(app_module, 'account_manager', mock_am):
|
|
resp = client.delete('/api/services/catalog/email/accounts/alice')
|
|
assert resp.status_code == 400
|
|
|
|
def test_deprovision_service_account_exception_returns_500(self, client):
|
|
mock_am = MagicMock()
|
|
mock_am.deprovision.side_effect = Exception('db down')
|
|
with patch.object(app_module, 'account_manager', mock_am):
|
|
resp = client.delete('/api/services/catalog/email/accounts/alice')
|
|
assert resp.status_code == 500
|
|
|
|
def test_get_service_account_credentials_exception_returns_500(self, client):
|
|
mock_am = MagicMock()
|
|
mock_am.get_credentials.side_effect = Exception('db fail')
|
|
with patch.object(app_module, 'account_manager', mock_am):
|
|
resp = client.get('/api/services/catalog/email/accounts/alice/credentials')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Exception paths for bus event/start/stop and log endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServiceBusEndpointExceptions:
|
|
"""Cover exception paths for service bus and log endpoints."""
|
|
|
|
def test_get_service_bus_events_exception_returns_500(self, client):
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.get_event_history.side_effect = Exception('bus crash')
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.get('/api/services/bus/events')
|
|
assert resp.status_code == 500
|
|
|
|
def test_start_service_exception_returns_500(self, client):
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.orchestrate_service_start.side_effect = Exception('crash')
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.post('/api/services/bus/services/email/start')
|
|
assert resp.status_code == 500
|
|
|
|
def test_stop_service_exception_returns_500(self, client):
|
|
mock_sbus = MagicMock()
|
|
mock_sbus.orchestrate_service_stop.side_effect = Exception('crash')
|
|
with patch.object(app_module, 'service_bus', mock_sbus):
|
|
resp = client.post('/api/services/bus/services/email/stop')
|
|
assert resp.status_code == 500
|
|
|
|
def test_search_logs_exception_returns_500(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.search_logs.side_effect = Exception('search fail')
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.post('/api/logs/search',
|
|
data='{}',
|
|
content_type='application/json')
|
|
assert resp.status_code == 500
|
|
|
|
def test_get_log_statistics_exception_returns_500(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.get_log_statistics.side_effect = Exception('stats fail')
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.get('/api/logs/statistics')
|
|
assert resp.status_code == 500
|
|
|
|
def test_rotate_logs_exception_returns_500(self, client):
|
|
mock_lm = MagicMock()
|
|
mock_lm.rotate_logs.side_effect = Exception('rotate fail')
|
|
with patch.object(app_module, 'log_manager', mock_lm):
|
|
resp = client.post('/api/logs/rotate',
|
|
data='{}',
|
|
content_type='application/json')
|
|
assert resp.status_code == 500
|