roof
615448b875
When ip_range changes in Settings, the new subnet is now applied to: - DNS zone records (network_manager.apply_ip_range) - Caddy virtual IPs (firewall_manager.ensure_caddy_virtual_ips) - iptables per-service rules (firewall_manager.update_service_ips) - docker-compose.yml static IPs if writable (ip_utils.update_docker_compose_ips) New module ip_utils.py derives all container IPs from the subnet using fixed offsets so the entire stack stays consistent from one setting. 321 tests pass (72 new tests added for ip_utils, apply_ip_range, update_service_ips). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
410 lines
16 KiB
Python
410 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Unit tests for NetworkManager class
|
|
"""
|
|
|
|
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 os
|
|
import json
|
|
import shutil
|
|
from unittest.mock import patch, MagicMock
|
|
from datetime import datetime
|
|
|
|
# Add parent directory to path for imports
|
|
import sys
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from network_manager import NetworkManager
|
|
|
|
class TestNetworkManager(unittest.TestCase):
|
|
"""Test cases for NetworkManager class"""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment"""
|
|
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)
|
|
|
|
# Create NetworkManager instance
|
|
self.network_manager = NetworkManager(self.data_dir, self.config_dir)
|
|
|
|
def tearDown(self):
|
|
"""Clean up test environment"""
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
def test_initialization(self):
|
|
"""Test NetworkManager initialization"""
|
|
self.assertEqual(self.network_manager.data_dir, self.data_dir)
|
|
self.assertEqual(self.network_manager.config_dir, self.config_dir)
|
|
self.assertTrue(os.path.exists(self.network_manager.dns_zones_dir))
|
|
self.assertTrue(os.path.exists(os.path.dirname(self.network_manager.dhcp_leases_file)))
|
|
|
|
def test_generate_zone_content(self):
|
|
"""Test DNS zone content generation"""
|
|
records = [
|
|
{'name': 'test1', 'type': 'A', 'value': '192.168.1.10', 'ttl': 3600},
|
|
{'name': 'test2', 'type': 'CNAME', 'value': 'test1', 'ttl': 1800}
|
|
]
|
|
|
|
content = self.network_manager._generate_zone_content('test.cell', records)
|
|
|
|
self.assertIn('test.cell', content)
|
|
self.assertIn('SOA', content)
|
|
self.assertIn('192.168.1.10', content)
|
|
self.assertIn('test1', content)
|
|
self.assertIn('CNAME', content)
|
|
|
|
def test_add_dns_record(self):
|
|
"""Test adding DNS record"""
|
|
success = self.network_manager.add_dns_record('test.cell', 'test', 'A', '192.168.1.100')
|
|
self.assertTrue(success)
|
|
|
|
# Check if zone file was created
|
|
zone_file = os.path.join(self.network_manager.dns_zones_dir, 'test.cell.zone')
|
|
self.assertTrue(os.path.exists(zone_file))
|
|
|
|
# Check content
|
|
with open(zone_file, 'r') as f:
|
|
content = f.read()
|
|
self.assertIn('test', content)
|
|
self.assertIn('192.168.1.100', content)
|
|
|
|
def test_remove_dns_record(self):
|
|
"""Test removing DNS record"""
|
|
# Add a record first
|
|
self.network_manager.add_dns_record('test.cell', 'test', 'A', '192.168.1.100')
|
|
|
|
# Remove it
|
|
success = self.network_manager.remove_dns_record('test.cell', 'test', 'A')
|
|
self.assertTrue(success)
|
|
|
|
# Check if record was removed
|
|
zone_file = os.path.join(self.network_manager.dns_zones_dir, 'test.cell.zone')
|
|
with open(zone_file, 'r') as f:
|
|
content = f.read()
|
|
self.assertNotIn('192.168.1.100', content)
|
|
|
|
def test_load_dns_records(self):
|
|
"""Test loading DNS records from zone file"""
|
|
# Create a test zone file
|
|
zone_file = os.path.join(self.network_manager.dns_zones_dir, 'test.cell.zone')
|
|
content = """$TTL 3600
|
|
@ IN SOA test.cell. admin.test.cell. (
|
|
2024010101 ; Serial
|
|
3600 ; Refresh
|
|
1800 ; Retry
|
|
1209600 ; Expire
|
|
3600 ; Minimum TTL
|
|
)
|
|
|
|
; Name servers
|
|
@ IN NS test.cell.
|
|
|
|
test1 3600 IN A 192.168.1.10
|
|
test2 1800 IN CNAME test1
|
|
"""
|
|
|
|
with open(zone_file, 'w') as f:
|
|
f.write(content)
|
|
|
|
records = self.network_manager._load_dns_records('test.cell')
|
|
|
|
self.assertEqual(len(records), 2)
|
|
self.assertEqual(records[0]['name'], 'test1')
|
|
self.assertEqual(records[0]['value'], '192.168.1.10')
|
|
self.assertEqual(records[1]['name'], 'test2')
|
|
self.assertEqual(records[1]['type'], 'CNAME')
|
|
|
|
def test_get_dhcp_leases(self):
|
|
"""Test getting DHCP leases"""
|
|
# Create a test leases file
|
|
leases_file = self.network_manager.dhcp_leases_file
|
|
content = """1234567890 aa:bb:cc:dd:ee:ff 192.168.1.100 testhost *
|
|
1234567891 11:22:33:44:55:66 192.168.1.101 anotherhost *
|
|
"""
|
|
|
|
with open(leases_file, 'w') as f:
|
|
f.write(content)
|
|
|
|
leases = self.network_manager.get_dhcp_leases()
|
|
|
|
self.assertEqual(len(leases), 2)
|
|
self.assertEqual(leases[0]['mac'], 'aa:bb:cc:dd:ee:ff')
|
|
self.assertEqual(leases[0]['ip'], '192.168.1.100')
|
|
self.assertEqual(leases[0]['hostname'], 'testhost')
|
|
self.assertEqual(leases[1]['mac'], '11:22:33:44:55:66')
|
|
self.assertEqual(leases[1]['ip'], '192.168.1.101')
|
|
|
|
def test_add_dhcp_reservation(self):
|
|
"""Test adding DHCP reservation"""
|
|
success = self.network_manager.add_dhcp_reservation('aa:bb:cc:dd:ee:ff', '192.168.1.100', 'testhost')
|
|
self.assertTrue(success)
|
|
|
|
# Check if reservation file was created
|
|
reservation_file = os.path.join(self.config_dir, 'dhcp', 'reservations.conf')
|
|
self.assertTrue(os.path.exists(reservation_file))
|
|
|
|
# Check content
|
|
with open(reservation_file, 'r') as f:
|
|
content = f.read()
|
|
self.assertIn('aa:bb:cc:dd:ee:ff', content)
|
|
self.assertIn('192.168.1.100', content)
|
|
self.assertIn('testhost', content)
|
|
|
|
def test_remove_dhcp_reservation(self):
|
|
"""Test removing DHCP reservation"""
|
|
# Add a reservation first
|
|
self.network_manager.add_dhcp_reservation('aa:bb:cc:dd:ee:ff', '192.168.1.100', 'testhost')
|
|
|
|
# Remove it
|
|
success = self.network_manager.remove_dhcp_reservation('aa:bb:cc:dd:ee:ff')
|
|
self.assertTrue(success)
|
|
|
|
# Check if reservation was removed
|
|
reservation_file = os.path.join(self.config_dir, 'dhcp', 'reservations.conf')
|
|
with open(reservation_file, 'r') as f:
|
|
content = f.read()
|
|
self.assertNotIn('aa:bb:cc:dd:ee:ff', content)
|
|
|
|
@patch('subprocess.run')
|
|
def test_get_ntp_status(self, mock_run):
|
|
"""Test getting NTP status"""
|
|
# Mock NTP service running
|
|
mock_run.return_value.stdout = 'cell-ntp\n'
|
|
mock_run.return_value.returncode = 0
|
|
|
|
status = self.network_manager.get_ntp_status()
|
|
|
|
self.assertTrue(status['running'])
|
|
self.assertIn('stats', status)
|
|
|
|
@patch('subprocess.run')
|
|
def test_get_ntp_status_not_running(self, mock_run):
|
|
"""Test getting NTP status when service is not running"""
|
|
# Mock NTP service not running
|
|
mock_run.return_value.stdout = ''
|
|
mock_run.return_value.returncode = 0
|
|
|
|
status = self.network_manager.get_ntp_status()
|
|
|
|
self.assertFalse(status['running'])
|
|
self.assertIn('stats', status)
|
|
|
|
@patch('socket.getaddrinfo')
|
|
def test_test_dns_resolution(self, mock_getaddrinfo):
|
|
"""Test DNS resolution testing"""
|
|
mock_getaddrinfo.return_value = [(None, None, None, None, ('192.168.1.100', 0))]
|
|
result = self.network_manager.test_dns_resolution('test.cell')
|
|
self.assertTrue(result['success'])
|
|
self.assertIn('192.168.1.100', result['output'])
|
|
|
|
@patch('socket.getaddrinfo')
|
|
def test_test_dns_resolution_failure(self, mock_getaddrinfo):
|
|
"""Test DNS resolution testing with failure"""
|
|
import socket
|
|
mock_getaddrinfo.side_effect = socket.gaierror('NXDOMAIN')
|
|
result = self.network_manager.test_dns_resolution('nonexistent.cell')
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('NXDOMAIN', result['error'])
|
|
|
|
@patch('subprocess.run')
|
|
def test_test_dhcp_functionality(self, mock_run):
|
|
"""Test DHCP functionality testing"""
|
|
# Mock DHCP service running
|
|
mock_run.return_value.stdout = 'cell-dhcp\n'
|
|
mock_run.return_value.returncode = 0
|
|
|
|
result = self.network_manager.test_dhcp_functionality()
|
|
|
|
self.assertTrue(result['running'])
|
|
self.assertIn('leases_count', result)
|
|
self.assertIn('leases', result)
|
|
|
|
@patch('subprocess.run')
|
|
def test_test_ntp_functionality(self, mock_run):
|
|
"""Test NTP functionality testing"""
|
|
# Mock NTP service running with tracking
|
|
mock_run.return_value.stdout = 'cell-ntp\n'
|
|
mock_run.return_value.returncode = 0
|
|
|
|
result = self.network_manager.test_ntp_functionality()
|
|
|
|
self.assertTrue(result['running'])
|
|
self.assertIn('ntp_test', result)
|
|
|
|
def test_update_dns_zone(self):
|
|
"""Test updating DNS zone"""
|
|
records = [
|
|
{'name': 'test1', 'type': 'A', 'value': '192.168.1.10', 'ttl': 3600},
|
|
{'name': 'test2', 'type': 'A', 'value': '192.168.1.11', 'ttl': 3600}
|
|
]
|
|
|
|
success = self.network_manager.update_dns_zone('test.cell', records)
|
|
self.assertTrue(success)
|
|
|
|
# Check if zone file was created
|
|
zone_file = os.path.join(self.network_manager.dns_zones_dir, 'test.cell.zone')
|
|
self.assertTrue(os.path.exists(zone_file))
|
|
|
|
# Check content
|
|
with open(zone_file, 'r') as f:
|
|
content = f.read()
|
|
self.assertIn('test1', content)
|
|
self.assertIn('test2', content)
|
|
self.assertIn('192.168.1.10', content)
|
|
self.assertIn('192.168.1.11', content)
|
|
|
|
class TestBootstrapDnsRecords(unittest.TestCase):
|
|
"""Test bootstrap_dns_records with dynamic IP derivation."""
|
|
|
|
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.nm = NetworkManager(self.data_dir, self.config_dir)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
@patch('subprocess.run')
|
|
def test_creates_zone_file(self, _mock):
|
|
self.nm.bootstrap_dns_records('mycell', 'cell')
|
|
zone_file = os.path.join(self.nm.dns_zones_dir, 'cell.zone')
|
|
self.assertTrue(os.path.exists(zone_file))
|
|
|
|
@patch('subprocess.run')
|
|
def test_contains_default_caddy_ip(self, _mock):
|
|
self.nm.bootstrap_dns_records('mycell', 'cell')
|
|
zone_file = os.path.join(self.nm.dns_zones_dir, 'cell.zone')
|
|
content = open(zone_file).read()
|
|
self.assertIn('172.20.0.2', content) # caddy
|
|
|
|
@patch('subprocess.run')
|
|
def test_custom_ip_range_used(self, _mock):
|
|
self.nm.bootstrap_dns_records('mycell', 'cell', ip_range='10.5.0.0/24')
|
|
zone_file = os.path.join(self.nm.dns_zones_dir, 'cell.zone')
|
|
content = open(zone_file).read()
|
|
self.assertIn('10.5.0.2', content) # caddy
|
|
self.assertIn('10.5.0.21', content) # vip_calendar
|
|
self.assertNotIn('172.20', content)
|
|
|
|
@patch('subprocess.run')
|
|
def test_idempotent_skips_existing_zone(self, _mock):
|
|
self.nm.bootstrap_dns_records('mycell', 'cell')
|
|
zone_file = os.path.join(self.nm.dns_zones_dir, 'cell.zone')
|
|
mtime1 = os.path.getmtime(zone_file)
|
|
self.nm.bootstrap_dns_records('mycell', 'cell')
|
|
mtime2 = os.path.getmtime(zone_file)
|
|
self.assertEqual(mtime1, mtime2)
|
|
|
|
|
|
class TestApplyIpRange(unittest.TestCase):
|
|
"""Test apply_ip_range rewrites DNS zone records."""
|
|
|
|
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.nm = NetworkManager(self.data_dir, self.config_dir)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
@patch('subprocess.run')
|
|
def test_zone_file_updated_with_new_ips(self, _mock):
|
|
# Bootstrap with default range, then change to 10.0.0.0/24
|
|
self.nm.bootstrap_dns_records('mycell', 'cell', '172.20.0.0/16')
|
|
result = self.nm.apply_ip_range('10.0.0.0/24', 'mycell', 'cell')
|
|
zone_file = os.path.join(self.nm.dns_zones_dir, 'cell.zone')
|
|
content = open(zone_file).read()
|
|
self.assertIn('10.0.0.2', content) # caddy
|
|
self.assertIn('10.0.0.21', content) # vip_calendar
|
|
self.assertNotIn('172.20', content)
|
|
|
|
@patch('subprocess.run')
|
|
def test_returns_restarted_on_success(self, _mock):
|
|
self.nm.bootstrap_dns_records('mycell', 'cell', '172.20.0.0/16')
|
|
result = self.nm.apply_ip_range('10.0.0.0/24', 'mycell', 'cell')
|
|
self.assertIn('cell-dns (reloaded)', result['restarted'])
|
|
|
|
@patch('subprocess.run')
|
|
def test_all_standard_records_present(self, _mock):
|
|
self.nm.apply_ip_range('10.1.2.0/24', 'pictest', 'mycell')
|
|
zone_file = os.path.join(self.nm.dns_zones_dir, 'mycell.zone')
|
|
content = open(zone_file).read()
|
|
for host in ('pictest', 'api', 'webui', 'calendar', 'files', 'mail', 'webmail', 'webdav'):
|
|
self.assertIn(host, content)
|
|
|
|
@patch('subprocess.run')
|
|
def test_same_range_updates_zone_without_error(self, _mock):
|
|
self.nm.bootstrap_dns_records('mycell', 'cell', '172.20.0.0/16')
|
|
result = self.nm.apply_ip_range('172.20.0.0/16', 'mycell', 'cell')
|
|
self.assertEqual(result['warnings'], [])
|
|
|
|
|
|
class TestCellDnsForwarding(unittest.TestCase):
|
|
"""Test add/remove cell DNS forwarding in Corefile."""
|
|
|
|
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(os.path.join(self.config_dir, 'dns'), exist_ok=True)
|
|
self.nm = NetworkManager(self.data_dir, self.config_dir)
|
|
self.corefile = os.path.join(self.config_dir, 'dns', 'Corefile')
|
|
with open(self.corefile, 'w') as f:
|
|
f.write('home.cell {\n file /data/home.cell.zone\n log\n}\n\n. {\n forward . 8.8.8.8\n log\n}\n')
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
@patch('subprocess.run')
|
|
def test_add_cell_dns_forward_appends_block(self, _mock):
|
|
self.nm.add_cell_dns_forward('remote.cell', '10.1.0.1')
|
|
with open(self.corefile) as f:
|
|
content = f.read()
|
|
self.assertIn('remote.cell', content)
|
|
self.assertIn('10.1.0.1', content)
|
|
self.assertIn('forward . 10.1.0.1', content)
|
|
|
|
@patch('subprocess.run')
|
|
def test_add_cell_dns_forward_idempotent(self, _mock):
|
|
self.nm.add_cell_dns_forward('remote.cell', '10.1.0.1')
|
|
self.nm.add_cell_dns_forward('remote.cell', '10.1.0.1')
|
|
with open(self.corefile) as f:
|
|
content = f.read()
|
|
self.assertEqual(content.count('forward . 10.1.0.1'), 1)
|
|
|
|
@patch('subprocess.run')
|
|
def test_remove_cell_dns_forward_cleans_block(self, _mock):
|
|
self.nm.add_cell_dns_forward('remote.cell', '10.1.0.1')
|
|
self.nm.remove_cell_dns_forward('remote.cell')
|
|
with open(self.corefile) as f:
|
|
content = f.read()
|
|
self.assertNotIn('remote.cell', content)
|
|
self.assertNotIn('10.1.0.1', content)
|
|
|
|
@patch('subprocess.run')
|
|
def test_remove_nonexistent_forward_is_noop(self, _mock):
|
|
before = open(self.corefile).read()
|
|
self.nm.remove_cell_dns_forward('nonexistent.cell')
|
|
after = open(self.corefile).read()
|
|
self.assertEqual(before, after)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |