roof
1f016de855
Unit Tests / test (push) Successful in 11m35s
- ConfigManager.get_effective_domain(): returns domain_name when DDNS
active (pic_ngo/cloudflare/duckdns), domain otherwise. Used by all
public-facing services so they use the real registered FQDN.
- ConfigManager.get_internal_domain(): always returns _identity.domain
(CoreDNS zone name, dnsmasq, cell-link invites — stays internal).
- Silent migration: if domain_mode != lan and domain is generic "cell",
auto-set to {cell_name}.local for unique CoreDNS zone naming.
- caddy_manager: fix custom_domain bug — cloudflare/http01 modes were
reading identity.get('custom_domain') which never exists; now reads
domain_name correctly.
- routes/config, app: expose effective_domain in GET /api/config and
/api/status responses.
- email_manager, routes/email: use get_effective_domain() for
OVERRIDE_HOSTNAME, POSTMASTER_ADDRESS, and new-user email defaults.
- ServiceBus.IDENTITY_CHANGED event: emitted from PUT /api/config and
POST /api/ddns/register after identity writes; caddy_manager and
email_manager subscribe to regenerate config automatically.
- Settings.jsx: hide Local Domain input in non-LAN modes; show
read-only effective_domain with "managed by DDNS" badge and an
Advanced toggle for the internal CoreDNS zone name.
- 11 new test classes covering all new helpers, event subscriptions,
caddy/email handlers, and the custom_domain fix.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
198 lines
8.9 KiB
Python
198 lines
8.9 KiB
Python
import sys
|
|
from pathlib import Path
|
|
|
|
# Add api directory to path
|
|
api_dir = Path(__file__).parent.parent / 'api'
|
|
sys.path.insert(0, str(api_dir))
|
|
import unittest
|
|
import tempfile
|
|
import shutil
|
|
import os
|
|
from unittest.mock import patch, MagicMock
|
|
from email_manager import EmailManager
|
|
|
|
class TestEmailManager(unittest.TestCase):
|
|
def setUp(self):
|
|
self.test_dir = tempfile.mkdtemp()
|
|
self.data_dir = os.path.join(self.test_dir, 'data')
|
|
self.config_dir = os.path.join(self.test_dir, 'config')
|
|
os.makedirs(self.data_dir, exist_ok=True)
|
|
os.makedirs(self.config_dir, exist_ok=True)
|
|
self.manager = EmailManager(data_dir=self.data_dir, config_dir=self.config_dir)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
def test_initialization(self):
|
|
self.assertTrue(os.path.exists(self.manager.email_dir))
|
|
self.assertTrue(os.path.exists(self.manager.postfix_dir))
|
|
self.assertTrue(os.path.exists(self.manager.dovecot_dir))
|
|
|
|
@patch.object(EmailManager, '_reload_email_services', return_value=True)
|
|
def test_create_and_delete_email_user(self, mock_reload):
|
|
result = self.manager.create_email_user('testuser', 'testdomain', 'password')
|
|
self.assertTrue(result)
|
|
result = self.manager.delete_email_user('testuser', 'testdomain')
|
|
self.assertTrue(result)
|
|
|
|
def test_list_email_users_empty(self):
|
|
users = self.manager.list_email_users()
|
|
self.assertIsInstance(users, list)
|
|
self.assertEqual(len(users), 0)
|
|
|
|
@patch('smtplib.SMTP')
|
|
def test_send_email(self, mock_smtp):
|
|
instance = mock_smtp.return_value.__enter__.return_value
|
|
instance.sendmail.return_value = {}
|
|
result = self.manager.send_email('from@cell', 'to@cell', 'Subject', 'Body')
|
|
self.assertTrue(result)
|
|
# Simulate error by raising in SMTP constructor
|
|
mock_smtp.side_effect = Exception('SMTP error')
|
|
result = self.manager.send_email('from@cell', 'to@cell', 'Subject', 'Body')
|
|
self.assertFalse(result)
|
|
|
|
@patch('imaplib.IMAP4_SSL')
|
|
def test_get_mailbox_info(self, mock_imap):
|
|
instance = mock_imap.return_value
|
|
instance.login.return_value = 'OK'
|
|
instance.select.return_value = ('OK', [b'1'])
|
|
instance.search.return_value = ('OK', [b'1 2 3'])
|
|
instance.fetch.return_value = ('OK', [(b'1', b'RFC822')])
|
|
info = self.manager.get_mailbox_info('testuser', 'testdomain')
|
|
self.assertIsInstance(info, dict)
|
|
instance.login.side_effect = Exception('IMAP error')
|
|
info = self.manager.get_mailbox_info('testuser', 'testdomain')
|
|
self.assertIn('error', info)
|
|
|
|
@patch('subprocess.run')
|
|
def test_get_email_status(self, mock_run):
|
|
mock_run.return_value.stdout = 'cell-mail\n'
|
|
mock_run.return_value.returncode = 0
|
|
status = self.manager.get_email_status()
|
|
self.assertIsInstance(status, dict)
|
|
self.assertIn('postfix_running', status)
|
|
self.assertIn('dovecot_running', status)
|
|
|
|
@patch('requests.get')
|
|
def test_test_email_connectivity(self, mock_get):
|
|
mock_get.return_value.status_code = 200
|
|
result = self.manager.test_email_connectivity()
|
|
self.assertIsInstance(result, dict)
|
|
mock_get.side_effect = Exception('HTTP error')
|
|
result = self.manager.test_email_connectivity()
|
|
self.assertIn('smtp', result)
|
|
|
|
@patch('subprocess.run')
|
|
def test_get_email_logs(self, mock_run):
|
|
mock_run.return_value.stdout = 'log line\n'
|
|
mock_run.return_value.returncode = 0
|
|
logs = self.manager.get_email_logs('all', 10)
|
|
self.assertIsInstance(logs, dict)
|
|
self.assertIn('postfix', logs)
|
|
self.assertIn('dovecot', logs)
|
|
|
|
def test_get_status(self):
|
|
status = self.manager.get_status()
|
|
self.assertIsInstance(status, dict)
|
|
self.assertIn('status', status)
|
|
|
|
def test_error_handling(self):
|
|
# Force errors by passing invalid arguments, should return False or error dict
|
|
self.assertFalse(self.manager.create_email_user(None, None, None))
|
|
self.assertFalse(self.manager.delete_email_user(None, None))
|
|
self.assertFalse(self.manager.send_email(None, None, None, None))
|
|
info = self.manager.get_mailbox_info(None, None)
|
|
self.assertIn('error', info)
|
|
|
|
class TestEmailManagerEffectiveDomain(unittest.TestCase):
|
|
"""Verify that email OVERRIDE_HOSTNAME and POSTMASTER_ADDRESS use the
|
|
caller-supplied domain (which should come from get_effective_domain in the
|
|
route layer when no explicit domain is provided by the client)."""
|
|
|
|
def setUp(self):
|
|
self.test_dir = tempfile.mkdtemp()
|
|
self.data_dir = os.path.join(self.test_dir, 'data')
|
|
self.config_dir = os.path.join(self.test_dir, 'config')
|
|
os.makedirs(os.path.join(self.config_dir, 'mail'), exist_ok=True)
|
|
os.makedirs(os.path.join(self.data_dir, 'email'), exist_ok=True)
|
|
with open(os.path.join(self.config_dir, 'mail', 'mailserver.env'), 'w') as f:
|
|
f.write('OVERRIDE_HOSTNAME=mail.cell\nPOSTMASTER_ADDRESS=admin@cell\n')
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
|
|
from email_manager import EmailManager
|
|
self.em = EmailManager(data_dir=self.data_dir, config_dir=self.config_dir)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
@patch('subprocess.run')
|
|
def test_email_hostname_uses_effective_domain_in_ddns_mode(self, mock_run):
|
|
"""When apply_config is called with domain='home.pic.ngo' (as provided
|
|
by the route layer via get_effective_domain), OVERRIDE_HOSTNAME and
|
|
POSTMASTER_ADDRESS should use 'home.pic.ngo', not the internal 'cell'."""
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = self.em.apply_config({'domain': 'home.pic.ngo'})
|
|
env = open(os.path.join(self.config_dir, 'mail', 'mailserver.env')).read()
|
|
self.assertIn('OVERRIDE_HOSTNAME=mail.home.pic.ngo', env)
|
|
self.assertIn('POSTMASTER_ADDRESS=admin@home.pic.ngo', env)
|
|
self.assertIn('cell-mail', result['restarted'])
|
|
|
|
|
|
class TestEmailManagerIdentityChangedSubscription(unittest.TestCase):
|
|
def setUp(self):
|
|
self.test_dir = tempfile.mkdtemp()
|
|
self.data_dir = os.path.join(self.test_dir, 'data')
|
|
self.config_dir = os.path.join(self.test_dir, 'config')
|
|
os.makedirs(self.data_dir, exist_ok=True)
|
|
os.makedirs(self.config_dir, exist_ok=True)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
def test_subscribes_to_identity_changed_on_init(self):
|
|
"""When service_bus is provided, __init__ subscribes to IDENTITY_CHANGED."""
|
|
from service_bus import EventType
|
|
mock_bus = MagicMock()
|
|
manager = EmailManager(
|
|
data_dir=self.data_dir,
|
|
config_dir=self.config_dir,
|
|
service_bus=mock_bus,
|
|
)
|
|
mock_bus.subscribe_to_event.assert_called_once_with(
|
|
EventType.IDENTITY_CHANGED, manager._on_identity_changed
|
|
)
|
|
|
|
def test_no_subscription_without_service_bus(self):
|
|
"""When service_bus is not provided, no subscription is attempted."""
|
|
mock_bus = MagicMock()
|
|
EmailManager(data_dir=self.data_dir, config_dir=self.config_dir)
|
|
mock_bus.subscribe_to_event.assert_not_called()
|
|
|
|
@patch.object(EmailManager, 'apply_config', return_value={'restarted': [], 'warnings': []})
|
|
def test_on_identity_changed_calls_apply_config(self, mock_apply):
|
|
"""_on_identity_changed calls apply_config with the effective_domain."""
|
|
manager = EmailManager(data_dir=self.data_dir, config_dir=self.config_dir)
|
|
event = MagicMock()
|
|
event.data = {'effective_domain': 'mycell.pic.ngo'}
|
|
manager._on_identity_changed(event)
|
|
mock_apply.assert_called_once_with({'domain': 'mycell.pic.ngo'})
|
|
|
|
@patch.object(EmailManager, 'apply_config', side_effect=Exception('boom'))
|
|
def test_on_identity_changed_swallows_exceptions(self, mock_apply):
|
|
"""_on_identity_changed must not propagate exceptions."""
|
|
manager = EmailManager(data_dir=self.data_dir, config_dir=self.config_dir)
|
|
event = MagicMock()
|
|
event.data = {'effective_domain': 'mycell.pic.ngo'}
|
|
manager._on_identity_changed(event) # must not raise
|
|
|
|
def test_on_identity_changed_skips_when_no_effective_domain(self):
|
|
"""_on_identity_changed does nothing when effective_domain is absent."""
|
|
manager = EmailManager(data_dir=self.data_dir, config_dir=self.config_dir)
|
|
event = MagicMock()
|
|
event.data = {'cell_name': 'mycell'}
|
|
with patch.object(manager, 'apply_config') as mock_apply:
|
|
manager._on_identity_changed(event)
|
|
mock_apply.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |