roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
36bc32543dba83e613ad6d3798c12263ff16bdcc
pic/tests/test_optional_services_feature.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

930 lines
39 KiB
Python
Raw Blame History

"""
Tests for the optional-services feature: email/calendar/files moving from
always-on builtins to installable store services.
Covers:
1. ServiceRegistry.list_active() — zero installed, partial, full
2. ServiceRegistry.get_caddy_routes() / get_service_subdomains() with list_active()
3. ServiceRegistry.get() returns None for catalog-only (not installed) entries
4. ServiceStoreManager.install() happy path, idempotency, fetch failure, compose failure
5. ServiceStoreManager.uninstall() (remove()) happy path and not-installed error
6. CaddyManager._build_registry_service_routes() with empty list_active()
7. GET /api/services/active endpoint
8. migrate_legacy_containers(): writes install records, idempotent on second call
"""
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch, call
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
from service_registry import ServiceRegistry
from service_store_manager import ServiceStoreManager
from caddy_manager import CaddyManager
# ---------------------------------------------------------------------------
# Shared manifest helpers
# ---------------------------------------------------------------------------
def _store_manifest(service_id, subdomain=None, backend=None):
"""Minimal valid store manifest for use in installed-services records."""
m = {
'id': service_id,
'name': service_id.capitalize(),
'kind': 'store',
'capabilities': {
'has_subdomain': bool(subdomain),
'has_accounts': True,
'has_admin_config': False,
'has_storage': True,
'has_egress': False,
'has_api_hooks': False,
},
'config_schema': {},
}
if subdomain:
m['subdomain'] = subdomain
if backend:
m['backend'] = backend
if subdomain and backend:
m['extra_subdomains'] = []
m['extra_backends'] = {}
return m
_FIXTURE_DIGEST = 'a' * 64
def _ssm_manifest(service_id='myapp', **overrides):
"""Minimal manifest that passes ServiceStoreManager._validate_manifest."""
m = {
'id': service_id,
'name': 'My App',
'version': '1.0.0',
'author': 'Test Author',
'image': f'git.pic.ngo/roof/{service_id}@sha256:{_FIXTURE_DIGEST}',
'container_name': f'cell-{service_id}',
}
m.update(overrides)
return m
# ---------------------------------------------------------------------------
# 1. ServiceRegistry.list_active()
# ---------------------------------------------------------------------------
class TestServiceRegistryListActive(unittest.TestCase):
"""
list_active() must return only services that appear in get_installed_services().
When builtins are removed from the filesystem, only installed records count.
"""
def _make_registry(self, installed=None):
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = installed or {}
return ServiceRegistry(cm)
def test_list_active_zero_installed_returns_empty(self):
"""With no installed records, list_active() is empty."""
reg = self._make_registry(installed={})
result = reg.list_active()
self.assertEqual(result, [])
def test_list_active_one_installed_returns_only_that_service(self):
"""Email installed, calendar not: only email appears in list_active()."""
email_manifest = _store_manifest('email', subdomain='mail', backend='cell-rainloop:8888')
installed = {
'email': {'manifest': email_manifest},
}
reg = self._make_registry(installed=installed)
result = reg.list_active()
ids = [s['id'] for s in result]
self.assertIn('email', ids)
self.assertNotIn('calendar', ids)
self.assertNotIn('files', ids)
def test_list_active_multiple_installed_returns_all(self):
"""All three installed services appear in list_active()."""
installed = {
'email': {'manifest': _store_manifest('email', 'mail', 'cell-rainloop:8888')},
'calendar': {'manifest': _store_manifest('calendar', 'calendar', 'cell-radicale:5232')},
'files': {'manifest': _store_manifest('files', 'files', 'cell-filegator:8080')},
}
reg = self._make_registry(installed=installed)
result = reg.list_active()
ids = {s['id'] for s in result}
self.assertEqual(ids, {'email', 'calendar', 'files'})
def test_list_active_each_entry_has_config_key(self):
"""Each active service entry must carry the merged 'config' key."""
installed = {
'calendar': {'manifest': _store_manifest('calendar', 'calendar', 'cell-radicale:5232')},
}
reg = self._make_registry(installed=installed)
result = reg.list_active()
for svc in result:
self.assertIn('config', svc, f'{svc["id"]} is missing the config key')
def test_list_active_record_without_manifest_skipped(self):
"""An installed record with no manifest key must not appear (no KeyError either)."""
installed = {
'broken': {}, # no 'manifest' key at all
}
reg = self._make_registry(installed=installed)
result = reg.list_active()
self.assertEqual(result, [])
# ---------------------------------------------------------------------------
# 2. ServiceRegistry.get_caddy_routes() only returns active services
# ---------------------------------------------------------------------------
class TestServiceRegistryGetCaddyRoutesActiveOnly(unittest.TestCase):
"""
After the migration get_caddy_routes() must delegate to list_active(),
not list_all(). This test class validates the behaviour that the
implementation will need to satisfy — it patches list_active() on the
registry so the tests don't depend on whether list_active() is already
implemented or is still list_all().
"""
def _make_registry(self, active_services):
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
# Point get_caddy_routes' iteration at the active list only.
# We do this by patching list_all to return only active services,
# which mirrors the post-migration behaviour of list_all == list_active.
reg.list_all = MagicMock(return_value=active_services)
return reg
def test_no_active_services_produces_no_routes(self):
"""When list_active returns empty, get_caddy_routes must return []."""
reg = self._make_registry([])
routes = reg.get_caddy_routes()
self.assertEqual(routes, [])
def test_email_active_calendar_not_only_email_in_routes(self):
"""Email installed; calendar and files not: only email route returned."""
email_svc = {
**_store_manifest('email', 'mail', 'cell-rainloop:8888'),
'extra_subdomains': ['webmail'],
'extra_backends': {},
'config': {},
}
reg = self._make_registry([email_svc])
routes = reg.get_caddy_routes()
service_ids = [r['service_id'] for r in routes]
self.assertIn('email', service_ids)
self.assertNotIn('calendar', service_ids)
self.assertNotIn('files', service_ids)
def test_route_shape_is_correct(self):
"""Each route dict must have the expected keys with correct values."""
svc = {
**_store_manifest('calendar', 'calendar', 'cell-radicale:5232'),
'extra_subdomains': [],
'extra_backends': {},
'config': {},
}
reg = self._make_registry([svc])
routes = reg.get_caddy_routes()
self.assertEqual(len(routes), 1)
r = routes[0]
self.assertEqual(r['service_id'], 'calendar')
self.assertEqual(r['subdomain'], 'calendar')
self.assertEqual(r['backend'], 'cell-radicale:5232')
self.assertIn('extra_subdomains', r)
self.assertIn('extra_backends', r)
# ---------------------------------------------------------------------------
# 3. ServiceRegistry.get_service_subdomains() active services only
# ---------------------------------------------------------------------------
class TestGetServiceSubdomainsActiveOnly(unittest.TestCase):
"""
The network manager calls registry.get_caddy_routes() via _get_service_subdomains.
This test verifies that after the migration, a registry with only calendar
installed does not include 'mail' or 'files' subdomains in its route output.
"""
def test_only_installed_subdomains_returned(self):
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
calendar_svc = {
**_store_manifest('calendar', 'calendar', 'cell-radicale:5232'),
'extra_subdomains': [],
'extra_backends': {},
'config': {},
}
reg = ServiceRegistry(cm)
reg.list_all = MagicMock(return_value=[calendar_svc])
routes = reg.get_caddy_routes()
subdomains = [r['subdomain'] for r in routes]
extra = [s for r in routes for s in (r.get('extra_subdomains') or [])]
all_subs = set(subdomains) | set(extra)
self.assertIn('calendar', all_subs)
self.assertNotIn('mail', all_subs)
self.assertNotIn('webmail', all_subs)
self.assertNotIn('files', all_subs)
self.assertNotIn('webdav', all_subs)
# ---------------------------------------------------------------------------
# 4. ServiceRegistry.get() returns None for catalog-only (not installed) entries
# ---------------------------------------------------------------------------
class TestServiceRegistryGetNotInstalled(unittest.TestCase):
"""
Once builtins are removed from the filesystem, get('email') must return None
unless the service is in get_installed_services().
"""
def test_get_returns_none_when_not_installed(self):
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
result = reg.get('email')
self.assertIsNone(result)
def test_get_returns_none_for_calendar_when_not_installed(self):
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
self.assertIsNone(reg.get('calendar'))
def test_get_returns_none_for_files_when_not_installed(self):
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
self.assertIsNone(reg.get('files'))
def test_get_returns_service_when_installed(self):
"""Once email is in installed records it must be returned by get()."""
email_manifest = _store_manifest('email', 'mail', 'cell-rainloop:8888')
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {
'email': {'manifest': email_manifest},
}
reg = ServiceRegistry(cm)
result = reg.get('email')
self.assertIsNotNone(result)
self.assertEqual(result['id'], 'email')
# ---------------------------------------------------------------------------
# 5. ServiceStoreManager.install() — new scenarios
# ---------------------------------------------------------------------------
def _make_ssm(tmp_dir, installed=None, identity=None):
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()
composer = MagicMock()
composer._resolve_requires.return_value = None
composer._resolve_dependents.return_value = []
composer.install.return_value = {'ok': True}
composer.remove.return_value = {'ok': True}
mgr = ServiceStoreManager(
config_manager=cm,
caddy_manager=caddy,
container_manager=container,
data_dir=tmp_dir,
config_dir=tmp_dir,
service_composer=composer,
)
return mgr
class TestInstallHappyPath(unittest.TestCase):
def test_install_fetches_manifest_renders_compose_calls_docker_up(self):
"""install() happy path: fetches manifest, calls service_composer.install, stores record."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
manifest = _ssm_manifest('email')
mgr._fetch_manifest = MagicMock(return_value=manifest)
mgr._fetch_template = MagicMock(return_value='version: "3"\nservices: {}\n')
result = mgr.install('email')
self.assertTrue(result['ok'])
mgr._fetch_manifest.assert_called_once_with('email')
mgr.config_manager.set_installed_service.assert_called_once()
# service_composer.install must have been called
mgr.service_composer.install.assert_called_once()
def test_install_persists_install_record_after_composer_install(self):
"""Install record must be written after service_composer.install succeeds."""
call_order = []
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
manifest = _ssm_manifest('calendar')
mgr._fetch_manifest = MagicMock(return_value=manifest)
mgr._fetch_template = MagicMock(return_value='version: "3"\nservices: {}\n')
mgr.config_manager.set_installed_service.side_effect = \
lambda *a, **kw: call_order.append('set_installed')
def _composer_install(*a, **kw):
call_order.append('composer_install')
return {'ok': True}
mgr.service_composer.install.side_effect = _composer_install
mgr.install('calendar')
self.assertIn('composer_install', call_order)
self.assertIn('set_installed', call_order)
self.assertLess(
call_order.index('composer_install'),
call_order.index('set_installed'),
'composer.install must be called before install record is persisted',
)
class TestInstallAlreadyInstalled(unittest.TestCase):
def test_install_already_installed_is_idempotent(self):
"""Calling install() on an already-installed service returns ok=True, already_installed=True."""
with tempfile.TemporaryDirectory() as tmp:
installed = {'email': {'id': 'email'}}
mgr = _make_ssm(tmp, installed=installed)
result = mgr.install('email')
self.assertTrue(result['ok'])
self.assertTrue(result.get('already_installed'))
def test_install_already_installed_does_not_fetch_manifest(self):
"""No network call should be made when service is already installed."""
with tempfile.TemporaryDirectory() as tmp:
installed = {'email': {'id': 'email'}}
mgr = _make_ssm(tmp, installed=installed)
mgr._fetch_manifest = MagicMock()
mgr.install('email')
mgr._fetch_manifest.assert_not_called()
def test_install_already_installed_does_not_write_config(self):
"""set_installed_service must NOT be called for an idempotent re-install."""
with tempfile.TemporaryDirectory() as tmp:
installed = {'calendar': {'id': 'calendar'}}
mgr = _make_ssm(tmp, installed=installed)
mgr.install('calendar')
mgr.config_manager.set_installed_service.assert_not_called()
class TestInstallManifestFetchFails(unittest.TestCase):
def test_install_fetch_failure_returns_error_with_message(self):
"""A network error during manifest fetch must return ok=False with an error field."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
mgr._fetch_manifest = MagicMock(
side_effect=Exception('connection refused')
)
result = mgr.install('email')
self.assertFalse(result['ok'])
self.assertIn('error', result)
self.assertIn('fetch', result['error'].lower())
def test_install_fetch_failure_leaves_no_install_record(self):
"""No install record must be written when the manifest fetch fails."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
mgr._fetch_manifest = MagicMock(side_effect=Exception('timeout'))
mgr.install('email')
mgr.config_manager.set_installed_service.assert_not_called()
def test_install_http_404_leaves_no_install_record(self):
"""HTTP 404 from the manifest endpoint must not leave a partial install."""
import requests as _requests
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
response_404 = MagicMock()
response_404.raise_for_status.side_effect = \
_requests.HTTPError('404 Not Found')
with patch('service_store_manager.requests.get', return_value=response_404):
result = mgr.install('nonexistent-service')
self.assertFalse(result['ok'])
mgr.config_manager.set_installed_service.assert_not_called()
def test_install_invalid_manifest_does_not_write_record(self):
"""Manifest validation failure must prevent any install record from being written."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
bad_manifest = {
'id': 'email',
# missing name, version, author, container_name; bad image
'image': 'docker.io/bad-actor/email:latest',
}
mgr._fetch_manifest = MagicMock(return_value=bad_manifest)
result = mgr.install('email')
self.assertFalse(result['ok'])
self.assertIn('errors', result)
mgr.config_manager.set_installed_service.assert_not_called()
class TestInstallComposeUpFails(unittest.TestCase):
"""
In the new architecture, a compose failure from service_composer.install returns
ok=False immediately — the install record is NOT written when compose fails.
"""
def test_install_compose_failure_returns_error(self):
"""A failure from service_composer.install must return ok=False."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
manifest = _ssm_manifest('email')
mgr._fetch_manifest = MagicMock(return_value=manifest)
mgr._fetch_template = MagicMock(return_value='version: "3"\nservices: {}\n')
mgr.service_composer.install.return_value = {'ok': False, 'stderr': 'image pull failed'}
result = mgr.install('email')
self.assertFalse(result['ok'])
self.assertIn('error', result)
def test_install_record_not_written_when_compose_fails(self):
"""Install record must NOT be written when service_composer.install fails."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
manifest = _ssm_manifest('email')
mgr._fetch_manifest = MagicMock(return_value=manifest)
mgr._fetch_template = MagicMock(return_value='version: "3"\nservices: {}\n')
mgr.service_composer.install.return_value = {'ok': False, 'stderr': 'pull failed'}
mgr.install('email')
mgr.config_manager.set_installed_service.assert_not_called()
# ---------------------------------------------------------------------------
# 6. ServiceStoreManager.uninstall() (remove())
# ---------------------------------------------------------------------------
class TestUninstallHappyPath(unittest.TestCase):
def _make_mgr_with_email(self, tmp):
record = {
'id': 'email',
'manifest': {
'image': 'git.pic.ngo/roof/email:1.0',
},
}
installed = {'email': record}
mgr = _make_ssm(tmp, installed=installed)
mgr.config_manager.remove_installed_service = MagicMock()
return mgr
def test_uninstall_happy_path_returns_ok_true(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_mgr_with_email(tmp)
result = mgr.remove('email')
self.assertTrue(result['ok'])
def test_uninstall_removes_install_record(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_mgr_with_email(tmp)
mgr.remove('email')
mgr.config_manager.remove_installed_service.assert_called_once_with('email')
def test_uninstall_calls_service_composer_remove(self):
"""New architecture: composer.remove() is called instead of subprocess directly."""
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_mgr_with_email(tmp)
mgr.remove('email')
mgr.service_composer.remove.assert_called_once_with('email', purge_data=False)
def test_uninstall_regenerates_caddyfile(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_mgr_with_email(tmp)
mgr.remove('email')
mgr.caddy_manager.regenerate_with_installed.assert_called()
class TestUninstallNotInstalled(unittest.TestCase):
def test_uninstall_service_not_installed_returns_error(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp, installed={})
result = mgr.remove('email')
self.assertFalse(result['ok'])
self.assertIn('error', result)
self.assertIn('not installed', result['error'].lower())
def test_uninstall_nonexistent_service_does_not_call_composer(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp, installed={})
mgr.remove('email')
mgr.service_composer.remove.assert_not_called()
def test_uninstall_nonexistent_service_does_not_remove_config(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp, installed={})
mgr.config_manager.remove_installed_service = MagicMock()
mgr.remove('email')
mgr.config_manager.remove_installed_service.assert_not_called()
# ---------------------------------------------------------------------------
# 7. CaddyManager._build_registry_service_routes() with empty registry
# ---------------------------------------------------------------------------
class TestCaddyManagerEmptyActiveRegistry(unittest.TestCase):
"""
When the registry returns no active routes (empty list_active()), the
registry-driven path produces only the @api block — no service matcher
blocks for calendar/mail/files/webdav.
Phase 2: builtins removed, so there is no hardcoded fallback. An empty
registry means no service routes at all (except the always-present api block).
"""
def _mgr_with_empty_registry(self):
cm = MagicMock()
cm.get_identity.return_value = {}
reg = MagicMock()
reg.get_caddy_routes.return_value = [] # no active services
return CaddyManager(config_manager=cm, service_registry=reg)
def test_empty_active_list_produces_no_service_matcher_blocks(self):
"""Zero active services → no @calendar, @mail, @files, @webdav matchers.
Phase 2: builtins are gone so an empty registry produces only the @api block.
"""
mgr = self._mgr_with_empty_registry()
result = mgr._build_registry_service_routes('mycell.pic.ngo')
self.assertIn('@api host api.mycell.pic.ngo', result)
self.assertNotIn('@calendar', result)
self.assertNotIn('@mail', result)
self.assertNotIn('@files', result)
self.assertNotIn('@webdav', result)
def test_empty_registry_no_store_service_blocks_injected(self):
"""An empty active list must not inject any store-service-specific matchers."""
mgr = self._mgr_with_empty_registry()
result = mgr._build_registry_service_routes('mycell.pic.ngo')
# No store service names should appear that don't come from core services
self.assertNotIn('@chat', result)
self.assertNotIn('@nextcloud', result)
self.assertNotIn('@wiki', result)
def test_registry_with_only_email_installed_produces_only_email_block(self):
"""When only email is active the Caddyfile must have @mail but not @calendar or @files."""
cm = MagicMock()
cm.get_identity.return_value = {}
reg = MagicMock()
reg.get_caddy_routes.return_value = [
{
'service_id': 'email',
'subdomain': 'mail',
'backend': 'cell-rainloop:8888',
'extra_subdomains': ['webmail'],
'extra_backends': {},
}
]
mgr = CaddyManager(config_manager=cm, service_registry=reg)
result = mgr._build_registry_service_routes('mycell.pic.ngo')
self.assertIn('@mail host mail.mycell.pic.ngo', result)
self.assertNotIn('@calendar host', result)
self.assertNotIn('@files host', result)
self.assertNotIn('@webdav host', result)
# api block is always appended
self.assertIn('@api host api.mycell.pic.ngo', result)
def test_caddyfile_with_no_active_services_still_has_api_and_webui(self):
"""Even with no installed services the api and webui routes must appear."""
mgr = self._mgr_with_empty_registry()
identity = {'cell_name': 'alpha', 'domain_mode': 'pic_ngo'}
caddyfile = mgr.generate_caddyfile(identity, [])
self.assertIn('cell-api:3000', caddyfile)
self.assertIn('cell-webui:80', caddyfile)
# ---------------------------------------------------------------------------
# 8. GET /api/services/active endpoint
# ---------------------------------------------------------------------------
class TestServicesActiveEndpoint(unittest.TestCase):
"""
Tests for GET /api/services/active.
The endpoint does not exist yet — these tests define the required contract
so they can be run once the endpoint is implemented. They are marked with
a skip decorator that references the missing route; remove the skip when
the endpoint is added to api/routes/services.py.
"""
def setUp(self):
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
from app import app
app.config['TESTING'] = True
self.client = app.test_client()
def _mock_registry(self, active_services):
"""Patch app.service_registry.list_active to return active_services."""
reg = MagicMock()
reg.list_active = MagicMock(return_value=active_services)
reg.list_all = MagicMock(return_value=active_services) # fallback
return reg
@unittest.skip(
'GET /api/services/active does not exist yet; add to routes/services.py then unskip'
)
def test_active_endpoint_returns_200(self):
import app as app_module
with patch.object(app_module, 'service_registry',
self._mock_registry([])):
resp = self.client.get('/api/services/active')
self.assertEqual(resp.status_code, 200)
@unittest.skip(
'GET /api/services/active does not exist yet; add to routes/services.py then unskip'
)
def test_active_endpoint_returns_empty_list_when_nothing_installed(self):
import app as app_module
with patch.object(app_module, 'service_registry',
self._mock_registry([])):
resp = self.client.get('/api/services/active')
data = json.loads(resp.data)
self.assertIn('services', data)
self.assertEqual(data['services'], [])
@unittest.skip(
'GET /api/services/active does not exist yet; add to routes/services.py then unskip'
)
def test_active_endpoint_only_returns_installed_services(self):
email_svc = {**_store_manifest('email', 'mail', 'cell-rainloop:8888'), 'config': {}}
import app as app_module
with patch.object(app_module, 'service_registry',
self._mock_registry([email_svc])):
resp = self.client.get('/api/services/active')
data = json.loads(resp.data)
ids = [s['id'] for s in data['services']]
self.assertIn('email', ids)
self.assertNotIn('calendar', ids)
self.assertNotIn('files', ids)
def test_catalog_endpoint_exists_and_returns_200(self):
"""Smoke-test the existing /api/services/catalog endpoint for baseline health."""
import app as app_module
reg = MagicMock()
reg.list_all.return_value = []
with patch.object(app_module, 'service_registry', reg):
resp = self.client.get('/api/services/catalog')
self.assertEqual(resp.status_code, 200)
data = json.loads(resp.data)
self.assertIn('services', data)
def test_catalog_single_entry_returns_404_for_uninstalled(self):
"""GET /api/services/catalog/<id> returns 404 when service is not installed."""
import app as app_module
reg = MagicMock()
reg.get.return_value = None # simulates uninstalled
with patch.object(app_module, 'service_registry', reg):
resp = self.client.get('/api/services/catalog/email')
self.assertEqual(resp.status_code, 404)
data = json.loads(resp.data)
self.assertIn('error', data)
def test_catalog_single_entry_returns_200_when_installed(self):
"""GET /api/services/catalog/<id> returns 200 when service is installed."""
import app as app_module
email_svc = {**_store_manifest('email', 'mail', 'cell-rainloop:8888'), 'config': {}}
reg = MagicMock()
reg.get.return_value = email_svc
with patch.object(app_module, 'service_registry', reg):
resp = self.client.get('/api/services/catalog/email')
self.assertEqual(resp.status_code, 200)
# ---------------------------------------------------------------------------
# 9. migrate_legacy_containers()
# ---------------------------------------------------------------------------
class TestMigrateLegacyContainers(unittest.TestCase):
"""
migrate_legacy_containers() is a new helper that should be called on startup
to write install records for any of {email, calendar, files} whose containers
are already running but have no install record yet (upgrade path).
The method does not exist yet; these tests define its required contract.
When implemented, remove the @unittest.skip decorators.
Expected behaviour:
- For each legacy service whose container is running and has no install
record, call config_manager.set_installed_service with an appropriate
record derived from the legacy manifest.
- If the install record already exists, do not overwrite it (idempotent).
- Calling migrate_legacy_containers() twice must produce the same number
of set_installed_service calls as calling it once (idempotent on second call).
"""
def _make_ssm_for_migration(self, tmp, running_containers, installed=None):
"""
Build a ServiceStoreManager whose container_manager mock reports
the given running_containers list.
"""
cm = MagicMock()
# First call: before migration. Second call (idempotency): after migration.
installed_before = installed or {}
installed_after = dict(installed_before)
# Simulate that after migration the records are present
cm.get_installed_services.side_effect = [installed_before, installed_after]
cm.get_identity.return_value = {'ip_range': '172.20.0.0/16', 'service_ips': {}}
container_mgr = MagicMock()
container_mgr.list_containers.return_value = running_containers
caddy = MagicMock()
mgr = ServiceStoreManager(
config_manager=cm,
caddy_manager=caddy,
container_manager=container_mgr,
data_dir=tmp,
config_dir=tmp,
)
mgr.compose_override = os.path.join(tmp, 'docker-compose.services.yml')
return mgr
@unittest.skip(
'migrate_legacy_containers() not yet implemented; '
'add to ServiceStoreManager then unskip'
)
def test_migrate_writes_record_for_running_email_container(self):
"""A running cell-mail container with no install record gets an install record written."""
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_ssm_for_migration(
tmp,
running_containers=[
{'name': 'cell-mail', 'status': 'running'},
],
installed={},
)
with patch('service_registry._BUILTINS_DIR',
str(Path(__file__).parent.parent / 'api' / 'services' / 'builtins')):
mgr.migrate_legacy_containers()
mgr.config_manager.set_installed_service.assert_called()
call_service_ids = [
c[0][0] for c in mgr.config_manager.set_installed_service.call_args_list
]
self.assertIn('email', call_service_ids)
@unittest.skip(
'migrate_legacy_containers() not yet implemented; '
'add to ServiceStoreManager then unskip'
)
def test_migrate_does_not_overwrite_existing_record(self):
"""If email already has an install record, migrate must not overwrite it."""
with tempfile.TemporaryDirectory() as tmp:
existing_record = {'id': 'email', 'installed_at': '2026-01-01T00:00:00'}
mgr = self._make_ssm_for_migration(
tmp,
running_containers=[{'name': 'cell-mail', 'status': 'running'}],
installed={'email': existing_record},
)
with patch('service_registry._BUILTINS_DIR',
str(Path(__file__).parent.parent / 'api' / 'services' / 'builtins')):
mgr.migrate_legacy_containers()
mgr.config_manager.set_installed_service.assert_not_called()
@unittest.skip(
'migrate_legacy_containers() not yet implemented; '
'add to ServiceStoreManager then unskip'
)
def test_migrate_is_idempotent_on_second_call(self):
"""Calling migrate twice must not produce more set_installed_service calls than once."""
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_ssm_for_migration(
tmp,
running_containers=[{'name': 'cell-mail', 'status': 'running'}],
installed={},
)
# Simulate that after first migration the record is present
# by making get_installed_services return {} first, then {'email': {...}}
mgr.config_manager.get_installed_services.side_effect = [
{}, # first call inside first migrate
{'email': {}}, # second call inside second migrate
]
with patch('service_registry._BUILTINS_DIR',
str(Path(__file__).parent.parent / 'api' / 'services' / 'builtins')):
mgr.migrate_legacy_containers()
first_call_count = mgr.config_manager.set_installed_service.call_count
mgr.migrate_legacy_containers()
second_call_count = mgr.config_manager.set_installed_service.call_count
self.assertEqual(
first_call_count, second_call_count,
'Second migrate call must not write any additional install records',
)
@unittest.skip(
'migrate_legacy_containers() not yet implemented; '
'add to ServiceStoreManager then unskip'
)
def test_migrate_only_migrates_known_legacy_services(self):
"""Non-legacy containers (e.g. cell-caddy) must not receive install records."""
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_ssm_for_migration(
tmp,
running_containers=[
{'name': 'cell-caddy', 'status': 'running'},
{'name': 'cell-coredns', 'status': 'running'},
],
installed={},
)
with patch('service_registry._BUILTINS_DIR',
str(Path(__file__).parent.parent / 'api' / 'services' / 'builtins')):
mgr.migrate_legacy_containers()
mgr.config_manager.set_installed_service.assert_not_called()
# ---------------------------------------------------------------------------
# 10. Phase 2 completion: verify builtins layer is fully removed
# ---------------------------------------------------------------------------
class TestPhase2CompletionChecks(unittest.TestCase):
"""
Confirms that Phase 2 (builtins removal) is complete.
These tests verify the post-migration state: no builtins directory,
no hardcoded fallbacks, and registry-only routing for all services.
"""
def test_builtins_dir_does_not_exist(self):
"""api/services/builtins/ must not exist after Phase 2."""
import api.service_registry as sr_module
self.assertFalse(hasattr(sr_module, '_BUILTINS_DIR'),
'service_registry must not export _BUILTINS_DIR after Phase 2')
def test_list_all_empty_without_installed_services(self):
"""list_all() returns [] when nothing is installed."""
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
services = reg.list_all()
ids = [s['id'] for s in services]
self.assertNotIn('email', ids)
self.assertNotIn('calendar', ids)
self.assertNotIn('files', ids)
self.assertEqual(ids, [])
def test_get_caddy_routes_empty_without_installed_services(self):
"""get_caddy_routes() returns [] when nothing is installed."""
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
routes = reg.get_caddy_routes()
self.assertEqual(routes, [])
def test_backup_plan_empty_without_installed_services(self):
"""get_backup_plan() returns [] when nothing is installed."""
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
plan = reg.get_backup_plan()
self.assertEqual(plan, [])
def test_get_returns_none_for_uninstalled_service(self):
"""get('calendar') returns None when calendar is not installed."""
cm = MagicMock()
cm.configs = {'calendar': {}}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
result = reg.get('calendar')
self.assertIsNone(result)
def test_caddy_empty_registry_produces_only_api_block(self):
"""Empty registry → no service matcher blocks (no hardcoded fallback)."""
reg = MagicMock()
reg.get_caddy_routes.return_value = []
cm = MagicMock()
cm.get_identity.return_value = {}
mgr = CaddyManager(config_manager=cm, service_registry=reg)
result = mgr._build_registry_service_routes('alpha.pic.ngo')
self.assertIn('@api host api.alpha.pic.ngo', result)
self.assertNotIn('@calendar', result)
self.assertNotIn('@mail', result)
self.assertNotIn('@files', result)
if __name__ == '__main__':
unittest.main()
Reference in New Issue View Git Blame Copy Permalink