roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0bfe95320b62a6bdbe23234cc963c5bc69e184d2
pic/tests/test_optional_services_feature.py
T
roof 0bfe95320b
Unit Tests / test (push) Successful in 11m31s
Details
feat: Phase 2 — remove builtins layer, ServiceRegistry is installed-only 
Builtins (email/calendar/files) are no longer baked into the API image.
ServiceRegistry now only knows about installed store services. When nothing
is installed, Caddy and DNS get no service routes — no hardcoded fallback.

Changes:
- service_registry.py: remove _BUILTINS_DIR, _builtin_ids, _builtin_manifest,
  _load_manifest; get() and list_all() now delegate entirely to installed services
- caddy_manager.py: remove _build_core_service_routes(); remove hardcoded
  fallback pairs from _http01_service_pairs(); empty registry → api block only
- network_manager.py: _get_service_subdomains() returns [] when no registry
- api/services/builtins/: deleted (email, calendar, files manifests)
- Tests updated throughout: removed builtin-dependent assertions, added
  installed-service fixtures, updated fallback expectations to api-only

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

969 lines
41 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
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}:latest',
'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()
mgr = ServiceStoreManager(
config_manager=cm,
caddy_manager=caddy,
container_manager=container,
data_dir=tmp_dir,
config_dir=tmp_dir,
)
mgr.compose_override = os.path.join(tmp_dir, 'docker-compose.services.yml')
return mgr
class TestInstallHappyPath(unittest.TestCase):
def test_install_fetches_manifest_renders_compose_calls_docker_up(self):
"""install() happy path: fetches manifest, writes compose, calls docker compose up."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
manifest = _ssm_manifest('email')
mgr._fetch_manifest = MagicMock(return_value=manifest)
mgr._write_compose_override = MagicMock()
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('email')
self.assertTrue(result['ok'])
mgr._fetch_manifest.assert_called_once_with('email')
mgr.config_manager.set_installed_service.assert_called_once()
# docker compose up must have been called
self.assertTrue(mock_run.called)
docker_cmd = mock_run.call_args[0][0]
self.assertIn('up', docker_cmd)
self.assertIn('-d', docker_cmd)
def test_install_persists_install_record_before_docker_up(self):
"""Install record must be written to config before docker compose up is attempted."""
call_order = []
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
manifest = _ssm_manifest('calendar')
mgr._fetch_manifest = MagicMock(return_value=manifest)
mgr._write_compose_override = MagicMock()
mgr.config_manager.set_installed_service.side_effect = \
lambda *a, **kw: call_order.append('set_installed')
with patch('firewall_manager.apply_service_rules'), \
patch('service_store_manager.subprocess.run') as mock_run:
def _docker(*a, **kw):
call_order.append('docker_up')
return MagicMock(returncode=0, stderr='')
mock_run.side_effect = _docker
mgr.install('calendar')
self.assertLess(
call_order.index('set_installed'),
call_order.index('docker_up'),
'install record must be written before docker compose up',
)
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)
with patch('firewall_manager.apply_service_rules'):
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()
with patch('firewall_manager.apply_service_rules'):
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)
with patch('firewall_manager.apply_service_rules'):
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()
class TestInstallComposeUpFails(unittest.TestCase):
"""
The current implementation writes the install record BEFORE docker compose up.
When compose up fails the install record is already written — that is the
existing (accepted) behaviour documented in the implementation.
These tests verify the error is surfaced correctly rather than silently swallowed,
and that the install record IS present (not rolled back) after a compose failure.
"""
def test_install_compose_failure_is_logged_not_raised(self):
"""A non-zero exit from docker compose up must not raise — it is logged."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
manifest = _ssm_manifest('email')
mgr._fetch_manifest = MagicMock(return_value=manifest)
mgr._write_compose_override = MagicMock()
with patch('firewall_manager.apply_service_rules'), \
patch('service_store_manager.subprocess.run') as mock_run:
mock_run.return_value = MagicMock(
returncode=1, stderr='image pull failed'
)
# Must not raise
result = mgr.install('email')
# ok is still True because the record was persisted (compose is best-effort)
self.assertTrue(result['ok'])
def test_install_record_written_even_when_compose_fails(self):
"""Install record must exist after compose failure (compose is best-effort)."""
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp)
manifest = _ssm_manifest('email')
mgr._fetch_manifest = MagicMock(return_value=manifest)
mgr._write_compose_override = MagicMock()
with patch('firewall_manager.apply_service_rules'), \
patch('service_store_manager.subprocess.run') as mock_run:
mock_run.return_value = MagicMock(returncode=1, stderr='pull failed')
mgr.install('email')
mgr.config_manager.set_installed_service.assert_called_once()
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()
# ---------------------------------------------------------------------------
# 6. ServiceStoreManager.uninstall() (remove())
# ---------------------------------------------------------------------------
class TestUninstallHappyPath(unittest.TestCase):
def _make_mgr_with_email(self, tmp):
record = {
'container_name': 'cell-email',
'service_ip': '172.20.0.20',
'manifest': {
'image': 'git.pic.ngo/roof/email:1.0',
'volumes': [],
},
'iptables_rules': [],
}
installed = {'email': record}
mgr = _make_ssm(tmp, installed=installed)
mgr.config_manager.remove_installed_service = MagicMock()
mgr.config_manager.get_installed_services.side_effect = [
installed, # first call: existence check
{}, # second call: after removal, compose rewrite
]
mgr._write_compose_override = MagicMock()
return mgr
def test_uninstall_happy_path_returns_ok_true(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_mgr_with_email(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('email')
self.assertTrue(result['ok'])
def test_uninstall_removes_install_record(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_mgr_with_email(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('email')
mgr.config_manager.remove_installed_service.assert_called_once_with('email')
def test_uninstall_calls_docker_compose_stop_and_rm(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_mgr_with_email(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('email')
calls_str = [str(c) for c in mock_run.call_args_list]
has_stop = any('stop' in c for c in calls_str)
has_rm = any('rm' in c for c in calls_str)
self.assertTrue(has_stop, 'docker compose stop should have been called')
self.assertTrue(has_rm, 'docker rm should have been called')
def test_uninstall_regenerates_caddyfile(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = self._make_mgr_with_email(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('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={})
with patch('firewall_manager.clear_service_rules'):
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_docker(self):
with tempfile.TemporaryDirectory() as tmp:
mgr = _make_ssm(tmp, installed={})
with patch('firewall_manager.clear_service_rules'), \
patch('service_store_manager.subprocess.run') as mock_run:
mgr.remove('email')
mock_run.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()
with patch('firewall_manager.clear_service_rules'):
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