""" 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 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 from ip_utils import CONTAINER_OFFSETS # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- 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 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': 'git.pic.ngo/roof/myapp:latest', '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_tag_passes(self): m = _valid_manifest(image='git.pic.ngo/roof/something:1.2.3') ok, errs = ServiceStoreManager._validate_manifest(m) self.assertTrue(ok) self.assertEqual(errs, []) def test_image_git_pic_ngo_roof_no_tag_passes(self): 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) # --------------------------------------------------------------------------- # _allocate_service_ip # --------------------------------------------------------------------------- class TestAllocateServiceIp(unittest.TestCase): def test_first_allocation_skips_reserved_offsets_and_returns_first_free(self): """The first free offset after SERVICE_POOL_START(20) must not be in CONTAINER_OFFSETS.""" reserved_offsets = set(CONTAINER_OFFSETS.values()) # Find expected first offset (>= 20, not reserved) expected_offset = None for off in range(20, 255): if off not in reserved_offsets: expected_offset = off break expected_ip = f'172.20.0.{expected_offset}' mgr = _make_manager() ip = mgr._allocate_service_ip('svc-alpha') self.assertEqual(ip, expected_ip) def test_first_allocation_returns_172_20_0_20_for_clean_pool(self): """Offset 20 is not in CONTAINER_OFFSETS, so it should be the first allocated IP.""" self.assertNotIn(20, CONTAINER_OFFSETS.values(), "If offset 20 is now reserved, update this test") mgr = _make_manager() ip = mgr._allocate_service_ip('svc1') self.assertEqual(ip, '172.20.0.20') def test_reserved_container_offsets_are_skipped(self): """No allocated IP should land on a CONTAINER_OFFSETS offset.""" reserved_offsets = set(CONTAINER_OFFSETS.values()) mgr = _make_manager() ip = mgr._allocate_service_ip('svc2') import ipaddress allocated_offset = int(ipaddress.IPv4Address(ip)) - int(ipaddress.IPv4Address('172.20.0.0')) self.assertNotIn(allocated_offset, reserved_offsets) def test_already_taken_ips_are_skipped(self): """Already-assigned service IPs in service_ips are not reallocated.""" identity = { 'ip_range': '172.20.0.0/16', 'service_ips': {'svc-existing': '172.20.0.20'}, } mgr = _make_manager(identity=identity) ip = mgr._allocate_service_ip('svc-new') # 172.20.0.20 is taken, so must get the next available one self.assertNotEqual(ip, '172.20.0.20') # Should be 172.20.0.21 (offset 21 is vip_calendar in CONTAINER_OFFSETS — skip it) # Find what the next free one should be reserved_offsets = set(CONTAINER_OFFSETS.values()) expected_offset = None for off in range(20, 255): if off not in reserved_offsets and f'172.20.0.{off}' != '172.20.0.20': expected_offset = off break self.assertEqual(ip, f'172.20.0.{expected_offset}') def test_multiple_taken_ips_skipped_sequentially(self): """Allocator advances past multiple taken IPs correctly.""" reserved_offsets = set(CONTAINER_OFFSETS.values()) # Pre-fill the first few non-reserved offsets free_offsets = [off for off in range(20, 255) if off not in reserved_offsets] # Take the first 3 service_ips = {f'svc{i}': f'172.20.0.{off}' for i, off in enumerate(free_offsets[:3])} identity = {'ip_range': '172.20.0.0/16', 'service_ips': service_ips} mgr = _make_manager(identity=identity) ip = mgr._allocate_service_ip('svc-fourth') self.assertEqual(ip, f'172.20.0.{free_offsets[3]}') def test_exhausted_pool_raises_runtime_error(self): """Fill all 20-254 non-reserved offsets and expect RuntimeError.""" reserved_offsets = set(CONTAINER_OFFSETS.values()) service_ips = {} idx = 0 for off in range(20, 255): if off not in reserved_offsets: service_ips[f'svc{idx}'] = f'172.20.0.{off}' idx += 1 identity = {'ip_range': '172.20.0.0/16', 'service_ips': service_ips} mgr = _make_manager(identity=identity) with self.assertRaises(RuntimeError) as ctx: mgr._allocate_service_ip('overflow') self.assertIn('exhausted', str(ctx.exception).lower()) def test_uses_ip_range_from_identity(self): """Allocation respects a different ip_range like 10.10.0.0/16.""" identity = {'ip_range': '10.10.0.0/16', 'service_ips': {}} mgr = _make_manager(identity=identity) ip = mgr._allocate_service_ip('svc') self.assertTrue(ip.startswith('10.10.'), f'Expected 10.10.x.x, got {ip}') # --------------------------------------------------------------------------- # _render_compose_override # --------------------------------------------------------------------------- class TestRenderComposeOverride(unittest.TestCase): def test_empty_records_produces_valid_yaml_with_empty_services(self): mgr = _make_manager() output = mgr._render_compose_override({}) doc = yaml.safe_load(output) self.assertIn('services', doc) self.assertEqual(doc['services'], {}) self.assertIn('networks', doc) self.assertIn('cell-network', doc['networks']) def test_empty_records_has_no_volumes_key(self): mgr = _make_manager() output = mgr._render_compose_override({}) doc = yaml.safe_load(output) self.assertNotIn('volumes', doc) def test_single_service_renders_correct_definition(self): mgr = _make_manager() records = { 'myapp': { 'container_name': 'cell-myapp', 'service_ip': '172.20.0.20', 'manifest': { 'image': 'git.pic.ngo/roof/myapp:1.0', }, } } output = mgr._render_compose_override(records) doc = yaml.safe_load(output) svc = doc['services']['cell-myapp'] self.assertEqual(svc['image'], 'git.pic.ngo/roof/myapp:1.0') self.assertEqual(svc['container_name'], 'cell-myapp') self.assertEqual(svc['networks']['cell-network']['ipv4_address'], '172.20.0.20') self.assertEqual(svc['restart'], 'unless-stopped') def test_named_volumes_declared_at_top_level(self): mgr = _make_manager() records = { 'myapp': { 'container_name': 'cell-myapp', 'service_ip': '172.20.0.20', 'manifest': { 'image': 'git.pic.ngo/roof/myapp:1.0', 'volumes': [ {'name': 'myapp-data', 'mount': '/data'}, {'name': 'myapp-config', 'mount': '/config'}, ], }, } } output = mgr._render_compose_override(records) doc = yaml.safe_load(output) self.assertIn('volumes', doc) self.assertIn('myapp-data', doc['volumes']) self.assertIn('myapp-config', doc['volumes']) def test_named_volumes_appear_in_service_volumes_list(self): mgr = _make_manager() records = { 'myapp': { 'container_name': 'cell-myapp', 'service_ip': '172.20.0.20', 'manifest': { 'image': 'git.pic.ngo/roof/myapp:1.0', 'volumes': [{'name': 'myapp-data', 'mount': '/data'}], }, } } output = mgr._render_compose_override(records) doc = yaml.safe_load(output) svc_volumes = doc['services']['cell-myapp']['volumes'] self.assertIn('myapp-data:/data', svc_volumes) def test_environment_rendered_in_service(self): mgr = _make_manager() records = { 'myapp': { 'container_name': 'cell-myapp', 'service_ip': '172.20.0.20', 'manifest': { 'image': 'git.pic.ngo/roof/myapp:1.0', 'env': [ {'key': 'FOO', 'value': 'bar'}, {'key': 'PORT', 'value': '8080'}, ], }, } } output = mgr._render_compose_override(records) doc = yaml.safe_load(output) env = doc['services']['cell-myapp']['environment'] self.assertEqual(env['FOO'], 'bar') self.assertEqual(env['PORT'], '8080') def test_no_volumes_key_in_service_when_manifest_has_no_volumes(self): mgr = _make_manager() records = { 'myapp': { 'container_name': 'cell-myapp', 'service_ip': '172.20.0.20', 'manifest': {'image': 'git.pic.ngo/roof/myapp:1.0'}, } } output = mgr._render_compose_override(records) doc = yaml.safe_load(output) self.assertNotIn('volumes', doc['services']['cell-myapp']) def test_network_declared_as_external(self): mgr = _make_manager() output = mgr._render_compose_override({}) doc = yaml.safe_load(output) self.assertTrue(doc['networks']['cell-network']['external']) # --------------------------------------------------------------------------- # 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 = MagicMock( status_code=200, json=lambda: self._fake_index(), ) mock_get.return_value.raise_for_status = MagicMock() 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 = MagicMock( status_code=200, json=lambda: self._fake_index(), ) mock_get.return_value.raise_for_status = MagicMock() 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 = MagicMock( status_code=200, json=lambda: self._fake_index(), ) mock_get.return_value.raise_for_status = MagicMock() 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 = MagicMock( status_code=200, json=lambda: self._fake_index(), ) mock_get.return_value.raise_for_status = MagicMock() 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 = MagicMock( status_code=200, json=lambda: self._fake_index(), ) mock_get.return_value.raise_for_status = MagicMock() 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 = MagicMock( status_code=200, json=lambda: {'services': self._fake_index()}, ) mock_get.return_value.raise_for_status = MagicMock() result = mgr.list_services() self.assertEqual(len(result['available']), 2) # --------------------------------------------------------------------------- # install # --------------------------------------------------------------------------- class TestInstall(unittest.TestCase): def _mock_fetch(self, mgr, manifest): mgr._fetch_manifest = MagicMock(return_value=manifest) def _mock_write_compose(self, mgr): mgr._write_compose_override = MagicMock() def test_install_already_installed_returns_ok_already_installed(self): installed = {'myapp': {'id': 'myapp'}} mgr = _make_manager(installed=installed) with patch('firewall_manager.apply_service_rules'): result = mgr.install('myapp') self.assertTrue(result['ok']) self.assertTrue(result.get('already_installed')) def test_install_invalid_manifest_returns_errors(self): mgr = _make_manager() bad_manifest = {'id': 'myapp', 'image': 'bad-registry.io/img:latest'} self._mock_fetch(mgr, bad_manifest) self._mock_write_compose(mgr) with patch('firewall_manager.apply_service_rules'): result = mgr.install('myapp') self.assertFalse(result['ok']) self.assertIn('errors', result) def test_install_valid_manifest_returns_ok_true(self): mgr = _make_manager() manifest = _valid_manifest(id='myapp', container_name='cell-myapp') self._mock_fetch(mgr, manifest) self._mock_write_compose(mgr) with patch('firewall_manager.apply_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') result = mgr.install('myapp') self.assertTrue(result['ok']) self.assertFalse(result.get('already_installed', False)) def test_install_returns_service_ip(self): mgr = _make_manager() manifest = _valid_manifest(id='myapp', container_name='cell-myapp') self._mock_fetch(mgr, manifest) self._mock_write_compose(mgr) with patch('firewall_manager.apply_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') result = mgr.install('myapp') self.assertIn('service_ip', result) self.assertTrue(result['service_ip'].startswith('172.20.')) def test_install_returns_container_name(self): mgr = _make_manager() manifest = _valid_manifest(id='myapp', container_name='cell-myapp') self._mock_fetch(mgr, manifest) self._mock_write_compose(mgr) with patch('firewall_manager.apply_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') result = mgr.install('myapp') self.assertEqual(result['container_name'], 'cell-myapp') def test_install_calls_set_installed_service(self): mgr = _make_manager() manifest = _valid_manifest(id='myapp', container_name='cell-myapp') self._mock_fetch(mgr, manifest) self._mock_write_compose(mgr) with patch('firewall_manager.apply_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') mgr.install('myapp') mgr.config_manager.set_installed_service.assert_called_once() args = mgr.config_manager.set_installed_service.call_args[0] self.assertEqual(args[0], 'myapp') def test_install_calls_caddy_regenerate_when_service_has_caddy_route(self): mgr = _make_manager() manifest = _valid_manifest( id='myapp', container_name='cell-myapp', caddy_route={'subdomain': 'myapp', 'upstream': 'cell-myapp:8080'}, ) self._mock_fetch(mgr, manifest) self._mock_write_compose(mgr) with patch('firewall_manager.apply_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') mgr.install('myapp') mgr.caddy_manager.regenerate_with_installed.assert_called() def test_install_saves_service_ip_in_identity(self): mgr = _make_manager() manifest = _valid_manifest(id='myapp', container_name='cell-myapp') self._mock_fetch(mgr, manifest) self._mock_write_compose(mgr) with patch('firewall_manager.apply_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') mgr.install('myapp') mgr.config_manager.set_identity_field.assert_called() call_args = mgr.config_manager.set_identity_field.call_args[0] self.assertEqual(call_args[0], 'service_ips') self.assertIn('myapp', call_args[1]) def test_install_fetch_failure_returns_error(self): mgr = _make_manager() mgr._fetch_manifest = MagicMock(side_effect=Exception('connection refused')) result = mgr.install('nonexistent') self.assertFalse(result['ok']) self.assertIn('error', result) self.assertIn('fetch', result['error'].lower()) # --------------------------------------------------------------------------- # remove # --------------------------------------------------------------------------- class TestRemove(unittest.TestCase): def _mgr_with_installed(self, tmp_dir, service_id='myapp'): record = { 'container_name': 'cell-myapp', 'service_ip': '172.20.0.20', 'manifest': {'image': 'git.pic.ngo/roof/myapp:1.0', 'volumes': []}, 'iptables_rules': [], } installed = {service_id: record} mgr = _make_manager(tmp_dir=tmp_dir, installed=installed) # After remove, config_manager.get_installed_services returns empty mgr.config_manager.remove_installed_service = MagicMock() mgr.config_manager.get_installed_services.side_effect = [ installed, # first call (inside remove, initial check) {}, # second call (after removal, for compose rewrite) ] mgr._write_compose_override = MagicMock() return mgr def test_remove_not_installed_returns_error(self): mgr = _make_manager() with patch('firewall_manager.clear_service_rules'): result = mgr.remove('nosuchapp') self.assertFalse(result['ok']) self.assertIn('error', result) self.assertIn('not installed', result['error']) def test_remove_installed_returns_ok_true(self, tmp_dir='/tmp/pic-ssm-rm-test'): import tempfile, shutil tmp = tempfile.mkdtemp() try: mgr = self._mgr_with_installed(tmp) with patch('firewall_manager.clear_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') result = mgr.remove('myapp') self.assertTrue(result['ok']) finally: shutil.rmtree(tmp, ignore_errors=True) def test_remove_calls_remove_installed_service(self): import tempfile, shutil tmp = tempfile.mkdtemp() try: mgr = self._mgr_with_installed(tmp) with patch('firewall_manager.clear_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') mgr.remove('myapp') mgr.config_manager.remove_installed_service.assert_called_once_with('myapp') finally: shutil.rmtree(tmp, ignore_errors=True) def test_remove_calls_caddy_regenerate(self): import tempfile, shutil tmp = tempfile.mkdtemp() try: mgr = self._mgr_with_installed(tmp) with patch('firewall_manager.clear_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') mgr.remove('myapp') mgr.caddy_manager.regenerate_with_installed.assert_called() finally: shutil.rmtree(tmp, ignore_errors=True) def test_remove_purge_data_calls_docker_volume_rm(self): import tempfile, shutil tmp = tempfile.mkdtemp() try: record = { 'container_name': 'cell-myapp', 'service_ip': '172.20.0.20', 'manifest': { 'image': 'git.pic.ngo/roof/myapp:1.0', 'volumes': [{'name': 'myapp-data', 'mount': '/data'}], }, } mgr = _make_manager(tmp_dir=tmp, installed={'myapp': record}) mgr.config_manager.remove_installed_service = MagicMock() mgr.config_manager.get_installed_services.side_effect = [ {'myapp': record}, {} ] mgr._write_compose_override = MagicMock() with patch('firewall_manager.clear_service_rules'), \ patch('service_store_manager.subprocess.run') as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr='') mgr.remove('myapp', purge_data=True) # Check that docker volume rm was called with the volume name calls = [str(c) for c in mock_run.call_args_list] self.assertTrue( any('myapp-data' in c for c in calls), f'Expected docker volume rm myapp-data in calls: {calls}', ) finally: shutil.rmtree(tmp, ignore_errors=True) if __name__ == '__main__': unittest.main()