#!/usr/bin/env python3 """ Additional tests for NetworkManager covering apply_domain, apply_cell_name, get_status, get_dns_status, get_network_info, apply_config, and input-validation paths not covered by the main test file. """ import sys import os import json import shutil import tempfile import unittest from pathlib import Path from unittest.mock import patch, MagicMock, call api_dir = Path(__file__).parent.parent / 'api' sys.path.insert(0, str(api_dir)) from network_manager import NetworkManager def _make_nm(tmp): data_dir = os.path.join(tmp, 'data') config_dir = os.path.join(tmp, 'config') os.makedirs(data_dir, exist_ok=True) os.makedirs(config_dir, exist_ok=True) return NetworkManager(data_dir, config_dir), data_dir, config_dir class TestApplyDomain(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) def tearDown(self): shutil.rmtree(self.tmp) @patch('subprocess.run') def test_apply_domain_no_config_files_returns_empty_warnings(self, _mock): """apply_domain with no corefile or zone files is graceful.""" result = self.nm.apply_domain('newcell', reload=False) self.assertIsInstance(result, dict) self.assertIn('restarted', result) self.assertIn('warnings', result) @patch('subprocess.run') def test_apply_domain_renames_and_rewrites_zone_file(self, _mock): dns_data = os.path.join(self.data_dir, 'dns') os.makedirs(dns_data, exist_ok=True) zone_content = """$TTL 3600 @ IN SOA oldcell. admin.oldcell. ( 2026010101 ; Serial 3600 1800 1209600 3600 ) @ IN NS oldcell. api 3600 IN A 10.0.0.1 """ with open(os.path.join(dns_data, 'oldcell.zone'), 'w') as f: f.write(zone_content) result = self.nm.apply_domain('newcell', reload=False) new_zone = os.path.join(dns_data, 'newcell.zone') self.assertTrue(os.path.exists(new_zone)) with open(new_zone) as f: written = f.read() self.assertIn('newcell.', written) @patch('subprocess.run') def test_apply_domain_reloads_dns_when_reload_true(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='') result = self.nm.apply_domain('newcell', reload=True) calls_str = str(mock_run.call_args_list) self.assertIn('SIGUSR1', calls_str) @patch('subprocess.run') def test_apply_domain_does_not_reload_when_reload_false(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='') result = self.nm.apply_domain('newcell', reload=False) calls_str = str(mock_run.call_args_list) self.assertNotIn('SIGUSR1', calls_str) class TestApplyCellName(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) self.dns_data = os.path.join(self.data_dir, 'dns') os.makedirs(self.dns_data, exist_ok=True) def tearDown(self): shutil.rmtree(self.tmp) def _write_zone(self, name, content): with open(os.path.join(self.dns_data, f'{name}.zone'), 'w') as f: f.write(content) @patch('subprocess.run') def test_apply_cell_name_updates_hostname_record(self, _mock): self._write_zone('cell', ( 'oldname 3600 IN A 10.0.0.1\n' 'api 3600 IN A 10.0.0.1\n' )) result = self.nm.apply_cell_name('oldname', 'newname') with open(os.path.join(self.dns_data, 'cell.zone')) as f: content = f.read() self.assertIn('newname', content) self.assertNotIn('oldname', content) @patch('subprocess.run') def test_apply_cell_name_empty_new_name_does_nothing(self, mock_run): result = self.nm.apply_cell_name('oldname', '') mock_run.assert_not_called() @patch('subprocess.run') def test_apply_cell_name_already_correct_does_not_write(self, mock_run): self._write_zone('cell', ( 'mycel 3600 IN A 10.0.0.1\n' 'api 3600 IN A 10.0.0.1\n' )) result = self.nm.apply_cell_name('mycel', 'mycel') # No reload should happen since name already matches calls_str = str(mock_run.call_args_list) self.assertNotIn('SIGUSR1', calls_str) @patch('subprocess.run') def test_apply_cell_name_detects_hostname_when_old_name_absent(self, _mock): """If old_name not in zone, detect hostname by non-service A record.""" self._write_zone('cell', ( 'detectedhost 3600 IN A 10.0.0.1\n' 'calendar 3600 IN A 10.0.0.1\n' )) result = self.nm.apply_cell_name('', 'newname') with open(os.path.join(self.dns_data, 'cell.zone')) as f: content = f.read() self.assertIn('newname', content) self.assertNotIn('detectedhost', content) @patch('subprocess.run') def test_apply_cell_name_skips_multi_label_zones(self, _mock): """Multi-label zones (e.g. pic2.pic.ngo) must not be modified.""" self._write_zone('cell', 'oldname 3600 IN A 10.0.0.1\n') self._write_zone('pic2.pic.ngo', 'api 3600 IN A 10.0.0.1\n') result = self.nm.apply_cell_name('oldname', 'newname') with open(os.path.join(self.dns_data, 'pic2.pic.ngo.zone')) as f: multi = f.read() # multi-label zone should be unchanged self.assertNotIn('newname', multi) class TestApplyConfig(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) def tearDown(self): shutil.rmtree(self.tmp) @patch('subprocess.run') def test_apply_config_empty_dict_returns_no_warnings(self, _mock): result = self.nm.apply_config({}) self.assertEqual(result['warnings'], []) @patch('subprocess.run') def test_apply_config_ntp_servers_updates_file(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='') ntp_dir = os.path.join(self.config_dir, 'ntp') os.makedirs(ntp_dir, exist_ok=True) ntp_conf = os.path.join(ntp_dir, 'chrony.conf') with open(ntp_conf, 'w') as f: f.write('server 0.pool.ntp.org iburst\nmakestep 1.0 3\n') result = self.nm.apply_config({'ntp_servers': ['1.1.1.1', '8.8.8.8']}) with open(ntp_conf) as f: content = f.read() self.assertIn('server 1.1.1.1 iburst', content) self.assertIn('server 8.8.8.8 iburst', content) self.assertNotIn('0.pool.ntp.org', content) class TestGetDnsStatus(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) def tearDown(self): shutil.rmtree(self.tmp) @patch('subprocess.run') def test_get_dns_status_returns_dict(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='cell-dns\n', stderr='') result = self.nm.get_dns_status() self.assertIn('running', result) self.assertIn('records_count', result) @patch('subprocess.run') def test_get_dns_status_running_true_when_container_up(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='cell-dns\n', stderr='') result = self.nm.get_dns_status() self.assertTrue(result['running']) @patch('subprocess.run') def test_get_dns_status_counts_records(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='cell-dns\n', stderr='') dns_data = os.path.join(self.data_dir, 'dns') os.makedirs(dns_data, exist_ok=True) with open(os.path.join(dns_data, 'cell.zone'), 'w') as f: f.write('api 3600 IN A 10.0.0.1\nwebui 3600 IN A 10.0.0.1\n') result = self.nm.get_dns_status() self.assertEqual(result['records_count'], 2) @patch('subprocess.run') def test_get_dns_status_not_running(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='') result = self.nm.get_dns_status() self.assertFalse(result['running']) class TestGetNetworkInfo(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) def tearDown(self): shutil.rmtree(self.tmp) @patch('subprocess.run') def test_get_network_info_returns_dict(self, mock_run): mock_run.return_value = MagicMock(returncode=1, stdout='', stderr='') result = self.nm.get_network_info() self.assertIsInstance(result, dict) @patch('subprocess.run') def test_get_network_info_includes_dns_servers(self, mock_run): mock_run.return_value = MagicMock(returncode=1, stdout='', stderr='') result = self.nm.get_network_info() self.assertIn('dns_servers', result) @patch('subprocess.run') def test_get_network_info_parses_interfaces_json(self, mock_run): iface_json = '[{"ifindex":1,"ifname":"lo","addr_info":[]}]' mock_run.return_value = MagicMock(returncode=0, stdout=iface_json, stderr='') result = self.nm.get_network_info() self.assertIn('interfaces', result) self.assertEqual(result['interfaces'][0]['ifname'], 'lo') class TestGetStatus(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) def tearDown(self): shutil.rmtree(self.tmp) @patch('subprocess.run') def test_get_status_returns_dict_with_required_keys(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='') result = self.nm.get_status() self.assertIn('running', result) self.assertIn('status', result) @patch('subprocess.run') @patch.dict(os.environ, {'DOCKER_CONTAINER': 'false'}) def test_get_status_non_docker_path(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='') result = self.nm.get_status() self.assertIn('dns_running', result) self.assertIn('ntp_running', result) class TestGetDnsRecords(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) def tearDown(self): shutil.rmtree(self.tmp) @patch('subprocess.run') def test_get_dns_records_returns_list(self, _mock): result = self.nm.get_dns_records() self.assertIsInstance(result, list) @patch('subprocess.run') def test_get_dns_records_from_zone_file(self, _mock): dns_data = os.path.join(self.data_dir, 'dns') os.makedirs(dns_data, exist_ok=True) with open(os.path.join(dns_data, 'cell.zone'), 'w') as f: f.write('api 3600 IN A 10.0.0.1\nwebui 3600 IN A 10.0.0.1\n') result = self.nm.get_dns_records() self.assertEqual(len(result), 2) names = [r['name'] for r in result] self.assertIn('api', names) self.assertIn('webui', names) class TestInputValidation(unittest.TestCase): """Test input validation guards in add_dns_record and update_dns_zone.""" def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) def tearDown(self): shutil.rmtree(self.tmp) def test_add_dns_record_invalid_zone_returns_false(self): result = self.nm.add_dns_record('../etc/passwd', 'test', 'A', '10.0.0.1') self.assertFalse(result) def test_add_dns_record_invalid_name_returns_false(self): result = self.nm.add_dns_record('cell', 'name with spaces', 'A', '10.0.0.1') self.assertFalse(result) def test_add_dns_record_invalid_value_returns_false(self): result = self.nm.add_dns_record('cell', 'test', 'A', '10.0.0.1; rm -rf /') self.assertFalse(result) @patch('subprocess.run') def test_update_dns_zone_invalid_record_name_returns_false(self, _mock): records = [{'name': 'valid', 'type': 'A', 'value': '10.0.0.1'}, {'name': 'bad\nname', 'type': 'A', 'value': '10.0.0.2'}] result = self.nm.update_dns_zone('cell', records) self.assertFalse(result) def test_update_dns_zone_invalid_zone_name_returns_false(self): result = self.nm.update_dns_zone('', []) self.assertFalse(result) class TestGetWgServerIp(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.nm, self.data_dir, self.config_dir = _make_nm(self.tmp) def tearDown(self): shutil.rmtree(self.tmp) def test_returns_fallback_when_no_conf(self): ip = self.nm._get_wg_server_ip() self.assertEqual(ip, '10.0.0.1') def test_parses_wg_conf_address(self): wg_dir = os.path.join(self.config_dir, 'wireguard', 'wg_confs') os.makedirs(wg_dir, exist_ok=True) with open(os.path.join(wg_dir, 'wg0.conf'), 'w') as f: f.write('[Interface]\nAddress = 172.16.0.1/24\nPrivateKey = abc\n') ip = self.nm._get_wg_server_ip() self.assertEqual(ip, '172.16.0.1') class TestGetDnsOverview(unittest.TestCase): """get_dns_overview composes config_manager, registry, and ddns_manager into a provider-aware structure without writing DNS.""" def setUp(self): self.tmp = tempfile.mkdtemp() self.data_dir = os.path.join(self.tmp, 'data') self.config_dir = os.path.join(self.tmp, 'config') os.makedirs(os.path.join(self.data_dir, 'dns'), exist_ok=True) self.registry = MagicMock() self.registry.get_caddy_routes.return_value = [ {'subdomain': 'mail', 'backend': 'cell-mail:80', 'extra_subdomains': [], 'extra_backends': {}}, ] self.nm = NetworkManager(self.data_dir, self.config_dir, service_registry=self.registry) def tearDown(self): shutil.rmtree(self.tmp) def _cm(self, identity, ddns=None, token=''): cm = MagicMock() cm.get_identity.return_value = identity cm.configs = {'ddns': ddns or {}} cm.get_ddns_token.return_value = token mode = identity.get('domain_mode', 'lan') if mode == 'lan': cm.get_effective_domain.return_value = identity.get('domain', 'cell') else: cm.get_effective_domain.return_value = identity.get('domain_name', '') cm.get_internal_domain.return_value = identity.get('domain', 'cell') return cm def test_lan_mode_has_no_public_records(self): cm = self._cm({'domain_mode': 'lan', 'domain': 'cell'}) ov = self.nm.get_dns_overview(cm, ddns_manager=None) self.assertEqual(ov['mode'], 'lan') self.assertEqual(ov['public_records'], []) self.assertEqual(ov['internal_domain'], 'cell') self.assertIsNone(ov['public_ip']) def test_pic_ngo_mode_apex_and_wildcard(self): cm = self._cm( {'domain_mode': 'pic_ngo', 'domain': 'cell', 'domain_name': 'mycell.pic.ngo'}, ddns={'provider': 'pic_ngo'}, token='tok') ddns_mgr = MagicMock() ddns_mgr.get_status.return_value = {'provider': 'pic_ngo'} ov = self.nm.get_dns_overview(cm, ddns_manager=ddns_mgr, public_ip='1.2.3.4') names = [r['name'] for r in ov['public_records']] self.assertIn('mycell.pic.ngo', names) self.assertIn('*.mycell.pic.ngo', names) self.assertTrue(all(r['value'] == '1.2.3.4' for r in ov['public_records'])) self.assertTrue(all(r['status'] == 'registered' for r in ov['public_records'])) self.assertEqual(ov['service_subdomains'][0]['fqdn'], 'mail.mycell.pic.ngo') def test_cloudflare_mode_per_service_records(self): cm = self._cm( {'domain_mode': 'cloudflare', 'domain': 'cell', 'domain_name': 'cell.example.com'}, ddns={'provider': 'cloudflare'}, token='cf') ov = self.nm.get_dns_overview(cm, ddns_manager=None, public_ip='5.5.5.5') names = [r['name'] for r in ov['public_records']] self.assertIn('cell.example.com', names) self.assertIn('mail.cell.example.com', names) self.assertNotIn('*.cell.example.com', names) def test_custom_mode_per_service_records(self): cm = self._cm( {'domain_mode': 'custom', 'domain': 'cell', 'domain_name': 'cell.example.org'}, ddns={'provider': 'custom'}) ov = self.nm.get_dns_overview(cm, ddns_manager=None, public_ip='6.6.6.6') names = [r['name'] for r in ov['public_records']] self.assertIn('cell.example.org', names) self.assertIn('mail.cell.example.org', names) self.assertFalse(ov['registration_status']['registered']) self.assertTrue(all(r['status'] == 'unregistered' for r in ov['public_records'])) def test_internal_records_come_from_zone_files(self): with open(os.path.join(self.data_dir, 'dns', 'cell.zone'), 'w') as f: f.write('api 3600 IN A 10.0.0.1\n') cm = self._cm({'domain_mode': 'lan', 'domain': 'cell'}) ov = self.nm.get_dns_overview(cm, ddns_manager=None) self.assertEqual(len(ov['internal_records']), 1) self.assertEqual(ov['internal_records'][0]['name'], 'api') if __name__ == '__main__': unittest.main()