#!/usr/bin/env python3 """Tests for ip_utils — IP derivation from subnet.""" import sys import os import tempfile import unittest import unittest.mock from pathlib import Path api_dir = Path(__file__).parent.parent / 'api' sys.path.insert(0, str(api_dir)) import ip_utils class TestGetServiceIps(unittest.TestCase): def setUp(self): self.ips = ip_utils.get_service_ips('172.20.0.0/16') def test_returns_all_keys(self): for name in ip_utils.CONTAINER_OFFSETS: self.assertIn(name, self.ips) def test_default_subnet_caddy(self): self.assertEqual(self.ips['caddy'], '172.20.0.2') def test_default_subnet_dns(self): self.assertEqual(self.ips['dns'], '172.20.0.3') def test_default_subnet_api(self): self.assertEqual(self.ips['api'], '172.20.0.10') def test_default_subnet_virtual_ips(self): self.assertEqual(self.ips['vip_calendar'], '172.20.0.21') self.assertEqual(self.ips['vip_files'], '172.20.0.22') self.assertEqual(self.ips['vip_mail'], '172.20.0.23') self.assertEqual(self.ips['vip_webdav'], '172.20.0.24') def test_different_subnet_shifts_all_ips(self): ips = ip_utils.get_service_ips('10.0.0.0/24') self.assertEqual(ips['caddy'], '10.0.0.2') self.assertEqual(ips['dns'], '10.0.0.3') self.assertEqual(ips['api'], '10.0.0.10') self.assertEqual(ips['vip_calendar'], '10.0.0.21') def test_non_zero_third_octet_subnet(self): ips = ip_utils.get_service_ips('192.168.5.0/24') self.assertEqual(ips['caddy'], '192.168.5.2') self.assertEqual(ips['vip_webdav'], '192.168.5.24') def test_strict_false_accepts_host_bit_set(self): # e.g. user types "172.20.0.1/16" — should work same as "172.20.0.0/16" ips = ip_utils.get_service_ips('172.20.0.1/16') self.assertEqual(ips['caddy'], '172.20.0.2') def test_all_ips_are_strings(self): for name, ip in self.ips.items(): self.assertIsInstance(ip, str, f'{name} is not a string') def test_all_ips_unique(self): self.assertEqual(len(set(self.ips.values())), len(self.ips)) class TestGetVirtualIps(unittest.TestCase): def test_returns_four_services(self): vips = ip_utils.get_virtual_ips('172.20.0.0/16') self.assertEqual(set(vips.keys()), {'calendar', 'files', 'mail', 'webdav'}) def test_values_match_get_service_ips(self): full = ip_utils.get_service_ips('172.20.0.0/16') vips = ip_utils.get_virtual_ips('172.20.0.0/16') self.assertEqual(vips['calendar'], full['vip_calendar']) self.assertEqual(vips['files'], full['vip_files']) self.assertEqual(vips['mail'], full['vip_mail']) self.assertEqual(vips['webdav'], full['vip_webdav']) def test_different_subnet(self): vips = ip_utils.get_virtual_ips('10.10.0.0/16') self.assertEqual(vips['calendar'], '10.10.0.21') self.assertEqual(vips['webdav'], '10.10.0.24') class TestWriteEnvFile(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.env_path = os.path.join(self.tmp, '.env') def tearDown(self): import shutil shutil.rmtree(self.tmp) def test_creates_file(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path) self.assertTrue(os.path.exists(self.env_path)) def test_returns_true_on_success(self): result = ip_utils.write_env_file('172.20.0.0/16', self.env_path) self.assertTrue(result) def test_returns_false_on_unwritable_path(self): with unittest.mock.patch('builtins.open', side_effect=OSError('Permission denied')): result = ip_utils.write_env_file('172.20.0.0/16', '/any/path/.env') self.assertFalse(result) def test_contains_cell_network(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path) content = open(self.env_path).read() self.assertIn('CELL_NETWORK=172.20.0.0/16', content) def test_contains_caddy_ip(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path) content = open(self.env_path).read() self.assertIn('CADDY_IP=172.20.0.2', content) def test_contains_all_env_vars(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path) content = open(self.env_path).read() for var in ip_utils.ENV_VAR_NAMES.values(): self.assertIn(var + '=', content, f'{var} missing from .env') def test_custom_subnet_generates_correct_ips(self): ip_utils.write_env_file('10.5.0.0/24', self.env_path) content = open(self.env_path).read() self.assertIn('CELL_NETWORK=10.5.0.0/24', content) self.assertIn('CADDY_IP=10.5.0.2', content) self.assertIn('DNS_IP=10.5.0.3', content) self.assertIn('API_IP=10.5.0.10', content) self.assertNotIn('172.20', content) def test_overwrite_updates_ips(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path) ip_utils.write_env_file('10.0.0.0/24', self.env_path) content = open(self.env_path).read() self.assertIn('CADDY_IP=10.0.0.2', content) self.assertNotIn('172.20', content) def test_each_line_is_key_equals_value(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path) for line in open(self.env_path).read().splitlines(): if line.strip(): self.assertIn('=', line, f'Bad line format: {line!r}') key, _, val = line.partition('=') self.assertTrue(key.isupper() or '_' in key) self.assertTrue(val.strip()) class TestPortDefaults(unittest.TestCase): def test_port_defaults_exist(self): self.assertIsInstance(ip_utils.PORT_DEFAULTS, dict) for key in ('dns_port', 'mail_smtp_port', 'wg_port', 'radicale_port', 'webdav_port', 'api_port', 'webui_port', 'rainloop_port', 'filegator_port'): self.assertIn(key, ip_utils.PORT_DEFAULTS) def test_port_env_var_names_cover_all_defaults(self): self.assertEqual(set(ip_utils.PORT_DEFAULTS.keys()), set(ip_utils.PORT_ENV_VAR_NAMES.keys())) def test_port_to_containers_covers_all_defaults(self): self.assertEqual(set(ip_utils.PORT_DEFAULTS.keys()), set(ip_utils.PORT_TO_CONTAINERS.keys())) def test_default_values(self): self.assertEqual(ip_utils.PORT_DEFAULTS['dns_port'], 53) self.assertEqual(ip_utils.PORT_DEFAULTS['mail_smtp_port'], 25) self.assertEqual(ip_utils.PORT_DEFAULTS['mail_submission_port'], 587) self.assertEqual(ip_utils.PORT_DEFAULTS['mail_imap_port'], 993) self.assertEqual(ip_utils.PORT_DEFAULTS['radicale_port'], 5232) self.assertEqual(ip_utils.PORT_DEFAULTS['webdav_port'], 8080) self.assertEqual(ip_utils.PORT_DEFAULTS['wg_port'], 51820) self.assertEqual(ip_utils.PORT_DEFAULTS['api_port'], 3000) self.assertEqual(ip_utils.PORT_DEFAULTS['webui_port'], 8081) self.assertEqual(ip_utils.PORT_DEFAULTS['rainloop_port'], 8888) self.assertEqual(ip_utils.PORT_DEFAULTS['filegator_port'], 8082) class TestWriteEnvFilePorts(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.env_path = os.path.join(self.tmp, '.env') def tearDown(self): import shutil shutil.rmtree(self.tmp) def test_contains_default_ports(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path) content = open(self.env_path).read() self.assertIn('DNS_PORT=53', content) self.assertIn('WG_PORT=51820', content) self.assertIn('MAIL_SMTP_PORT=25', content) self.assertIn('MAIL_SUBMISSION_PORT=587', content) self.assertIn('MAIL_IMAP_PORT=993', content) self.assertIn('RADICALE_PORT=5232', content) self.assertIn('WEBDAV_PORT=8080', content) self.assertIn('API_PORT=3000', content) self.assertIn('WEBUI_PORT=8081', content) self.assertIn('RAINLOOP_PORT=8888', content) self.assertIn('FILEGATOR_PORT=8082', content) def test_custom_ports_override_defaults(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path, ports={'wg_port': 12345, 'api_port': 4000}) content = open(self.env_path).read() self.assertIn('WG_PORT=12345', content) self.assertIn('API_PORT=4000', content) self.assertIn('DNS_PORT=53', content) # unchanged default def test_custom_ports_do_not_leak_default(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path, ports={'wg_port': 12345}) content = open(self.env_path).read() self.assertNotIn('WG_PORT=51820', content) self.assertIn('WG_PORT=12345', content) def test_all_port_env_vars_present(self): ip_utils.write_env_file('172.20.0.0/16', self.env_path) content = open(self.env_path).read() for var in ip_utils.PORT_ENV_VAR_NAMES.values(): self.assertIn(var + '=', content, f'{var} missing from .env') class TestWriteEnvFileInPlace(unittest.TestCase): """write_env_file must update the file in-place (same inode) so Docker file bind-mounts inside containers see the change immediately. os.replace() would create a new inode and the bind-mount would remain pointing at the stale inode.""" def setUp(self): self.tmp = tempfile.mkdtemp() self.env_path = os.path.join(self.tmp, '.env') # Pre-create the file so it has an initial inode with open(self.env_path, 'w') as f: f.write('INITIAL=1\n') self.initial_inode = os.stat(self.env_path).st_ino def tearDown(self): import shutil shutil.rmtree(self.tmp) def test_same_inode_after_write(self): """Inode must NOT change after write_env_file — bind-mounts track the inode.""" ip_utils.write_env_file('172.20.0.0/16', self.env_path) after_inode = os.stat(self.env_path).st_ino self.assertEqual(self.initial_inode, after_inode, 'write_env_file changed the file inode — Docker bind-mounts ' 'would not see the update') def test_same_inode_after_port_change(self): """Inode must be preserved even when port values change.""" ip_utils.write_env_file('172.20.0.0/16', self.env_path, {'wg_port': 51820}) inode_first = os.stat(self.env_path).st_ino ip_utils.write_env_file('172.20.0.0/16', self.env_path, {'wg_port': 51821}) inode_second = os.stat(self.env_path).st_ino self.assertEqual(inode_first, inode_second, 'write_env_file changed inode on second write') self.assertIn('WG_PORT=51821', open(self.env_path).read()) def test_content_visible_via_open_after_write(self): """After write_env_file the new content is immediately readable through the same file descriptor path (same inode).""" ip_utils.write_env_file('172.20.0.0/16', self.env_path, {'wg_port': 9999}) content = open(self.env_path).read() self.assertIn('WG_PORT=9999', content) self.assertNotIn('INITIAL=1', content) if __name__ == '__main__': unittest.main()