""" 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()