roof
87c321c1c9
Unit Tests / test (push) Successful in 11m21s
ServiceStoreManager.install() now delegates container lifecycle to ServiceComposer (per-service docker-compose.yml) instead of appending to a shared compose override. This eliminates IP pool allocation, compose override rendering, and the single-stack docker exec approach. Changes: - service_composer.py: add _resolve_requires(), _resolve_dependents(), reapply_active_services() — dependency graph and startup reapply - service_store_manager.py: rewrite install() and remove() to use ServiceComposer; add _fetch_template(); delete _allocate_service_ip(), _render_compose_override(), _write_compose_override(); remove() now guards against removing services that others depend on - managers.py: pass service_composer= to ServiceStoreManager - Tests: 13 new composer dep tests; TestInstall/TestRemove rewritten for the new composer-driven path; test_optional_services_feature.py updated Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
641 lines
28 KiB
Python
641 lines
28 KiB
Python
"""
|
|
Unit tests for ServiceComposer.
|
|
|
|
All subprocess calls and filesystem writes are mocked — no Docker daemon required.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch, mock_open, call
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
|
|
|
|
from service_composer import ServiceComposer, _SECRET_RE
|
|
|
|
|
|
# ── Helpers ─────────────────────────────────────────────────────────────────
|
|
|
|
def _make_cm(identity=None, service_config=None) -> MagicMock:
|
|
cm = MagicMock()
|
|
ident = identity or {'cell_name': 'testcell', 'domain': 'cell.local', 'domain_mode': 'lan'}
|
|
cm.get_identity.return_value = ident
|
|
cm.get_effective_domain.return_value = ident.get('domain', 'cell.local')
|
|
cm.configs = {}
|
|
if service_config:
|
|
cm.configs.update(service_config)
|
|
return cm
|
|
|
|
|
|
def _make_manifest(service_id='myservice', kind='store', schema=None):
|
|
return {
|
|
'id': service_id,
|
|
'kind': kind,
|
|
'config_schema': schema or {
|
|
'port': {'type': 'integer', 'default': 8080},
|
|
'username': {'type': 'string', 'default': 'admin'},
|
|
},
|
|
'containers': [f'cell-{service_id}'],
|
|
}
|
|
|
|
|
|
def _composer(cm=None, data_dir=None):
|
|
if data_dir is None:
|
|
data_dir = '/fake/data'
|
|
return ServiceComposer(config_manager=cm or _make_cm(), data_dir=data_dir)
|
|
|
|
|
|
# ── Template rendering ────────────────────────────────────────────────────────
|
|
|
|
class TestRenderTemplate(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.cm = _make_cm(service_config={'myservice': {'port': 9090}})
|
|
self.composer = _composer(self.cm)
|
|
|
|
def test_substitutes_pic_cfg_uppercase(self):
|
|
manifest = _make_manifest()
|
|
template = 'PORT=${PIC_CFG_PORT}'
|
|
result = self.composer.render_template('myservice', manifest, template)
|
|
self.assertEqual(result, 'PORT=9090')
|
|
|
|
def test_substitutes_default_when_no_saved_config(self):
|
|
cm = _make_cm()
|
|
composer = _composer(cm)
|
|
manifest = _make_manifest()
|
|
template = 'USER=${PIC_CFG_USERNAME}'
|
|
result = composer.render_template('myservice', manifest, template)
|
|
self.assertEqual(result, 'USER=admin')
|
|
|
|
def test_pic_domain_substituted(self):
|
|
manifest = _make_manifest()
|
|
template = 'DOMAIN=${PIC_DOMAIN}'
|
|
result = self.composer.render_template('myservice', manifest, template)
|
|
self.assertIn('cell.local', result)
|
|
|
|
def test_pic_cell_name_substituted(self):
|
|
manifest = _make_manifest()
|
|
template = 'CELL=${PIC_CELL_NAME}'
|
|
result = self.composer.render_template('myservice', manifest, template)
|
|
self.assertIn('testcell', result)
|
|
|
|
def test_pic_service_id_substituted(self):
|
|
manifest = _make_manifest(service_id='myservice')
|
|
template = 'ID=${PIC_SERVICE_ID}'
|
|
result = self.composer.render_template('myservice', manifest, template)
|
|
self.assertEqual(result, 'ID=myservice')
|
|
|
|
def test_pic_secret_generated_and_substituted(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
manifest = _make_manifest()
|
|
template = 'PASS=${PIC_SECRET_DB_PASS}'
|
|
result = composer.render_template('myservice', manifest, template)
|
|
self.assertNotIn('${PIC_SECRET_DB_PASS}', result)
|
|
self.assertNotEqual(result, 'PASS=')
|
|
# Secret is a non-empty string
|
|
password = result.replace('PASS=', '')
|
|
self.assertTrue(len(password) > 8)
|
|
|
|
def test_pic_secret_stable_across_calls(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
manifest = _make_manifest()
|
|
template = 'P=${PIC_SECRET_MY_PASS}'
|
|
r1 = composer.render_template('myservice', manifest, template)
|
|
r2 = composer.render_template('myservice', manifest, template)
|
|
self.assertEqual(r1, r2)
|
|
|
|
def test_pic_secret_different_per_service(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
m1 = _make_manifest('svc1')
|
|
m2 = _make_manifest('svc2')
|
|
t = 'P=${PIC_SECRET_PASS}'
|
|
r1 = composer.render_template('svc1', m1, t)
|
|
r2 = composer.render_template('svc2', m2, t)
|
|
self.assertNotEqual(r1, r2)
|
|
|
|
def test_multiple_secrets_all_replaced(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
manifest = _make_manifest()
|
|
template = 'A=${PIC_SECRET_KEY_A}\nB=${PIC_SECRET_KEY_B}'
|
|
result = composer.render_template('myservice', manifest, template)
|
|
self.assertNotIn('${PIC_SECRET_', result)
|
|
|
|
def test_no_unknown_vars_left_from_schema(self):
|
|
# Use a fresh composer with no saved config so defaults apply
|
|
composer = _composer(_make_cm())
|
|
manifest = _make_manifest(schema={
|
|
'port': {'type': 'integer', 'default': 3000},
|
|
})
|
|
template = 'PORT=${PIC_CFG_PORT}\nOTHER=${PIC_CFG_UNKNOWN}'
|
|
result = composer.render_template('myservice', manifest, template)
|
|
# Known var substituted with default, unknown left alone (no crash)
|
|
self.assertIn('PORT=3000', result)
|
|
self.assertIn('${PIC_CFG_UNKNOWN}', result)
|
|
|
|
|
|
# ── Write compose file ────────────────────────────────────────────────────────
|
|
|
|
class TestWriteCompose(unittest.TestCase):
|
|
|
|
def test_writes_rendered_content_to_correct_path(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
cm = _make_cm()
|
|
composer = ServiceComposer(config_manager=cm, data_dir=tmpdir)
|
|
manifest = _make_manifest()
|
|
template = (
|
|
'version: "3.8"\n'
|
|
'services:\n'
|
|
' app:\n'
|
|
' image: nginx\n'
|
|
' environment:\n'
|
|
' PORT: "${PIC_CFG_PORT}"\n'
|
|
'networks:\n'
|
|
' cell-network:\n'
|
|
' external: true\n'
|
|
)
|
|
composer.write_compose('myservice', manifest, template)
|
|
|
|
expected_path = os.path.join(
|
|
tmpdir, 'services', 'myservice', 'docker-compose.yml'
|
|
)
|
|
self.assertTrue(os.path.exists(expected_path))
|
|
with open(expected_path) as f:
|
|
content = f.read()
|
|
self.assertIn('8080', content)
|
|
|
|
def test_has_compose_file_false_before_write(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
self.assertFalse(composer.has_compose_file('newservice'))
|
|
|
|
def test_has_compose_file_true_after_write(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = ServiceComposer(config_manager=_make_cm(), data_dir=tmpdir)
|
|
manifest = _make_manifest()
|
|
valid_template = (
|
|
'version: "3.8"\n'
|
|
'services:\n'
|
|
' app:\n'
|
|
' image: nginx\n'
|
|
'networks:\n'
|
|
' cell-network:\n'
|
|
' external: true\n'
|
|
)
|
|
composer.write_compose('myservice', manifest, valid_template)
|
|
self.assertTrue(composer.has_compose_file('myservice'))
|
|
|
|
def test_atomic_write_via_tmp_file(self):
|
|
"""If fsync fails, the compose file should not be partially written."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = ServiceComposer(config_manager=_make_cm(), data_dir=tmpdir)
|
|
manifest = _make_manifest()
|
|
# Should not raise even if fsync not available
|
|
valid_template = (
|
|
'version: "3.8"\n'
|
|
'services:\n'
|
|
' app:\n'
|
|
' image: nginx\n'
|
|
'networks:\n'
|
|
' cell-network:\n'
|
|
' external: true\n'
|
|
)
|
|
composer.write_compose('myservice', manifest, valid_template)
|
|
path = os.path.join(tmpdir, 'services', 'myservice', 'docker-compose.yml')
|
|
self.assertTrue(os.path.exists(path))
|
|
|
|
|
|
# ── Secrets ───────────────────────────────────────────────────────────────────
|
|
|
|
class TestSecrets(unittest.TestCase):
|
|
|
|
def test_secrets_persisted_to_file(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
composer._get_or_create_secret('svc', 'PIC_SECRET_PASS')
|
|
secrets_path = os.path.join(tmpdir, 'service_secrets.json')
|
|
self.assertTrue(os.path.exists(secrets_path))
|
|
with open(secrets_path) as f:
|
|
data = json.load(f)
|
|
self.assertIn('svc', data)
|
|
self.assertIn('PIC_SECRET_PASS', data['svc'])
|
|
|
|
def test_clear_secrets_removes_service_entry(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
composer._get_or_create_secret('svc', 'PIC_SECRET_KEY')
|
|
composer._clear_secrets('svc')
|
|
secrets = composer._load_secrets()
|
|
self.assertNotIn('svc', secrets)
|
|
|
|
def test_clear_secrets_noop_when_no_secrets_file(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
# Should not raise
|
|
composer._clear_secrets('nonexistent')
|
|
|
|
def test_load_secrets_returns_empty_when_file_missing(self):
|
|
composer = _composer(data_dir='/nonexistent/path')
|
|
self.assertEqual(composer._load_secrets(), {})
|
|
|
|
|
|
# ── Subprocess execution ──────────────────────────────────────────────────────
|
|
|
|
class TestDockerComposeExecution(unittest.TestCase):
|
|
|
|
def _composer_with_compose_file(self, tmpdir, service_id='myservice'):
|
|
composer = ServiceComposer(config_manager=_make_cm(), data_dir=tmpdir)
|
|
svc_dir = os.path.join(tmpdir, 'services', service_id)
|
|
os.makedirs(svc_dir)
|
|
with open(os.path.join(svc_dir, 'docker-compose.yml'), 'w') as f:
|
|
f.write('services:\n app:\n image: nginx\n')
|
|
return composer
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_up_calls_docker_compose_up(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = self._composer_with_compose_file(tmpdir)
|
|
composer.up('myservice')
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn('up', cmd)
|
|
self.assertIn('-d', cmd)
|
|
self.assertIn('--project-name', cmd)
|
|
self.assertIn('pic-myservice', cmd)
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_down_calls_docker_compose_down(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = self._composer_with_compose_file(tmpdir)
|
|
composer.down('myservice')
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn('down', cmd)
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_down_with_purge_passes_volumes_flag(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = self._composer_with_compose_file(tmpdir)
|
|
composer.down('myservice', remove_volumes=True)
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn('--volumes', cmd)
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_restart_calls_docker_compose_restart(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = self._composer_with_compose_file(tmpdir)
|
|
composer.restart('myservice')
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn('restart', cmd)
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_status_parses_json_output(self, mock_run):
|
|
container_info = {'Name': 'myservice-app', 'State': 'running'}
|
|
mock_run.return_value = MagicMock(
|
|
returncode=0,
|
|
stdout=json.dumps(container_info),
|
|
stderr='',
|
|
)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = self._composer_with_compose_file(tmpdir)
|
|
result = composer.status('myservice')
|
|
self.assertTrue(result['ok'])
|
|
self.assertEqual(len(result['containers']), 1)
|
|
self.assertEqual(result['containers'][0]['Name'], 'myservice-app')
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_status_returns_empty_containers_on_bad_json(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='not json', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = self._composer_with_compose_file(tmpdir)
|
|
result = composer.status('myservice')
|
|
self.assertEqual(result['containers'], [])
|
|
|
|
def test_store_cmd_returns_error_when_no_compose_file(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = _composer(data_dir=tmpdir)
|
|
result = composer.up('nonexistent')
|
|
self.assertFalse(result['ok'])
|
|
self.assertIn('No compose file', result['error'])
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_up_uses_600s_timeout(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = self._composer_with_compose_file(tmpdir)
|
|
composer.up('myservice')
|
|
_, kwargs = mock_run.call_args
|
|
self.assertGreaterEqual(kwargs.get('timeout', 0), 600)
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_run_returns_error_on_timeout(self, mock_run):
|
|
import subprocess
|
|
mock_run.side_effect = subprocess.TimeoutExpired(cmd=[], timeout=120)
|
|
composer = _composer()
|
|
result = composer._run(['docker', 'compose', 'up'])
|
|
self.assertFalse(result['ok'])
|
|
self.assertIn('timed out', result['error'])
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_run_returns_false_on_nonzero_exit(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=1, stdout='', stderr='error msg')
|
|
composer = _composer()
|
|
result = composer._run(['docker', 'compose', 'up'])
|
|
self.assertFalse(result['ok'])
|
|
self.assertEqual(result['stderr'], 'error msg')
|
|
|
|
|
|
# ── Builtin lifecycle ─────────────────────────────────────────────────────────
|
|
|
|
class TestBuiltinLifecycle(unittest.TestCase):
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_restart_builtin_includes_container_names(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
composer = _composer()
|
|
composer.restart_builtin(['cell-radicale'])
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn('cell-radicale', cmd)
|
|
self.assertIn('restart', cmd)
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_status_builtin_includes_container_names(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
composer = _composer()
|
|
composer.status_builtin(['cell-mail', 'cell-rainloop'])
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn('cell-mail', cmd)
|
|
self.assertIn('cell-rainloop', cmd)
|
|
|
|
def test_restart_builtin_empty_list_returns_error(self):
|
|
composer = _composer()
|
|
result = composer.restart_builtin([])
|
|
self.assertFalse(result['ok'])
|
|
|
|
def test_status_builtin_empty_list_returns_error(self):
|
|
composer = _composer()
|
|
result = composer.status_builtin([])
|
|
self.assertFalse(result['ok'])
|
|
|
|
|
|
# ── Unified dispatch ──────────────────────────────────────────────────────────
|
|
|
|
class TestUnifiedDispatch(unittest.TestCase):
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_restart_service_builtin_uses_main_compose(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
composer = _composer()
|
|
manifest = _make_manifest(kind='builtin')
|
|
manifest['containers'] = ['cell-myservice']
|
|
composer.restart_service('myservice', manifest)
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn('cell-myservice', cmd)
|
|
# Main compose flag present
|
|
self.assertIn('-f', cmd)
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_restart_service_store_uses_per_service_compose(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = ServiceComposer(config_manager=_make_cm(), data_dir=tmpdir)
|
|
# Create compose file for the store service
|
|
svc_dir = os.path.join(tmpdir, 'services', 'storesvc')
|
|
os.makedirs(svc_dir)
|
|
with open(os.path.join(svc_dir, 'docker-compose.yml'), 'w') as f:
|
|
f.write('services:\n app:\n image: nginx\n')
|
|
manifest = _make_manifest('storesvc', kind='store')
|
|
composer.restart_service('storesvc', manifest)
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn('pic-storesvc', cmd)
|
|
|
|
|
|
# ── Remove ────────────────────────────────────────────────────────────────────
|
|
|
|
class TestServiceIdValidation(unittest.TestCase):
|
|
|
|
def test_valid_ids_accepted(self):
|
|
for sid in ('email', 'my-service', 'svc123', 'a1b2-c3'):
|
|
ServiceComposer._validate_service_id(sid) # should not raise
|
|
|
|
def test_dotdot_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
ServiceComposer._validate_service_id('..')
|
|
|
|
def test_dot_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
ServiceComposer._validate_service_id('.')
|
|
|
|
def test_slash_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
ServiceComposer._validate_service_id('evil/path')
|
|
|
|
def test_uppercase_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
ServiceComposer._validate_service_id('MyService')
|
|
|
|
def test_empty_string_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
ServiceComposer._validate_service_id('')
|
|
|
|
def test_newline_in_config_value_stripped(self):
|
|
"""A newline in a config value must not create a new YAML key (injection)."""
|
|
cm = _make_cm(service_config={'svc': {'port': '80\nnewline_attack: true'}})
|
|
composer = _composer(cm)
|
|
manifest = _make_manifest(schema={'port': {'type': 'string', 'default': '80'}})
|
|
result = composer.render_template('svc', manifest, 'PORT=${PIC_CFG_PORT}')
|
|
# The newline is stripped — 'newline_attack' is concatenated, not a separate YAML key
|
|
self.assertNotIn('\n', result)
|
|
|
|
|
|
class TestRemove(unittest.TestCase):
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_remove_deletes_compose_file(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = ServiceComposer(config_manager=_make_cm(), data_dir=tmpdir)
|
|
svc_dir = os.path.join(tmpdir, 'services', 'oldsvc')
|
|
os.makedirs(svc_dir)
|
|
compose_file = os.path.join(svc_dir, 'docker-compose.yml')
|
|
with open(compose_file, 'w') as f:
|
|
f.write('services: {}')
|
|
composer.remove('oldsvc', purge_data=False)
|
|
self.assertFalse(os.path.exists(compose_file))
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_remove_purge_deletes_service_directory(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = ServiceComposer(config_manager=_make_cm(), data_dir=tmpdir)
|
|
svc_dir = os.path.join(tmpdir, 'services', 'purgesvc')
|
|
os.makedirs(svc_dir)
|
|
with open(os.path.join(svc_dir, 'docker-compose.yml'), 'w') as f:
|
|
f.write('services: {}')
|
|
with open(os.path.join(svc_dir, 'data.txt'), 'w') as f:
|
|
f.write('important data')
|
|
composer.remove('purgesvc', purge_data=True)
|
|
self.assertFalse(os.path.exists(svc_dir))
|
|
|
|
@patch('service_composer.subprocess.run')
|
|
def test_remove_purge_clears_secrets(self, mock_run):
|
|
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
composer = ServiceComposer(config_manager=_make_cm(), data_dir=tmpdir)
|
|
composer._get_or_create_secret('purgesvc', 'PIC_SECRET_KEY')
|
|
svc_dir = os.path.join(tmpdir, 'services', 'purgesvc')
|
|
os.makedirs(svc_dir)
|
|
with open(os.path.join(svc_dir, 'docker-compose.yml'), 'w') as f:
|
|
f.write('services: {}')
|
|
composer.remove('purgesvc', purge_data=True)
|
|
secrets = composer._load_secrets()
|
|
self.assertNotIn('purgesvc', secrets)
|
|
|
|
|
|
# ── Parse ps json ─────────────────────────────────────────────────────────────
|
|
|
|
class TestParsePsJson(unittest.TestCase):
|
|
|
|
def test_single_json_object(self):
|
|
line = json.dumps({'Name': 'c1', 'State': 'running'})
|
|
result = ServiceComposer._parse_ps_json(line)
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0]['Name'], 'c1')
|
|
|
|
def test_multiple_json_lines(self):
|
|
lines = '\n'.join([
|
|
json.dumps({'Name': 'c1'}),
|
|
json.dumps({'Name': 'c2'}),
|
|
])
|
|
result = ServiceComposer._parse_ps_json(lines)
|
|
self.assertEqual(len(result), 2)
|
|
|
|
def test_ignores_blank_lines(self):
|
|
lines = '\n'.join([json.dumps({'Name': 'c1'}), '', json.dumps({'Name': 'c2'})])
|
|
result = ServiceComposer._parse_ps_json(lines)
|
|
self.assertEqual(len(result), 2)
|
|
|
|
def test_returns_empty_list_for_empty_output(self):
|
|
self.assertEqual(ServiceComposer._parse_ps_json(''), [])
|
|
|
|
def test_bad_json_lines_skipped(self):
|
|
lines = '\n'.join(['not json', json.dumps({'Name': 'c1'})])
|
|
result = ServiceComposer._parse_ps_json(lines)
|
|
self.assertEqual(len(result), 1)
|
|
|
|
|
|
# ── Dependency resolution ─────────────────────────────────────────────────────
|
|
|
|
class TestServiceComposerDeps(unittest.TestCase):
|
|
|
|
def _composer(self):
|
|
cm = MagicMock()
|
|
cm.configs = {}
|
|
cm.get_installed_services.return_value = {}
|
|
cm.get_identity.return_value = {}
|
|
cm.get_effective_domain.return_value = 'test.cell'
|
|
return ServiceComposer(config_manager=cm, data_dir='/tmp/test')
|
|
|
|
def test_resolve_requires_no_requires(self):
|
|
composer = self._composer()
|
|
manifest = {'id': 'webmail', 'requires': []}
|
|
result = composer._resolve_requires(manifest, {})
|
|
self.assertIsNone(result)
|
|
|
|
def test_resolve_requires_dep_installed(self):
|
|
composer = self._composer()
|
|
manifest = {'id': 'webmail', 'requires': ['email']}
|
|
installed = {'email': {'manifest': {'id': 'email'}}}
|
|
result = composer._resolve_requires(manifest, installed)
|
|
self.assertIsNone(result)
|
|
|
|
def test_resolve_requires_dep_missing(self):
|
|
composer = self._composer()
|
|
manifest = {'id': 'webmail', 'requires': ['email']}
|
|
result = composer._resolve_requires(manifest, {})
|
|
self.assertIsNotNone(result)
|
|
self.assertIn('email', result)
|
|
|
|
def test_resolve_requires_multiple_deps_partial(self):
|
|
composer = self._composer()
|
|
manifest = {'id': 'x', 'requires': ['email', 'calendar']}
|
|
installed = {'email': {'manifest': {'id': 'email'}}}
|
|
result = composer._resolve_requires(manifest, installed)
|
|
self.assertIsNotNone(result)
|
|
self.assertIn('calendar', result)
|
|
self.assertNotIn('email', result)
|
|
|
|
def test_resolve_requires_no_requires_key(self):
|
|
composer = self._composer()
|
|
manifest = {'id': 'files'} # no 'requires' key
|
|
result = composer._resolve_requires(manifest, {})
|
|
self.assertIsNone(result)
|
|
|
|
def test_resolve_dependents_none(self):
|
|
composer = self._composer()
|
|
installed = {
|
|
'email': {'manifest': {'id': 'email', 'requires': []}},
|
|
}
|
|
deps = composer._resolve_dependents('email', installed)
|
|
self.assertEqual(deps, [])
|
|
|
|
def test_resolve_dependents_found(self):
|
|
composer = self._composer()
|
|
installed = {
|
|
'email': {'manifest': {'id': 'email', 'requires': []}},
|
|
'webmail': {'manifest': {'id': 'webmail', 'requires': ['email']}},
|
|
}
|
|
deps = composer._resolve_dependents('email', installed)
|
|
self.assertIn('webmail', deps)
|
|
|
|
def test_resolve_dependents_excludes_self(self):
|
|
composer = self._composer()
|
|
installed = {
|
|
'email': {'manifest': {'id': 'email', 'requires': ['email']}}, # weird edge case
|
|
}
|
|
deps = composer._resolve_dependents('email', installed)
|
|
self.assertNotIn('email', deps)
|
|
|
|
def test_resolve_dependents_empty_installed(self):
|
|
composer = self._composer()
|
|
deps = composer._resolve_dependents('email', {})
|
|
self.assertEqual(deps, [])
|
|
|
|
def test_reapply_active_services_calls_up(self):
|
|
cm = MagicMock()
|
|
cm.get_installed_services.return_value = {'email': {'manifest': {'id': 'email'}}}
|
|
composer = ServiceComposer(config_manager=cm, data_dir='/tmp/test')
|
|
composer.has_compose_file = MagicMock(return_value=True)
|
|
composer.up = MagicMock(return_value={'ok': True})
|
|
composer.reapply_active_services()
|
|
composer.up.assert_called_once_with('email')
|
|
|
|
def test_reapply_active_services_skips_missing_compose(self):
|
|
cm = MagicMock()
|
|
cm.get_installed_services.return_value = {'email': {'manifest': {'id': 'email'}}}
|
|
composer = ServiceComposer(config_manager=cm, data_dir='/tmp/test')
|
|
composer.has_compose_file = MagicMock(return_value=False)
|
|
composer.up = MagicMock()
|
|
composer.reapply_active_services()
|
|
composer.up.assert_not_called()
|
|
|
|
def test_reapply_active_services_empty(self):
|
|
cm = MagicMock()
|
|
cm.get_installed_services.return_value = {}
|
|
composer = ServiceComposer(config_manager=cm, data_dir='/tmp/test')
|
|
composer.up = MagicMock()
|
|
composer.reapply_active_services()
|
|
composer.up.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|