""" Unit tests for ServiceRegistry. Tests verify that the registry merges config correctly from installed store services, returns expected routes/backup plans, and handles missing or broken records gracefully. There are no builtins — only installed store services. """ import sys import unittest from pathlib import Path from unittest.mock import MagicMock sys.path.insert(0, str(Path(__file__).parent.parent / 'api')) from service_registry import ServiceRegistry # --------------------------------------------------------------------------- # Shared manifests used across multiple test classes # --------------------------------------------------------------------------- _CALENDAR_MANIFEST = { 'schema_version': 1, 'id': 'calendar', 'name': 'Calendar', 'kind': 'store', 'capabilities': { 'has_subdomain': True, 'has_accounts': False, 'has_admin_config': True, 'has_storage': True, 'has_egress': False, 'has_api_hooks': False, }, 'subdomain': 'calendar', 'backend': 'cell-radicale:5232', 'extra_subdomains': [], 'extra_backends': {}, 'config_schema': { 'port': {'type': 'integer', 'default': 5232, 'label': 'Port'}, }, 'peer_config_template': { 'caldav_url': 'https://calendar.{domain}/radicale/{peer.username}/', 'password': '{peer.service_credentials.calendar.password}', }, 'backup': { 'volumes': [ {'container': 'cell-radicale', 'path': '/data', 'name': 'radicale_data'} ] }, } _EMAIL_MANIFEST = { 'schema_version': 1, 'id': 'email', 'name': 'Email', 'kind': 'store', 'capabilities': { 'has_subdomain': True, 'has_accounts': True, 'has_admin_config': True, 'has_storage': True, 'has_egress': True, 'has_api_hooks': False, }, 'subdomain': 'mail', 'backend': 'cell-rainloop:8888', 'extra_subdomains': ['webmail'], 'extra_backends': {}, 'config_schema': { 'smtp_port': {'type': 'integer', 'default': 587, 'label': 'SMTP Port'}, 'imap_port': {'type': 'integer', 'default': 993, 'label': 'IMAP Port'}, }, 'peer_config_template': { 'smtp_server': 'mail.{domain}', 'imap_server': 'mail.{domain}', 'password': '{peer.service_credentials.email.password}', }, 'backup': { 'volumes': [ {'container': 'cell-mail', 'path': '/var/mail', 'name': 'maildata'}, ] }, } _FILES_MANIFEST = { 'schema_version': 1, 'id': 'files', 'name': 'Files', 'kind': 'store', 'capabilities': { 'has_subdomain': True, 'has_accounts': False, 'has_admin_config': False, 'has_storage': True, 'has_egress': False, 'has_api_hooks': False, }, 'subdomain': 'files', 'backend': 'cell-filegator:8080', 'extra_subdomains': ['webdav'], 'extra_backends': {'webdav': 'cell-webdav:80'}, 'config_schema': {}, 'peer_config_template': { 'files_url': 'https://files.{domain}/', }, 'backup': { 'volumes': [ {'container': 'cell-filegator', 'path': '/data', 'name': 'filegator'}, {'container': 'cell-webdav', 'path': '/data', 'name': 'files'}, ] }, } def _make_cm(configs: dict = None, installed: dict = None) -> MagicMock: cm = MagicMock() cm.configs = configs or {} cm.get_installed_services.return_value = installed or {} return cm # --------------------------------------------------------------------------- # TestServiceRegistryListAll # --------------------------------------------------------------------------- class TestServiceRegistryListAll(unittest.TestCase): def test_list_all_empty_when_nothing_installed(self): cm = _make_cm() reg = ServiceRegistry(cm) self.assertEqual(reg.list_all(), []) def test_list_all_returns_installed_services(self): cm = _make_cm(installed={'calendar': {'manifest': _CALENDAR_MANIFEST}}) reg = ServiceRegistry(cm) ids = [s['id'] for s in reg.list_all()] self.assertIn('calendar', ids) def test_each_service_has_config_key(self): cm = _make_cm(installed={ 'calendar': {'manifest': _CALENDAR_MANIFEST}, 'email': {'manifest': _EMAIL_MANIFEST}, }) reg = ServiceRegistry(cm) for svc in reg.list_all(): self.assertIn('config', svc, f'{svc["id"]} missing config key') def test_no_duplicate_ids(self): cm = _make_cm(installed={ 'calendar': {'manifest': _CALENDAR_MANIFEST}, 'email': {'manifest': _EMAIL_MANIFEST}, }) reg = ServiceRegistry(cm) ids = [s['id'] for s in reg.list_all()] self.assertEqual(len(ids), len(set(ids))) # --------------------------------------------------------------------------- # TestServiceRegistryConfigMerge # --------------------------------------------------------------------------- class TestServiceRegistryConfigMerge(unittest.TestCase): def test_defaults_used_when_no_saved_config(self): cm = _make_cm( configs={'calendar': {}}, installed={'calendar': {'manifest': _CALENDAR_MANIFEST}}, ) reg = ServiceRegistry(cm) svc = reg.get('calendar') self.assertEqual(svc['config']['port'], 5232) def test_saved_config_overrides_defaults(self): cm = _make_cm( configs={'calendar': {'port': 9999}}, installed={'calendar': {'manifest': _CALENDAR_MANIFEST}}, ) reg = ServiceRegistry(cm) svc = reg.get('calendar') self.assertEqual(svc['config']['port'], 9999) def test_unknown_saved_keys_excluded(self): cm = _make_cm( configs={'calendar': {'port': 5232, 'unknown_field': 'x'}}, installed={'calendar': {'manifest': _CALENDAR_MANIFEST}}, ) reg = ServiceRegistry(cm) svc = reg.get('calendar') self.assertNotIn('unknown_field', svc['config']) def test_partial_override_keeps_other_defaults(self): cm = _make_cm( configs={'email': {'smtp_port': 2525}}, installed={'email': {'manifest': _EMAIL_MANIFEST}}, ) reg = ServiceRegistry(cm) svc = reg.get('email') self.assertEqual(svc['config']['smtp_port'], 2525) self.assertEqual(svc['config']['imap_port'], 993) # --------------------------------------------------------------------------- # TestServiceRegistryGet # --------------------------------------------------------------------------- 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_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') def test_get_returns_none_when_installed_record_has_no_manifest(self): self.cm.get_installed_services.return_value = { 'broken': {} # record exists but has no 'manifest' key } self.assertIsNone(self.registry.get('broken')) # --------------------------------------------------------------------------- # TestServiceRegistryGetCaddyRoutes # --------------------------------------------------------------------------- class TestServiceRegistryGetCaddyRoutes(unittest.TestCase): def setUp(self): self.cm = _make_cm() self.registry = ServiceRegistry(self.cm) 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]) def test_installed_service_with_subdomain_appears_in_routes(self): self.cm.get_installed_services.return_value = { 'calendar': {'manifest': _CALENDAR_MANIFEST}, } routes = self.registry.get_caddy_routes() route_ids = [r['service_id'] for r in routes] self.assertIn('calendar', route_ids) cal_route = next(r for r in routes if r['service_id'] == 'calendar') self.assertEqual(cal_route['subdomain'], 'calendar') self.assertEqual(cal_route['backend'], 'cell-radicale:5232') # --------------------------------------------------------------------------- # TestServiceRegistryGetBackupPlan # --------------------------------------------------------------------------- class TestServiceRegistryGetBackupPlan(unittest.TestCase): def setUp(self): self.cm = _make_cm() self.registry = ServiceRegistry(self.cm) 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]) def test_installed_service_with_storage_in_backup_plan(self): self.cm.get_installed_services.return_value = { 'calendar': {'manifest': _CALENDAR_MANIFEST}, } plan = self.registry.get_backup_plan() plan_ids = [p['service_id'] for p in plan] self.assertIn('calendar', plan_ids) 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) # --------------------------------------------------------------------------- # TestServiceRegistryGetPeerServiceInfo # --------------------------------------------------------------------------- class TestServiceRegistryGetPeerServiceInfo(unittest.TestCase): def setUp(self): self.cm = _make_cm( configs={'calendar': {}}, installed={'calendar': {'manifest': _CALENDAR_MANIFEST}}, ) 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']) # --------------------------------------------------------------------------- # TestServiceRegistryConfigMergeTypeCoercion # --------------------------------------------------------------------------- class TestServiceRegistryConfigMergeTypeCoercion(unittest.TestCase): def test_string_in_config_coerced_to_int(self): cm = _make_cm( configs={'calendar': {'port': '9999'}}, installed={'calendar': {'manifest': _CALENDAR_MANIFEST}}, ) 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( configs={'calendar': {'port': 'not_a_number'}}, installed={'calendar': {'manifest': _CALENDAR_MANIFEST}}, ) reg = ServiceRegistry(cm) svc = reg.get('calendar') self.assertEqual(svc['config']['port'], 5232) # --------------------------------------------------------------------------- # TestServiceRegistryRobustness # --------------------------------------------------------------------------- class TestServiceRegistryRobustness(unittest.TestCase): """Registry must not crash when records are corrupt or missing.""" def test_installed_record_with_no_manifest_skipped(self): cm = _make_cm(installed={'broken': {}}) reg = ServiceRegistry(cm) self.assertIsNone(reg.get('broken')) def test_list_all_skips_records_without_id(self): cm = _make_cm(installed={ 'noid': {'manifest': { 'name': 'No ID Service', 'kind': 'store', 'capabilities': {}, 'config_schema': {}, # 'id' key intentionally absent }}, 'calendar': {'manifest': _CALENDAR_MANIFEST}, }) reg = ServiceRegistry(cm) result = reg.list_all() ids = [s['id'] for s in result] self.assertNotIn(None, ids) self.assertIn('calendar', ids) self.assertEqual(len(result), 1) if __name__ == '__main__': unittest.main()