roof
2f5370bd98
Unit Tests / test (push) Successful in 11m24s
These files were created during Steps 1-4 of the services architecture but were never staged: AccountManager (per-service credential provisioning), ServiceComposer (docker-compose lifecycle), built-in service manifests for email/calendar/files, and their test suites (158 tests). Also un-tracks .coverage binaries that were accidentally committed. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
348 lines
13 KiB
Python
348 lines
13 KiB
Python
"""
|
|
Unit tests for ServiceRegistry.
|
|
|
|
Tests load actual built-in manifests from api/services/builtins/ and verify
|
|
that the registry merges config correctly, returns expected routes/backup plans,
|
|
and handles missing manifests gracefully.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
|
|
|
|
from service_registry import ServiceRegistry, _BUILTINS_DIR
|
|
|
|
|
|
def _make_cm(configs: dict = None) -> MagicMock:
|
|
cm = MagicMock()
|
|
cm.configs = configs or {}
|
|
cm.get_installed_services.return_value = {}
|
|
return cm
|
|
|
|
|
|
class TestBuiltinManifests(unittest.TestCase):
|
|
"""Verify the built-in manifest files are valid JSON with required fields."""
|
|
|
|
def _load(self, service_id: str) -> dict:
|
|
path = os.path.join(_BUILTINS_DIR, service_id, 'manifest.json')
|
|
self.assertTrue(os.path.exists(path), f'Missing manifest for {service_id}')
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
def _assert_required(self, manifest: dict):
|
|
for field in ('schema_version', 'id', 'name', 'kind', 'capabilities'):
|
|
self.assertIn(field, manifest, f'Missing required field: {field}')
|
|
caps = manifest['capabilities']
|
|
for cap in ('has_subdomain', 'has_accounts', 'has_admin_config',
|
|
'has_storage', 'has_egress', 'has_api_hooks'):
|
|
self.assertIn(cap, caps, f'Missing capability flag: {cap}')
|
|
|
|
def test_email_manifest_valid(self):
|
|
m = self._load('email')
|
|
self._assert_required(m)
|
|
self.assertEqual(m['id'], 'email')
|
|
self.assertEqual(m['kind'], 'builtin')
|
|
self.assertIn('mail', [m.get('subdomain')] + (m.get('extra_subdomains') or []))
|
|
self.assertIn('webmail', m.get('extra_subdomains', []))
|
|
self.assertEqual(m['capabilities']['has_accounts'], True)
|
|
|
|
def test_calendar_manifest_valid(self):
|
|
m = self._load('calendar')
|
|
self._assert_required(m)
|
|
self.assertEqual(m['id'], 'calendar')
|
|
self.assertEqual(m['subdomain'], 'calendar')
|
|
|
|
def test_files_manifest_valid(self):
|
|
m = self._load('files')
|
|
self._assert_required(m)
|
|
self.assertEqual(m['id'], 'files')
|
|
self.assertIn('webdav', m.get('extra_subdomains', []))
|
|
|
|
def test_all_builtins_have_backup_volumes(self):
|
|
for svc_id in ('email', 'calendar', 'files'):
|
|
m = self._load(svc_id)
|
|
volumes = m.get('backup', {}).get('volumes')
|
|
self.assertTrue(volumes, f'{svc_id}: backup.volumes must not be empty')
|
|
for vol in volumes:
|
|
for field in ('container', 'path', 'name'):
|
|
self.assertIn(field, vol,
|
|
f'{svc_id}: backup volume entry missing {field!r}')
|
|
|
|
def test_all_builtins_have_peer_config_template(self):
|
|
for svc_id in ('email', 'calendar', 'files'):
|
|
m = self._load(svc_id)
|
|
self.assertTrue(m.get('peer_config_template'),
|
|
f'{svc_id}: peer_config_template must not be empty')
|
|
|
|
def test_config_schema_defaults_are_correct_types(self):
|
|
for svc_id in ('email', 'calendar', 'files'):
|
|
m = self._load(svc_id)
|
|
for field, spec in (m.get('config_schema') or {}).items():
|
|
if 'default' in spec:
|
|
if spec['type'] == 'integer':
|
|
self.assertIsInstance(
|
|
spec['default'], int,
|
|
f'{svc_id}.{field}: integer default must be int')
|
|
elif spec['type'] == 'string':
|
|
self.assertIsInstance(
|
|
spec['default'], str,
|
|
f'{svc_id}.{field}: string default must be str')
|
|
|
|
|
|
class TestServiceRegistryListAll(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.cm = _make_cm()
|
|
self.registry = ServiceRegistry(self.cm)
|
|
|
|
def test_lists_three_builtins(self):
|
|
services = self.registry.list_all()
|
|
ids = [s['id'] for s in services]
|
|
self.assertIn('email', ids)
|
|
self.assertIn('calendar', ids)
|
|
self.assertIn('files', ids)
|
|
|
|
def test_builtins_come_before_store_services(self):
|
|
self.cm.get_installed_services.return_value = {
|
|
'zstore': {'manifest': {
|
|
'id': 'zstore', 'name': 'Z Store', 'kind': 'store',
|
|
'capabilities': {}, 'config_schema': {}
|
|
}}
|
|
}
|
|
services = self.registry.list_all()
|
|
ids = [s['id'] for s in services]
|
|
# builtins (email, calendar, files) should all appear before zstore
|
|
for builtin_id in ('email', 'calendar', 'files'):
|
|
self.assertLess(ids.index(builtin_id), ids.index('zstore'))
|
|
|
|
def test_each_service_has_config_key(self):
|
|
for svc in self.registry.list_all():
|
|
self.assertIn('config', svc, f'{svc["id"]} missing config key')
|
|
|
|
def test_no_duplicate_ids(self):
|
|
services = self.registry.list_all()
|
|
ids = [s['id'] for s in services]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
|
|
|
|
class TestServiceRegistryConfigMerge(unittest.TestCase):
|
|
|
|
def test_defaults_used_when_no_saved_config(self):
|
|
cm = _make_cm({'calendar': {}})
|
|
reg = ServiceRegistry(cm)
|
|
svc = reg.get('calendar')
|
|
self.assertEqual(svc['config']['port'], 5232)
|
|
|
|
def test_saved_config_overrides_defaults(self):
|
|
cm = _make_cm({'calendar': {'port': 9999}})
|
|
reg = ServiceRegistry(cm)
|
|
svc = reg.get('calendar')
|
|
self.assertEqual(svc['config']['port'], 9999)
|
|
|
|
def test_unknown_saved_keys_excluded(self):
|
|
cm = _make_cm({'calendar': {'port': 5232, 'unknown_field': 'x'}})
|
|
reg = ServiceRegistry(cm)
|
|
svc = reg.get('calendar')
|
|
self.assertNotIn('unknown_field', svc['config'])
|
|
|
|
def test_partial_override_keeps_other_defaults(self):
|
|
cm = _make_cm({'email': {'smtp_port': 2525}})
|
|
reg = ServiceRegistry(cm)
|
|
svc = reg.get('email')
|
|
self.assertEqual(svc['config']['smtp_port'], 2525)
|
|
self.assertEqual(svc['config']['imap_port'], 993)
|
|
|
|
|
|
class TestServiceRegistryGet(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.cm = _make_cm()
|
|
self.registry = ServiceRegistry(self.cm)
|
|
|
|
def test_returns_none_for_unknown_id(self):
|
|
self.assertIsNone(self.registry.get('nonexistent_service'))
|
|
|
|
def test_returns_builtin_by_id(self):
|
|
svc = self.registry.get('email')
|
|
self.assertIsNotNone(svc)
|
|
self.assertEqual(svc['id'], 'email')
|
|
|
|
def test_returns_store_service_from_installed(self):
|
|
self.cm.get_installed_services.return_value = {
|
|
'mywiki': {'manifest': {
|
|
'id': 'mywiki', 'name': 'Wiki', 'kind': 'store',
|
|
'capabilities': {}, 'config_schema': {}
|
|
}}
|
|
}
|
|
svc = self.registry.get('mywiki')
|
|
self.assertIsNotNone(svc)
|
|
self.assertEqual(svc['id'], 'mywiki')
|
|
|
|
|
|
class TestServiceRegistryGetCaddyRoutes(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.cm = _make_cm()
|
|
self.registry = ServiceRegistry(self.cm)
|
|
|
|
def test_all_builtins_appear_in_routes(self):
|
|
routes = self.registry.get_caddy_routes()
|
|
route_ids = [r['service_id'] for r in routes]
|
|
for svc_id in ('email', 'calendar', 'files'):
|
|
self.assertIn(svc_id, route_ids)
|
|
|
|
def test_email_route_has_webmail_extra_subdomain(self):
|
|
routes = self.registry.get_caddy_routes()
|
|
email_route = next(r for r in routes if r['service_id'] == 'email')
|
|
self.assertIn('webmail', email_route['extra_subdomains'])
|
|
|
|
def test_files_route_has_webdav_extra_subdomain(self):
|
|
routes = self.registry.get_caddy_routes()
|
|
files_route = next(r for r in routes if r['service_id'] == 'files')
|
|
self.assertIn('webdav', files_route['extra_subdomains'])
|
|
|
|
def test_services_without_subdomain_excluded(self):
|
|
self.cm.get_installed_services.return_value = {
|
|
'nosubdomain': {'manifest': {
|
|
'id': 'nosubdomain', 'name': 'NoSub', 'kind': 'store',
|
|
'capabilities': {'has_subdomain': False},
|
|
'config_schema': {}
|
|
}}
|
|
}
|
|
routes = self.registry.get_caddy_routes()
|
|
self.assertNotIn('nosubdomain', [r['service_id'] for r in routes])
|
|
|
|
|
|
class TestServiceRegistryGetBackupPlan(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.cm = _make_cm()
|
|
self.registry = ServiceRegistry(self.cm)
|
|
|
|
def test_all_builtins_in_backup_plan(self):
|
|
plan = self.registry.get_backup_plan()
|
|
plan_ids = [p['service_id'] for p in plan]
|
|
for svc_id in ('email', 'calendar', 'files'):
|
|
self.assertIn(svc_id, plan_ids)
|
|
|
|
def test_email_backup_includes_maildata_volume(self):
|
|
plan = self.registry.get_backup_plan()
|
|
email_plan = next(p for p in plan if p['service_id'] == 'email')
|
|
names = [v['name'] for v in email_plan['volumes']]
|
|
self.assertIn('maildata', names)
|
|
vol = next(v for v in email_plan['volumes'] if v['name'] == 'maildata')
|
|
self.assertEqual(vol['container'], 'cell-mail')
|
|
self.assertEqual(vol['path'], '/var/mail')
|
|
|
|
def test_calendar_backup_includes_radicale_volume(self):
|
|
plan = self.registry.get_backup_plan()
|
|
cal_plan = next(p for p in plan if p['service_id'] == 'calendar')
|
|
names = [v['name'] for v in cal_plan['volumes']]
|
|
self.assertIn('radicale_data', names)
|
|
vol = next(v for v in cal_plan['volumes'] if v['name'] == 'radicale_data')
|
|
self.assertEqual(vol['container'], 'cell-radicale')
|
|
self.assertEqual(vol['path'], '/data')
|
|
|
|
def test_files_backup_includes_both_volumes(self):
|
|
plan = self.registry.get_backup_plan()
|
|
files_plan = next(p for p in plan if p['service_id'] == 'files')
|
|
names = {v['name'] for v in files_plan['volumes']}
|
|
self.assertIn('filegator', names)
|
|
self.assertIn('files', names)
|
|
|
|
def test_service_without_storage_excluded(self):
|
|
self.cm.get_installed_services.return_value = {
|
|
'nostorage': {'manifest': {
|
|
'id': 'nostorage', 'name': 'NoStorage', 'kind': 'store',
|
|
'capabilities': {'has_storage': False},
|
|
'config_schema': {}
|
|
}}
|
|
}
|
|
plan = self.registry.get_backup_plan()
|
|
self.assertNotIn('nostorage', [p['service_id'] for p in plan])
|
|
|
|
|
|
class TestServiceRegistryGetPeerServiceInfo(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.cm = _make_cm({'calendar': {}})
|
|
self.registry = ServiceRegistry(self.cm)
|
|
|
|
def test_fills_domain_placeholder(self):
|
|
info = self.registry.get_peer_service_info(
|
|
'calendar', 'alice', 'example.com', {})
|
|
self.assertIn('example.com', info['caldav_url'])
|
|
|
|
def test_fills_peer_username(self):
|
|
info = self.registry.get_peer_service_info(
|
|
'calendar', 'bob', 'example.com', {})
|
|
self.assertIn('bob', info['caldav_url'])
|
|
|
|
def test_fills_credentials(self):
|
|
info = self.registry.get_peer_service_info(
|
|
'calendar', 'alice', 'example.com', {'password': 'secret123'})
|
|
self.assertEqual(info['password'], 'secret123')
|
|
|
|
def test_returns_none_for_unknown_service(self):
|
|
result = self.registry.get_peer_service_info(
|
|
'unknown_svc', 'alice', 'example.com', {})
|
|
self.assertIsNone(result)
|
|
|
|
def test_username_url_encoded_in_peer_url(self):
|
|
info = self.registry.get_peer_service_info(
|
|
'calendar', 'alice/../../etc', 'example.com', {})
|
|
self.assertNotIn('../', info['caldav_url'])
|
|
self.assertIn('alice%2F', info['caldav_url'])
|
|
|
|
def test_domain_not_altered_by_username(self):
|
|
info = self.registry.get_peer_service_info(
|
|
'calendar', 'alice@evil.com', 'legit.example.com', {})
|
|
self.assertIn('legit.example.com', info['caldav_url'])
|
|
|
|
|
|
class TestServiceRegistryConfigMergeTypeCoercion(unittest.TestCase):
|
|
|
|
def test_string_in_config_coerced_to_int(self):
|
|
cm = _make_cm({'calendar': {'port': '9999'}})
|
|
reg = ServiceRegistry(cm)
|
|
svc = reg.get('calendar')
|
|
self.assertIsInstance(svc['config']['port'], int)
|
|
self.assertEqual(svc['config']['port'], 9999)
|
|
|
|
def test_unconvertible_value_falls_back_to_default(self):
|
|
cm = _make_cm({'calendar': {'port': 'not_a_number'}})
|
|
reg = ServiceRegistry(cm)
|
|
svc = reg.get('calendar')
|
|
self.assertEqual(svc['config']['port'], 5232)
|
|
|
|
|
|
class TestServiceRegistryWithBrokenManifest(unittest.TestCase):
|
|
"""Registry must not crash when a manifest file is corrupt or missing."""
|
|
|
|
def test_missing_builtins_dir_returns_empty(self):
|
|
with patch('service_registry._BUILTINS_DIR', '/nonexistent/path'):
|
|
reg = ServiceRegistry(_make_cm())
|
|
self.assertEqual(reg.list_all(), [])
|
|
|
|
def test_malformed_json_manifest_skipped(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
bad_dir = os.path.join(tmpdir, 'bad_svc')
|
|
os.makedirs(bad_dir)
|
|
with open(os.path.join(bad_dir, 'manifest.json'), 'w') as f:
|
|
f.write('this is not json {{{')
|
|
with patch('service_registry._BUILTINS_DIR', tmpdir):
|
|
reg = ServiceRegistry(_make_cm())
|
|
# Should not raise; just return empty list
|
|
result = reg.list_all()
|
|
self.assertEqual(result, [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|