roof
a906c26b5d
Unit Tests / test (push) Successful in 11m25s
acme_ca and the pic_ngo DNS credentials ({$PIC_NGO_DDNS_TOKEN},
{$PIC_NGO_DDNS_API}) were written as Caddy env-var placeholders, but the
Caddy container does not inherit the API container's environment, so the
substitutions always failed — Caddy saw bare directive names with no
arguments and rejected the Caddyfile.
- _global_acme_block: only emit the acme_ca directive when ACME_CA_URL is
actually set; omitting it makes Caddy default to Let's Encrypt production.
- _caddyfile_pic_ngo: embed the DDNS_TOTP_SECRET and DDNS_URL values directly
into the Caddyfile at write time rather than relying on Caddy env expansion.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
330 lines
13 KiB
Python
330 lines
13 KiB
Python
"""Tests for CaddyManager — Caddyfile generation per domain mode plus
|
|
admin-API reload, health check, and consecutive-failure bookkeeping.
|
|
"""
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import requests
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'api'))
|
|
|
|
from caddy_manager import CaddyManager # noqa: E402
|
|
|
|
|
|
def _mgr(tmpdir=None, identity=None):
|
|
"""Build a CaddyManager backed by a mock config_manager."""
|
|
cm = MagicMock()
|
|
cm.get_identity.return_value = identity or {}
|
|
mgr = CaddyManager(
|
|
config_manager=cm,
|
|
data_dir=tmpdir or '/tmp/pic-test-data',
|
|
config_dir=tmpdir or '/tmp/pic-test-config',
|
|
)
|
|
return mgr
|
|
|
|
|
|
CALENDAR_ROUTE = (
|
|
"handle /calendar* {\n"
|
|
" reverse_proxy cell-radicale:5232\n"
|
|
"}"
|
|
)
|
|
FILES_ROUTE = (
|
|
"handle /files* {\n"
|
|
" reverse_proxy cell-filegator:8080\n"
|
|
"}"
|
|
)
|
|
|
|
|
|
class TestGenerateCaddyfileLan(unittest.TestCase):
|
|
def test_lan_mode_has_auto_https_off_and_no_acme(self):
|
|
mgr = _mgr()
|
|
identity = {'cell_name': 'mycell', 'domain_mode': 'lan'}
|
|
out = mgr.generate_caddyfile(identity, [])
|
|
self.assertIn('auto_https off', out)
|
|
# No ACME anywhere
|
|
self.assertNotIn('acme_ca', out)
|
|
self.assertNotIn('acme_email', out)
|
|
self.assertNotIn('dns pic_ngo', out)
|
|
self.assertNotIn('dns cloudflare', out)
|
|
# Internal-CA TLS pair
|
|
self.assertIn('tls /etc/caddy/internal/cert.pem '
|
|
'/etc/caddy/internal/key.pem', out)
|
|
# Cell hostname plus virtual IP listener
|
|
self.assertIn('http://mycell.cell', out)
|
|
self.assertIn('http://172.20.0.2:80', out)
|
|
|
|
|
|
class TestGenerateCaddyfilePicNgo(unittest.TestCase):
|
|
def test_pic_ngo_has_dns_plugin_and_wildcard(self):
|
|
mgr = _mgr()
|
|
identity = {'cell_name': 'alpha', 'domain_mode': 'pic_ngo'}
|
|
import os
|
|
with unittest.mock.patch.dict(os.environ, {
|
|
'DDNS_TOTP_SECRET': 'TESTSECRET123',
|
|
'DDNS_URL': 'https://ddns.pic.ngo/api/v1',
|
|
}):
|
|
out = mgr.generate_caddyfile(identity, [])
|
|
self.assertIn('dns pic_ngo', out)
|
|
self.assertIn('*.alpha.pic.ngo', out)
|
|
self.assertIn('alpha.pic.ngo', out)
|
|
# Credentials are resolved at write time and embedded — no {$VAR} placeholders
|
|
self.assertIn('token TESTSECRET123', out)
|
|
self.assertIn('api_base_url https://ddns.pic.ngo/api/v1', out)
|
|
self.assertNotIn('{$PIC_NGO_DDNS_TOKEN}', out)
|
|
self.assertNotIn('{$PIC_NGO_DDNS_API}', out)
|
|
self.assertIn('email admin@alpha.pic.ngo', out)
|
|
# acme_ca is omitted when ACME_CA_URL is not set
|
|
self.assertNotIn('acme_ca', out)
|
|
|
|
def test_pic_ngo_acme_ca_included_when_env_set(self):
|
|
mgr = _mgr()
|
|
identity = {'cell_name': 'alpha', 'domain_mode': 'pic_ngo'}
|
|
import os
|
|
with unittest.mock.patch.dict(os.environ, {
|
|
'DDNS_TOTP_SECRET': 'TESTSECRET123',
|
|
'DDNS_URL': 'https://ddns.pic.ngo/api/v1',
|
|
'ACME_CA_URL': 'https://acme-staging-v02.api.letsencrypt.org/directory',
|
|
}):
|
|
out = mgr.generate_caddyfile(identity, [])
|
|
self.assertIn('acme_ca https://acme-staging-v02.api.letsencrypt.org/directory', out)
|
|
|
|
def test_pic_ngo_has_api_route_without_registry(self):
|
|
mgr = _mgr()
|
|
identity = {'cell_name': 'alpha', 'domain_mode': 'pic_ngo'}
|
|
out = mgr.generate_caddyfile(identity, [])
|
|
# Without a registry only the api block is present
|
|
self.assertIn('@api host api.alpha.pic.ngo', out)
|
|
self.assertIn('reverse_proxy cell-api:3000', out)
|
|
self.assertNotIn('@calendar', out)
|
|
self.assertNotIn('@mail', out)
|
|
self.assertNotIn('@files', out)
|
|
|
|
|
|
class TestGenerateCaddyfileCloudflare(unittest.TestCase):
|
|
def test_cloudflare_has_dns_cloudflare(self):
|
|
mgr = _mgr()
|
|
identity = {
|
|
'cell_name': 'beta',
|
|
'domain_mode': 'cloudflare',
|
|
'domain_name': 'example.com',
|
|
}
|
|
out = mgr.generate_caddyfile(identity, [])
|
|
self.assertIn('dns cloudflare {$CF_API_TOKEN}', out)
|
|
self.assertIn('*.example.com', out)
|
|
self.assertIn('email {$ACME_EMAIL}', out)
|
|
# acme_ca is omitted when ACME_CA_URL is not set in the environment
|
|
self.assertNotIn('acme_ca', out)
|
|
|
|
def test_caddyfile_cloudflare_uses_domain_name(self):
|
|
"""Caddyfile must use domain_name for TLS host, not any 'custom_domain' key."""
|
|
mgr = _mgr()
|
|
identity = {
|
|
'cell_name': 'beta',
|
|
'domain_mode': 'cloudflare',
|
|
'domain_name': 'home.example.com',
|
|
'domain': 'home.local',
|
|
}
|
|
out = mgr.generate_caddyfile(identity, [])
|
|
self.assertIn('*.home.example.com', out)
|
|
self.assertIn('home.example.com', out)
|
|
# Must not use the internal domain for TLS
|
|
self.assertNotIn('*.home.local', out)
|
|
# 'custom_domain' must not appear literally as a key in the output
|
|
self.assertNotIn('custom_domain', out)
|
|
# Without a registry only the api block is emitted for subdomain routing
|
|
self.assertIn('@api host api.home.example.com', out)
|
|
self.assertNotIn('@calendar', out)
|
|
self.assertNotIn('@files', out)
|
|
|
|
|
|
class TestGenerateCaddyfileDuckDns(unittest.TestCase):
|
|
def test_duckdns_has_dns_duckdns(self):
|
|
mgr = _mgr()
|
|
identity = {'cell_name': 'gamma', 'domain_mode': 'duckdns'}
|
|
out = mgr.generate_caddyfile(identity, [])
|
|
self.assertIn('dns duckdns {$DUCKDNS_TOKEN}', out)
|
|
self.assertIn('*.gamma.duckdns.org', out)
|
|
self.assertIn('@api host api.gamma.duckdns.org', out)
|
|
self.assertNotIn('@calendar', out)
|
|
self.assertNotIn('@files', out)
|
|
|
|
|
|
class TestGenerateCaddyfileHttp01(unittest.TestCase):
|
|
def test_http01_no_tls_block_and_per_service_blocks(self):
|
|
mgr = _mgr()
|
|
identity = {
|
|
'cell_name': 'delta',
|
|
'domain_mode': 'http01',
|
|
'domain_name': 'delta.noip.me',
|
|
}
|
|
# Store-plugin service (not a core service name)
|
|
services = [
|
|
{'name': 'chat', 'caddy_route': 'reverse_proxy cell-chat:8090'},
|
|
]
|
|
out = mgr.generate_caddyfile(identity, services)
|
|
# No wildcard, no DNS-01 plugins.
|
|
self.assertNotIn('*.delta', out)
|
|
self.assertNotIn('dns ', out)
|
|
# No explicit tls block — Caddy uses HTTP-01 by default.
|
|
self.assertNotIn('tls {', out)
|
|
# Without a registry only the api block is generated
|
|
self.assertIn('api.delta.noip.me {', out)
|
|
self.assertNotIn('calendar.delta.noip.me {', out)
|
|
self.assertNotIn('files.delta.noip.me {', out)
|
|
self.assertNotIn('mail.delta.noip.me {', out)
|
|
# Installed plugin service block still works
|
|
self.assertIn('chat.delta.noip.me {', out)
|
|
self.assertIn('reverse_proxy cell-chat:8090', out)
|
|
|
|
def test_http01_installed_service_with_caddy_route_appears(self):
|
|
"""An installed service with a caddy_route produces its own per-host block."""
|
|
mgr = _mgr()
|
|
identity = {
|
|
'cell_name': 'delta',
|
|
'domain_mode': 'http01',
|
|
'domain_name': 'delta.noip.me',
|
|
}
|
|
services = [{'name': 'notes', 'caddy_route': 'reverse_proxy cell-other:9000'}]
|
|
out = mgr.generate_caddyfile(identity, services)
|
|
self.assertIn('notes.delta.noip.me {', out)
|
|
self.assertIn('reverse_proxy cell-other:9000', out)
|
|
|
|
|
|
class TestServiceRoutesIncluded(unittest.TestCase):
|
|
def test_installed_service_route_appears_in_output(self):
|
|
mgr = _mgr()
|
|
identity = {'cell_name': 'eps', 'domain_mode': 'lan'}
|
|
services = [
|
|
{'name': 'calendar', 'caddy_route': CALENDAR_ROUTE},
|
|
{'name': 'files', 'caddy_route': FILES_ROUTE},
|
|
]
|
|
out = mgr.generate_caddyfile(identity, services)
|
|
self.assertIn('handle /calendar*', out)
|
|
self.assertIn('reverse_proxy cell-radicale:5232', out)
|
|
self.assertIn('handle /files*', out)
|
|
self.assertIn('reverse_proxy cell-filegator:8080', out)
|
|
# Core routes still emitted
|
|
self.assertIn('reverse_proxy cell-api:3000', out)
|
|
self.assertIn('reverse_proxy cell-webui:80', out)
|
|
|
|
|
|
class TestReloadCaddyAdminAPI(unittest.TestCase):
|
|
def test_reload_calls_admin_api_load_endpoint(self):
|
|
mgr = _mgr()
|
|
# Point at a tmp Caddyfile so we can read it back during reload.
|
|
import tempfile
|
|
tmp = tempfile.NamedTemporaryFile('w', delete=False, suffix='.caddyfile')
|
|
tmp.write(":80 { reverse_proxy cell-webui:80 }\n")
|
|
tmp.close()
|
|
mgr.caddyfile_path = tmp.name
|
|
|
|
with patch('caddy_manager.requests.post') as mock_post:
|
|
mock_post.return_value = MagicMock(status_code=200, text='ok')
|
|
ok = mgr.reload_caddy()
|
|
|
|
self.assertTrue(ok)
|
|
mock_post.assert_called_once()
|
|
args, kwargs = mock_post.call_args
|
|
# First positional arg is the URL
|
|
self.assertEqual(args[0], 'http://cell-caddy:2019/load')
|
|
self.assertEqual(kwargs['headers']['Content-Type'], 'text/caddyfile')
|
|
self.assertIn('cell-webui:80', kwargs['data'])
|
|
os.unlink(tmp.name)
|
|
|
|
|
|
class TestHealthCheck(unittest.TestCase):
|
|
def test_returns_true_on_200(self):
|
|
mgr = _mgr()
|
|
with patch('caddy_manager.requests.get') as mock_get:
|
|
mock_get.return_value = MagicMock(status_code=200)
|
|
self.assertTrue(mgr.check_caddy_health())
|
|
mock_get.assert_called_once()
|
|
# URL must be the admin API root
|
|
self.assertIn('cell-caddy:2019', mock_get.call_args[0][0])
|
|
|
|
def test_returns_false_on_connection_error(self):
|
|
mgr = _mgr()
|
|
with patch('caddy_manager.requests.get',
|
|
side_effect=requests.ConnectionError('refused')):
|
|
self.assertFalse(mgr.check_caddy_health())
|
|
|
|
def test_returns_false_on_non_200(self):
|
|
mgr = _mgr()
|
|
with patch('caddy_manager.requests.get') as mock_get:
|
|
mock_get.return_value = MagicMock(status_code=500)
|
|
self.assertFalse(mgr.check_caddy_health())
|
|
|
|
|
|
class TestFailureCounter(unittest.TestCase):
|
|
def test_increments_and_resets(self):
|
|
mgr = _mgr()
|
|
self.assertEqual(mgr.get_health_failure_count(), 0)
|
|
self.assertEqual(mgr.increment_health_failure(), 1)
|
|
self.assertEqual(mgr.increment_health_failure(), 2)
|
|
self.assertEqual(mgr.increment_health_failure(), 3)
|
|
self.assertEqual(mgr.get_health_failure_count(), 3)
|
|
mgr.reset_health_failures()
|
|
self.assertEqual(mgr.get_health_failure_count(), 0)
|
|
|
|
|
|
class TestCertStatus(unittest.TestCase):
|
|
def test_returns_default_when_no_tls_in_identity(self):
|
|
mgr = _mgr(identity={'cell_name': 'x', 'domain_mode': 'lan'})
|
|
out = mgr.get_cert_status()
|
|
self.assertEqual(out['status'], 'unknown')
|
|
self.assertIsNone(out['expiry'])
|
|
self.assertIsNone(out['days_remaining'])
|
|
|
|
def test_returns_tls_block_when_present(self):
|
|
mgr = _mgr(identity={
|
|
'cell_name': 'x',
|
|
'domain_mode': 'pic_ngo',
|
|
'tls': {
|
|
'status': 'valid',
|
|
'expiry': '2026-08-01T00:00:00Z',
|
|
'days_remaining': 84,
|
|
},
|
|
})
|
|
out = mgr.get_cert_status()
|
|
self.assertEqual(out['status'], 'valid')
|
|
self.assertEqual(out['expiry'], '2026-08-01T00:00:00Z')
|
|
self.assertEqual(out['days_remaining'], 84)
|
|
|
|
|
|
class TestCaddyManagerIdentityChangedSubscription(unittest.TestCase):
|
|
def test_subscribes_to_identity_changed_on_init(self):
|
|
"""When service_bus is provided, CaddyManager subscribes to IDENTITY_CHANGED."""
|
|
from service_bus import EventType
|
|
mock_bus = MagicMock()
|
|
mgr = CaddyManager(config_manager=MagicMock(), service_bus=mock_bus)
|
|
mock_bus.subscribe_to_event.assert_called_once_with(
|
|
EventType.IDENTITY_CHANGED, mgr._on_identity_changed
|
|
)
|
|
|
|
def test_no_subscription_without_service_bus(self):
|
|
"""When service_bus is omitted, no subscription is attempted."""
|
|
mock_bus = MagicMock()
|
|
CaddyManager(config_manager=MagicMock())
|
|
mock_bus.subscribe_to_event.assert_not_called()
|
|
|
|
def test_on_identity_changed_calls_regenerate_with_installed(self):
|
|
"""_on_identity_changed calls regenerate_with_installed([])."""
|
|
mgr = _mgr()
|
|
with patch.object(mgr, 'regenerate_with_installed', return_value=True) as mock_regen:
|
|
event = MagicMock()
|
|
mgr._on_identity_changed(event)
|
|
mock_regen.assert_called_once_with([])
|
|
|
|
def test_on_identity_changed_swallows_exceptions(self):
|
|
"""_on_identity_changed must not propagate exceptions."""
|
|
mgr = _mgr()
|
|
with patch.object(mgr, 'regenerate_with_installed', side_effect=Exception('boom')):
|
|
event = MagicMock()
|
|
mgr._on_identity_changed(event) # must not raise
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|