""" Tests for ServiceStoreManager — manifest validation, IP allocation, compose-override rendering, index listing, install, and remove. All external I/O (requests, subprocess, docker, config_manager, caddy_manager, container_manager) is mocked so these tests run without any live infrastructure. """ import json import os import sys import time import unittest from unittest.mock import MagicMock, patch, call import yaml sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'api')) from service_store_manager import ServiceStoreManager # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_streaming_mock(data): """Return a MagicMock that simulates a requests streaming response for ``data``.""" encoded = json.dumps(data).encode() raw = MagicMock() raw.read.return_value = encoded mock_resp = MagicMock(status_code=200) mock_resp.raise_for_status = MagicMock() mock_resp.raw = raw return mock_resp def _make_manager(tmp_dir=None, installed=None, identity=None): """Build a ServiceStoreManager backed by mock dependencies.""" cm = MagicMock() cm.get_installed_services.return_value = installed or {} cm.get_identity.return_value = identity or { 'ip_range': '172.20.0.0/16', 'service_ips': {}, } caddy = MagicMock() container = MagicMock() d = tmp_dir or '/tmp/pic-ssm-test' mgr = ServiceStoreManager( config_manager=cm, caddy_manager=caddy, container_manager=container, data_dir=d, config_dir=d, ) # Redirect compose override writes to a temp location so tests don't need /app mgr.compose_override = os.path.join(d, 'docker-compose.services.yml') return mgr _VALID_IMAGE = ( 'git.pic.ngo/roof/myapp@sha256:' + 'a' * 64 ) def _valid_manifest(**overrides): """Return a minimal valid manifest, with optional field overrides.""" m = { 'id': 'myapp', 'name': 'My App', 'version': '1.0.0', 'author': 'Test Author', 'image': _VALID_IMAGE, 'container_name': 'cell-myapp', } m.update(overrides) return m # --------------------------------------------------------------------------- # _validate_manifest — required fields # --------------------------------------------------------------------------- class TestValidateManifestRequiredFields(unittest.TestCase): def test_valid_manifest_passes(self): ok, errs = ServiceStoreManager._validate_manifest(_valid_manifest()) self.assertTrue(ok) self.assertEqual(errs, []) def test_missing_id_produces_error(self): m = _valid_manifest() del m['id'] ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) self.assertTrue(any('id' in e for e in errs)) def test_missing_name_produces_error(self): m = _valid_manifest() del m['name'] ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) self.assertTrue(any('name' in e for e in errs)) def test_missing_version_produces_error(self): m = _valid_manifest() del m['version'] ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) self.assertTrue(any('version' in e for e in errs)) def test_missing_author_produces_error(self): m = _valid_manifest() del m['author'] ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) self.assertTrue(any('author' in e for e in errs)) def test_missing_image_produces_error(self): m = _valid_manifest() del m['image'] ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) self.assertTrue(any('image' in e for e in errs)) def test_missing_container_name_produces_error(self): m = _valid_manifest() del m['container_name'] ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) self.assertTrue(any('container_name' in e for e in errs)) def test_all_required_fields_missing_produces_six_errors(self): ok, errs = ServiceStoreManager._validate_manifest({}) self.assertFalse(ok) self.assertEqual(len(errs), 6) # --------------------------------------------------------------------------- # _validate_manifest — image allowlist # --------------------------------------------------------------------------- class TestValidateManifestImage(unittest.TestCase): def test_image_outside_allowlist_rejected(self): m = _valid_manifest(image='docker.io/library/nginx:latest') ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) self.assertTrue(any('image must match' in e for e in errs)) def test_image_matching_git_pic_ngo_roof_with_digest_passes(self): digest = 'a' * 64 m = _valid_manifest(image=f'git.pic.ngo/roof/something@sha256:{digest}') ok, errs = ServiceStoreManager._validate_manifest(m) self.assertTrue(ok) self.assertEqual(errs, []) def test_image_tag_only_first_party_allowed(self): # First-party images without a digest pin are allowed (warning only). m = _valid_manifest(image='git.pic.ngo/roof/something:1.2.3') ok, errs = ServiceStoreManager._validate_manifest(m) self.assertTrue(ok) def test_image_git_pic_ngo_roof_no_tag_allowed(self): # No tag and no digest — Docker defaults to :latest; allowed for first-party. m = _valid_manifest(image='git.pic.ngo/roof/myservice') ok, errs = ServiceStoreManager._validate_manifest(m) self.assertTrue(ok) def test_image_wrong_registry_rejected(self): m = _valid_manifest(image='ghcr.io/roof/myapp:latest') ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) def test_image_partial_match_rejected(self): # Must be at root of git.pic.ngo/roof/, not nested elsewhere m = _valid_manifest(image='evil.git.pic.ngo/roof/myapp:latest') ok, errs = ServiceStoreManager._validate_manifest(m) self.assertFalse(ok) # --------------------------------------------------------------------------- # _validate_manifest — volume mounts # --------------------------------------------------------------------------- class TestValidateManifestVolumes(unittest.TestCase): def _make_with_volume(self, mount): m = _valid_manifest() m['volumes'] = [{'name': 'mydata', 'mount': mount}] return m def test_root_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/')) self.assertFalse(ok) self.assertTrue(any('Forbidden volume mount' in e for e in errs)) def test_etc_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/etc')) self.assertFalse(ok) def test_var_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/var')) self.assertFalse(ok) def test_proc_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/proc')) self.assertFalse(ok) def test_sys_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/sys')) self.assertFalse(ok) def test_dev_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/dev')) self.assertFalse(ok) def test_app_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/app')) self.assertFalse(ok) def test_run_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/run')) self.assertFalse(ok) def test_boot_mount_rejected(self): ok, errs = ServiceStoreManager._validate_manifest(self._make_with_volume('/boot')) self.assertFalse(ok) def test_home_roof_pic_prefix_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_volume('/home/roof/pic/data') ) self.assertFalse(ok) self.assertTrue(any('/home/roof/pic' in e for e in errs)) def test_home_roof_pic_exact_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_volume('/home/roof/pic') ) self.assertFalse(ok) def test_safe_data_mount_passes(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_volume('/data/myservice') ) self.assertTrue(ok) self.assertEqual(errs, []) def test_safe_srv_mount_passes(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_volume('/srv/myapp') ) self.assertTrue(ok) # --------------------------------------------------------------------------- # _validate_manifest — iptables rules # --------------------------------------------------------------------------- class TestValidateManifestIptables(unittest.TestCase): def _make_with_rule(self, **rule_fields): m = _valid_manifest() base_rule = { 'type': 'ACCEPT', 'dest_ip': '${SERVICE_IP}', 'dest_port': 8080, 'proto': 'tcp', } base_rule.update(rule_fields) m['iptables_rules'] = [base_rule] return m def test_valid_rule_passes(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(type='ACCEPT', dest_ip='${SERVICE_IP}', dest_port=8080) ) self.assertTrue(ok) self.assertEqual(errs, []) def test_type_not_accept_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(type='DROP') ) self.assertFalse(ok) self.assertTrue(any('type must be ACCEPT' in e for e in errs)) def test_type_reject_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(type='REJECT') ) self.assertFalse(ok) def test_dest_ip_not_service_ip_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(dest_ip='10.0.0.1') ) self.assertFalse(ok) self.assertTrue(any('dest_ip must be exactly' in e for e in errs)) def test_port_zero_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(dest_port=0) ) self.assertFalse(ok) self.assertTrue(any('dest_port' in e for e in errs)) def test_port_65536_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(dest_port=65536) ) self.assertFalse(ok) def test_port_1_accepted(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(dest_port=1) ) self.assertTrue(ok) def test_port_65535_accepted(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(dest_port=65535) ) self.assertTrue(ok) def test_port_as_string_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(dest_port='8080') ) self.assertFalse(ok) def test_proto_invalid_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(proto='icmp') ) self.assertFalse(ok) self.assertTrue(any('proto' in e for e in errs)) def test_proto_udp_accepted(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_rule(proto='udp') ) self.assertTrue(ok) # --------------------------------------------------------------------------- # _validate_manifest — env values # --------------------------------------------------------------------------- class TestValidateManifestEnv(unittest.TestCase): def _make_with_env(self, value): m = _valid_manifest() m['env'] = [{'key': 'MY_VAR', 'value': value}] return m def test_safe_alphanumeric_value_passes(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_env('hello123') ) self.assertTrue(ok) def test_safe_value_with_allowed_chars_passes(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_env('user@example.com') ) self.assertTrue(ok) def test_command_substitution_dollar_paren_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_env('$(cmd)') ) self.assertFalse(ok) self.assertTrue(any('disallowed characters' in e for e in errs)) def test_backtick_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_env('`cmd`') ) self.assertFalse(ok) def test_semicolon_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_env('val;rm -rf /') ) self.assertFalse(ok) def test_pipe_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_env('val|cat /etc/passwd') ) self.assertFalse(ok) def test_empty_value_passes(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_env('') ) self.assertTrue(ok) # --------------------------------------------------------------------------- # _validate_manifest — caddy_route subdomain # --------------------------------------------------------------------------- class TestValidateManifestSubdomain(unittest.TestCase): def _make_with_subdomain(self, subdomain): m = _valid_manifest() m['caddy_route'] = {'subdomain': subdomain, 'upstream': 'cell-myapp:8080'} return m def test_valid_subdomain_passes(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('myapp') ) self.assertTrue(ok) self.assertEqual(errs, []) def test_reserved_api_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('api') ) self.assertFalse(ok) self.assertTrue(any('reserved' in e for e in errs)) def test_reserved_admin_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('admin') ) self.assertFalse(ok) def test_reserved_www_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('www') ) self.assertFalse(ok) def test_reserved_webui_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('webui') ) self.assertFalse(ok) def test_subdomain_with_uppercase_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('MyApp') ) self.assertFalse(ok) self.assertTrue(any('subdomain must match' in e for e in errs)) def test_subdomain_starting_with_digit_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('1app') ) self.assertFalse(ok) def test_subdomain_with_underscore_rejected(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('my_app') ) self.assertFalse(ok) def test_subdomain_with_hyphen_passes(self): ok, errs = ServiceStoreManager._validate_manifest( self._make_with_subdomain('my-app') ) self.assertTrue(ok) def test_no_subdomain_in_caddy_route_passes(self): m = _valid_manifest() m['caddy_route'] = {'upstream': 'cell-myapp:8080'} ok, errs = ServiceStoreManager._validate_manifest(m) self.assertTrue(ok) def test_no_caddy_route_passes(self): ok, errs = ServiceStoreManager._validate_manifest(_valid_manifest()) self.assertTrue(ok) # --------------------------------------------------------------------------- # get_status # --------------------------------------------------------------------------- class TestGetStatus(unittest.TestCase): def test_returns_dict_with_required_keys(self): mgr = _make_manager(installed={'svc1': {}, 'svc2': {}}) status = mgr.get_status() self.assertIn('service', status) self.assertIn('running', status) self.assertIn('installed_count', status) def test_installed_count_reflects_config_manager(self): mgr = _make_manager(installed={'svc1': {}, 'svc2': {}, 'svc3': {}}) self.assertEqual(mgr.get_status()['installed_count'], 3) def test_installed_count_zero_when_none_installed(self): mgr = _make_manager(installed={}) self.assertEqual(mgr.get_status()['installed_count'], 0) def test_running_is_true(self): mgr = _make_manager() self.assertTrue(mgr.get_status()['running']) def test_service_name_is_service_store(self): mgr = _make_manager() self.assertEqual(mgr.get_status()['service'], 'service_store') # --------------------------------------------------------------------------- # list_services / fetch_index (caching) # --------------------------------------------------------------------------- class TestListServices(unittest.TestCase): def _fake_index(self): return [ {'id': 'svc1', 'name': 'Service One'}, {'id': 'svc2', 'name': 'Service Two'}, ] def test_returns_available_and_installed_keys(self): mgr = _make_manager() with patch('service_store_manager.requests.get') as mock_get: mock_get.return_value = _make_streaming_mock(self._fake_index()) result = mgr.list_services() self.assertIn('available', result) self.assertIn('installed', result) def test_available_list_comes_from_index(self): mgr = _make_manager() with patch('service_store_manager.requests.get') as mock_get: mock_get.return_value = _make_streaming_mock(self._fake_index()) result = mgr.list_services() self.assertEqual(len(result['available']), 2) self.assertEqual(result['available'][0]['id'], 'svc1') def test_installed_flag_reflects_config_manager(self): installed = {'svc1': {'id': 'svc1', 'name': 'Service One'}} mgr = _make_manager(installed=installed) with patch('service_store_manager.requests.get') as mock_get: mock_get.return_value = _make_streaming_mock(self._fake_index()) result = mgr.list_services() self.assertIn('svc1', result['installed']) def test_cache_prevents_second_http_request_within_ttl(self): mgr = _make_manager() with patch('service_store_manager.requests.get') as mock_get: mock_get.return_value = _make_streaming_mock(self._fake_index()) mgr.fetch_index() mgr.fetch_index() # Only one HTTP call despite two fetches mock_get.assert_called_once() def test_cache_expires_after_ttl_and_refetches(self): mgr = _make_manager() mgr._cache_ttl = 1 # 1 second TTL for the test with patch('service_store_manager.requests.get') as mock_get: mock_get.return_value = _make_streaming_mock(self._fake_index()) mgr.fetch_index() # Simulate TTL expiry by winding back the cache timestamp mgr._index_cache_time -= 2 mgr.fetch_index() self.assertEqual(mock_get.call_count, 2) def test_index_as_dict_with_services_key(self): """Index JSON wrapped in {'services': [...]} is also handled.""" mgr = _make_manager() with patch('service_store_manager.requests.get') as mock_get: mock_get.return_value = _make_streaming_mock({'services': self._fake_index()}) result = mgr.list_services() self.assertEqual(len(result['available']), 2) # --------------------------------------------------------------------------- # install (new architecture: ServiceComposer-driven) # --------------------------------------------------------------------------- def _make_ssm(config_manager=None, manifest=None, template='version: "3"\nservices: {}\n'): """Build a ServiceStoreManager with a mock service_composer.""" cm = config_manager or MagicMock() if config_manager is None: cm.get_installed_services.return_value = {} caddy = MagicMock() composer = MagicMock() composer._resolve_requires.return_value = None # no missing deps composer._resolve_dependents.return_value = [] composer.install.return_value = {'ok': True} ssm = ServiceStoreManager( config_manager=cm, caddy_manager=caddy, container_manager=MagicMock(), service_composer=composer, ) if manifest is not None: ssm._fetch_manifest = MagicMock(return_value=manifest) ssm._fetch_template = MagicMock(return_value=template) return ssm, cm, caddy, composer class TestInstall(unittest.TestCase): def test_install_already_installed_returns_ok_already_installed(self): cm = MagicMock() cm.get_installed_services.return_value = {'myapp': {'id': 'myapp'}} ssm, _, _, _ = _make_ssm(config_manager=cm) result = ssm.install('myapp') self.assertTrue(result['ok']) self.assertTrue(result.get('already_installed')) def test_install_fetch_failure_returns_error(self): ssm, _, _, _ = _make_ssm() ssm._fetch_manifest = MagicMock(side_effect=Exception('connection refused')) result = ssm.install('myapp') self.assertFalse(result['ok']) self.assertIn('error', result) self.assertIn('fetch', result['error'].lower()) def test_install_invalid_manifest_returns_errors(self): bad_manifest = {'id': 'myapp', 'image': 'bad-registry.io/img:latest'} ssm, _, _, _ = _make_ssm(manifest=bad_manifest) result = ssm.install('myapp') self.assertFalse(result['ok']) self.assertIn('errors', result) def test_install_missing_dep_returns_error(self): manifest = _valid_manifest(id='myapp', container_name='cell-myapp') ssm, _, _, composer = _make_ssm(manifest=manifest) composer._resolve_requires.return_value = 'Required services not installed: email' result = ssm.install('myapp') self.assertFalse(result['ok']) self.assertIn('error', result) self.assertIn('email', result['error']) def test_install_template_fetch_failure_returns_error(self): manifest = _valid_manifest(id='myapp', container_name='cell-myapp') ssm, _, _, _ = _make_ssm(manifest=manifest) ssm._fetch_template = MagicMock(side_effect=Exception('404 Not Found')) result = ssm.install('myapp') self.assertFalse(result['ok']) self.assertIn('error', result) self.assertIn('compose template', result['error'].lower()) def test_install_composer_install_failure_returns_error(self): manifest = _valid_manifest(id='myapp', container_name='cell-myapp') ssm, _, _, composer = _make_ssm(manifest=manifest) composer.install.return_value = {'ok': False, 'stderr': 'docker error'} result = ssm.install('myapp') self.assertFalse(result['ok']) self.assertIn('error', result) def test_install_calls_set_installed_service(self): manifest = _valid_manifest(id='myapp', container_name='cell-myapp') ssm, cm, _, _ = _make_ssm(manifest=manifest) ssm.install('myapp') cm.set_installed_service.assert_called_once() args = cm.set_installed_service.call_args[0] self.assertEqual(args[0], 'myapp') def test_install_record_contains_manifest(self): manifest = _valid_manifest(id='myapp', container_name='cell-myapp') ssm, cm, _, _ = _make_ssm(manifest=manifest) ssm.install('myapp') record = cm.set_installed_service.call_args[0][1] self.assertIn('manifest', record) def test_install_calls_caddy_regenerate(self): manifest = _valid_manifest(id='myapp', container_name='cell-myapp') ssm, _, caddy, _ = _make_ssm(manifest=manifest) ssm.install('myapp') caddy.regenerate_with_installed.assert_called() def test_install_returns_ok_true(self): manifest = _valid_manifest(id='myapp', container_name='cell-myapp') ssm, _, _, _ = _make_ssm(manifest=manifest) result = ssm.install('myapp') self.assertTrue(result['ok']) self.assertFalse(result.get('already_installed', False)) def test_install_without_composer_stores_record(self): """When service_composer=None, skip compose but still store the install record.""" manifest = _valid_manifest(id='myapp', container_name='cell-myapp') cm = MagicMock() cm.get_installed_services.return_value = {} caddy = MagicMock() ssm = ServiceStoreManager( config_manager=cm, caddy_manager=caddy, container_manager=MagicMock(), service_composer=None, ) ssm._fetch_manifest = MagicMock(return_value=manifest) ssm._fetch_template = MagicMock(return_value='version: "3"\nservices: {}\n') result = ssm.install('myapp') self.assertTrue(result['ok']) cm.set_installed_service.assert_called_once() # --------------------------------------------------------------------------- # remove (new architecture: ServiceComposer-driven) # --------------------------------------------------------------------------- class TestRemove(unittest.TestCase): def _make_mgr_with_installed(self, service_id='myapp'): record = { 'id': service_id, 'manifest': {'id': service_id, 'image': 'git.pic.ngo/roof/myapp:1.0'}, } installed = {service_id: record} cm = MagicMock() cm.get_installed_services.return_value = installed caddy = MagicMock() composer = MagicMock() composer._resolve_dependents.return_value = [] composer.remove.return_value = {'ok': True} ssm = ServiceStoreManager( config_manager=cm, caddy_manager=caddy, container_manager=MagicMock(), service_composer=composer, ) return ssm, cm, caddy, composer def test_remove_not_installed_returns_error(self): cm = MagicMock() cm.get_installed_services.return_value = {} ssm = ServiceStoreManager( config_manager=cm, caddy_manager=MagicMock(), container_manager=MagicMock(), ) result = ssm.remove('nosuchapp') self.assertFalse(result['ok']) self.assertIn('error', result) self.assertIn('not installed', result['error']) def test_remove_with_dependents_returns_error(self): ssm, _, _, composer = self._make_mgr_with_installed() composer._resolve_dependents.return_value = ['webmail'] result = ssm.remove('myapp') self.assertFalse(result['ok']) self.assertIn('error', result) self.assertIn('webmail', result['error']) def test_remove_calls_composer_remove(self): ssm, _, _, composer = self._make_mgr_with_installed() ssm.remove('myapp') composer.remove.assert_called_once_with('myapp', purge_data=False) def test_remove_calls_remove_installed_service(self): ssm, cm, _, _ = self._make_mgr_with_installed() ssm.remove('myapp') cm.remove_installed_service.assert_called_once_with('myapp') def test_remove_calls_caddy_regenerate(self): ssm, _, caddy, _ = self._make_mgr_with_installed() ssm.remove('myapp') caddy.regenerate_with_installed.assert_called() def test_remove_returns_ok_true(self): ssm, _, _, _ = self._make_mgr_with_installed() result = ssm.remove('myapp') self.assertTrue(result['ok']) def test_remove_purge_data_passed_to_composer(self): ssm, _, _, composer = self._make_mgr_with_installed() ssm.remove('myapp', purge_data=True) composer.remove.assert_called_once_with('myapp', purge_data=True) class TestReapplyOnStartup(unittest.TestCase): def _make_ssm_with_installed(self): ssm = _make_manager(installed={'svc1': {'service_ip': '172.20.1.10', 'iptables_rules': []}}) ssm.caddy_manager = MagicMock() return ssm def test_reapply_calls_egress_apply_all_when_wired(self): ssm = self._make_ssm_with_installed() mock_egress = MagicMock() ssm.egress_manager = mock_egress with patch('firewall_manager.apply_service_rules'): ssm.reapply_on_startup() mock_egress.apply_all.assert_called_once() def test_reapply_skips_egress_when_not_wired(self): """reapply_on_startup must not raise when egress_manager is None.""" ssm = self._make_ssm_with_installed() ssm.egress_manager = None with patch('firewall_manager.apply_service_rules'): ssm.reapply_on_startup() # must not raise def test_reapply_egress_failure_is_nonfatal(self): ssm = self._make_ssm_with_installed() mock_egress = MagicMock() mock_egress.apply_all.side_effect = RuntimeError('iptables error') ssm.egress_manager = mock_egress with patch('firewall_manager.apply_service_rules'): ssm.reapply_on_startup() # must not raise def test_reapply_always_regenerates_caddy_even_with_no_services(self): """Caddy must be regenerated on startup even when no store services are installed. Regression guard: a cell rename or fresh install produces a stale Caddyfile (wrong domain) if reapply_on_startup() returns early before calling caddy. """ ssm = _make_manager(installed={}) # no installed services ssm.caddy_manager = MagicMock() ssm.reapply_on_startup() ssm.caddy_manager.regenerate_with_installed.assert_called_once_with([]) def test_reapply_regenerates_caddy_with_installed_routes(self): """When services are installed their caddy_route dicts are passed to regenerate.""" route = {'subdomain': 'myapp', 'backend': 'http://myapp:8080'} ssm = _make_manager(installed={'myapp': {'service_ip': '172.20.1.10', 'iptables_rules': [], 'caddy_route': route}}) ssm.caddy_manager = MagicMock() with patch('firewall_manager.apply_service_rules'): ssm.reapply_on_startup() ssm.caddy_manager.regenerate_with_installed.assert_called_once_with([route]) if __name__ == '__main__': unittest.main()