roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
18b50d08c1abd83790c47f7f2925c116aec20ece
pic/tests/test_optional_services_feature.py
T
roof 18b50d08c1
Unit Tests / test (push) Successful in 11m31s
Details
fix: post-Phase-0 corrections — data-dir bind mounts, reserved subdomains, list_active() 
Three related fixes discovered during review of Phase 0 and Phase 1 manifests:

1. validate_rendered_compose(): add allowed_data_dir param. After ${PIC_DATA_DIR}
   substitution, compose templates produce absolute paths; without this the
   validator would reject every service install.  ServiceComposer.write_compose()
   now passes its resolved data_dir so only the designated data directory is
   exempt — /etc, /proc, docker.sock etc. still blocked.

2. _RESERVED_SUBDOMAINS: remove service-level subdomains (mail, calendar, files,
   webdav, webmail). The reserved list should protect PIC infrastructure endpoints
   (api, webui, admin) — not service subdomains that official store services
   (calendar, files, webmail) must be allowed to claim.  Aligns with the
   existing _RESERVED_SUBS in service_registry.py.

3. ServiceRegistry.list_active(): new method returning only installed store
   services (no builtins). This is the forward-looking API that Phase 2 will
   make the primary read path once builtins are deleted. Adding it now unblocks
   the QA agent's test_optional_services_feature.py which was already testing
   the expected Phase 2 behaviour.

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

1058 lines
46 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 and no builtins on disk, list_active() is empty."""
reg = self._make_registry(installed={})
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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_in_builtins_and_not_installed(self):
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
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 no service matcher blocks — it falls back
to the hardcoded _build_core_service_routes.
The important test for this feature is that a registry returning [] from
get_caddy_routes produces no service blocks in a NEW install where
email/calendar/files have NOT been installed yet.
The existing fallback behaviour (empty → hardcoded) is already tested in
test_caddy_registry_integration.py:TestBuildRegistryServiceRoutes. These
new tests verify what happens when we pass a registry that explicitly
signals zero active services (e.g. all three were just uninstalled).
"""
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
when we override the fallback behaviour by returning hardcoded routes
only because registry is empty.
NOTE: the current implementation falls back to _build_core_service_routes
when the registry returns []. This test documents that existing behaviour.
When list_active() is wired in and builtins are removed, this test will
need updating to assert no service matchers appear. For now it pins the
fallback contract.
"""
mgr = self._mgr_with_empty_registry()
result = mgr._build_registry_service_routes('mycell.pic.ngo')
# Current contract: empty registry → hardcoded fallback is used
expected = CaddyManager._build_core_service_routes('mycell.pic.ngo')
self.assertEqual(result, expected)
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. Existing tests that will break when builtins are removed from disk
# (documented with the exact assertion that breaks)
# ---------------------------------------------------------------------------
class TestDocumentedBreakagePoints(unittest.TestCase):
"""
These tests do NOT fail right now because the builtin manifests are still
on disk. They are here to document which assertions in the existing test
suite will start failing the moment
api/services/builtins/{email,calendar,files}/manifest.json are deleted.
Each test runs the existing assertion in isolation so you can confirm it
fails after deletion by running this class with -v.
"""
def _load_builtin(self, service_id):
from service_registry import _BUILTINS_DIR
path = os.path.join(_BUILTINS_DIR, service_id, 'manifest.json')
if not os.path.exists(path):
self.skipTest(f'builtin manifest for {service_id!r} already removed')
with open(path) as f:
return json.load(f)
# --- test_service_registry.py::TestBuiltinManifests ---
def test_BREAKAGE_email_manifest_exists_on_disk(self):
"""
test_service_registry.py::TestBuiltinManifests::test_email_manifest_valid
BREAKS because _load('email') calls os.path.exists on the builtin path
and asserts True.
"""
self._load_builtin('email') # will raise AssertionError once file is deleted
def test_BREAKAGE_calendar_manifest_exists_on_disk(self):
"""test_calendar_manifest_valid breaks for the same reason."""
self._load_builtin('calendar')
def test_BREAKAGE_files_manifest_exists_on_disk(self):
"""test_files_manifest_valid breaks for the same reason."""
self._load_builtin('files')
# --- test_service_registry.py::TestServiceRegistryListAll ---
def test_BREAKAGE_list_all_returns_three_builtins(self):
"""
test_service_registry.py::TestServiceRegistryListAll::test_lists_three_builtins
asserts: assertIn('email', ids), assertIn('calendar', ids), assertIn('files', ids)
All three will fail when builtins are removed unless install records exist.
"""
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
services = reg.list_all()
ids = [s['id'] for s in services]
# Demonstrates the breakage: these assertions will fail
self.assertNotIn('email', ids,
'After builtin removal email must NOT appear in list_all without an install record')
self.assertNotIn('calendar', ids)
self.assertNotIn('files', ids)
# --- test_service_registry.py::TestServiceRegistryGetCaddyRoutes ---
def test_BREAKAGE_get_caddy_routes_empty_without_builtins(self):
"""
TestServiceRegistryGetCaddyRoutes::test_all_builtins_appear_in_routes
assertIn('email', route_ids) etc will all fail when builtins removed.
"""
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
routes = reg.get_caddy_routes()
route_ids = [r['service_id'] for r in routes]
self.assertEqual(route_ids, [],
'With no builtins on disk and nothing installed, routes must be empty')
# --- test_service_registry.py::TestServiceRegistryGetBackupPlan ---
def test_BREAKAGE_backup_plan_empty_without_builtins(self):
"""
TestServiceRegistryGetBackupPlan::test_all_builtins_in_backup_plan
will fail for all three service IDs.
"""
cm = MagicMock()
cm.configs = {}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
plan = reg.get_backup_plan()
self.assertEqual(plan, [],
'Backup plan must be empty when builtins removed and nothing installed')
# --- test_service_registry.py::TestServiceRegistryConfigMerge ---
def test_BREAKAGE_config_merge_returns_none_without_builtin(self):
"""
TestServiceRegistryConfigMerge::test_defaults_used_when_no_saved_config
calls reg.get('calendar') and asserts result['config']['port'] == 5232.
Once the calendar manifest is gone from disk, get('calendar') returns None
and None['config'] raises TypeError.
"""
cm = MagicMock()
cm.configs = {'calendar': {}}
cm.get_installed_services.return_value = {}
reg = ServiceRegistry(cm)
with patch('service_registry._BUILTINS_DIR', '/nonexistent/builtins'):
result = reg.get('calendar')
self.assertIsNone(result,
'get(calendar) must be None when builtin is removed and no install record exists')
# --- test_caddy_registry_integration.py — fallback to hardcoded ---
def test_BREAKAGE_caddy_with_empty_registry_falls_back_to_hardcoded(self):
"""
TestCaddyfileWithRegistry::test_pic_ngo_fallback_when_registry_empty
currently tests that an empty registry list falls back to hardcoded routes
which include calendar/mail/files.
When list_active() is wired in and builtins are gone, returning [] should
mean NO service routes at all — the fallback to hardcoded must also be removed.
The existing test assertion 'assertIn @calendar...' will then be WRONG.
This test documents the collision: currently the fallback is correct
behaviour; after the migration it becomes a bug.
"""
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')
# Documents current (pre-migration) contract:
# Empty registry → hardcoded fallback → calendar still appears
self.assertIn('@calendar host calendar.alpha.pic.ngo', result,
'Pre-migration: empty registry falls back to hardcoded; '
'this assertion must be INVERTED after the migration is complete')
if __name__ == '__main__':
unittest.main()
Reference in New Issue View Git Blame Copy Permalink