feat: dynamic ip_range propagation to DNS, firewall, and docker-compose
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>
This commit is contained in:
@@ -0,0 +1,152 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for ip_utils — IP derivation from subnet."""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
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 TestUpdateDockerComposeIps(unittest.TestCase):
|
||||
COMPOSE_TEMPLATE = """\
|
||||
version: '3.3'
|
||||
services:
|
||||
caddy:
|
||||
networks:
|
||||
cell-network:
|
||||
ipv4_address: 172.20.0.2
|
||||
dns:
|
||||
networks:
|
||||
cell-network:
|
||||
ipv4_address: 172.20.0.3
|
||||
api:
|
||||
networks:
|
||||
cell-network:
|
||||
ipv4_address: 172.20.0.10
|
||||
networks:
|
||||
cell-network:
|
||||
ipam:
|
||||
config:
|
||||
- subnet: 172.20.0.0/16
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False)
|
||||
self.tmp.write(self.COMPOSE_TEMPLATE)
|
||||
self.tmp.close()
|
||||
|
||||
def tearDown(self):
|
||||
os.unlink(self.tmp.name)
|
||||
|
||||
def test_returns_false_for_missing_file(self):
|
||||
self.assertFalse(
|
||||
ip_utils.update_docker_compose_ips('172.20.0.0/16', '10.0.0.0/24', '/nonexistent/path.yml')
|
||||
)
|
||||
|
||||
def test_subnet_updated(self):
|
||||
ip_utils.update_docker_compose_ips('172.20.0.0/16', '10.0.0.0/24', self.tmp.name)
|
||||
with open(self.tmp.name) as f:
|
||||
content = f.read()
|
||||
self.assertIn('10.0.0.0/24', content)
|
||||
self.assertNotIn('172.20.0.0/16', content)
|
||||
|
||||
def test_caddy_ip_updated(self):
|
||||
ip_utils.update_docker_compose_ips('172.20.0.0/16', '10.0.0.0/24', self.tmp.name)
|
||||
with open(self.tmp.name) as f:
|
||||
content = f.read()
|
||||
self.assertIn('10.0.0.2', content)
|
||||
self.assertNotIn('172.20.0.2', content)
|
||||
|
||||
def test_api_ip_updated(self):
|
||||
ip_utils.update_docker_compose_ips('172.20.0.0/16', '10.0.0.0/24', self.tmp.name)
|
||||
with open(self.tmp.name) as f:
|
||||
content = f.read()
|
||||
self.assertIn('10.0.0.10', content)
|
||||
self.assertNotIn('172.20.0.10', content)
|
||||
|
||||
def test_returns_true_on_success(self):
|
||||
result = ip_utils.update_docker_compose_ips('172.20.0.0/16', '10.0.0.0/24', self.tmp.name)
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_noop_when_ranges_same(self):
|
||||
ip_utils.update_docker_compose_ips('172.20.0.0/16', '172.20.0.0/16', self.tmp.name)
|
||||
with open(self.tmp.name) as f:
|
||||
content = f.read()
|
||||
self.assertEqual(content, self.COMPOSE_TEMPLATE)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user