roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1f2f9d9f6e0c1e09c1f008a259cd14b63105b6e7
pic/tests/test_service_store_manager.py
T
roof 1f2f9d9f6e
Unit Tests / test (push) Successful in 11m18s
Details
feat: add manifest_validator.py — security chokepoint for compose and manifest validation 
Rejects privileged compose configs (network_mode:host, pid:host, ipc:host,
userns_mode:host, cap_add:ALL, string commands, missing cell-network,
reserved container names). Validates manifest schema_version=3, image
digest pinning (sha256 required, :tag-only rejected), and provision hook
format. Wired into ServiceComposer.write_compose() and
ServiceStoreManager.install() as a single enforcement point.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 18:45:45 -04:00

776 lines
28 KiB
Python
Raw Blame History

"""
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_rejected(self):
# Digest pinning is required; tag-only images are rejected.
m = _valid_manifest(image='git.pic.ngo/roof/something:1.2.3')
ok, errs = ServiceStoreManager._validate_manifest(m)
self.assertFalse(ok)
def test_image_git_pic_ngo_roof_no_tag_rejected(self):
# No tag and no digest — rejected because digest pin is required.
m = _valid_manifest(image='git.pic.ngo/roof/myservice')
ok, errs = ServiceStoreManager._validate_manifest(m)
self.assertFalse(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)
if __name__ == '__main__':
unittest.main()
Reference in New Issue View Git Blame Copy Permalink