roof
3a35cf72d3
Tests assumed write to /nonexistent/... fails, but CI runs as root where Linux allows creating any path. Use unittest.mock.patch on builtins.open with OSError side_effect so the test is environment-independent. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
266 lines
11 KiB
Python
266 lines
11 KiB
Python
#!/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()
|