fix(oci): reserve down and pending private network definitions

oci/remote/allocate_private_network.py

@@ -16,6 +16,47 @@ def command_output(*command: str) -> str:
     return result.stdout
 
 
+def _interface_files() -> list[Path]:
+    """Include pending Proxmox network edits and down interface definitions."""
+    paths = [Path('/etc/network/interfaces'), Path('/etc/network/interfaces.new')]
+    paths.extend(Path('/etc/network/interfaces.d').glob('*'))
+    return [path for path in paths if path.is_file()]
+
+
+def _configured_networks(text: str) -> list[ipaddress.IPv4Network]:
+    networks = []
+    addresses = []
+    netmask = None
+    inet = False
+
+    def collect():
+        for address in addresses:
+            value = address if '/' in address else address+'/'+(netmask or '32')
+            try:
+                interface = ipaddress.ip_interface(value)
+                if interface.version == 4:
+                    networks.append(interface.network)
+            except ValueError:
+                continue
+
+    for line in text.splitlines():
+        words = line.partition('#')[0].split()
+        if not words:
+            continue
+        if words[0] == 'iface':
+            collect()
+            addresses = []
+            netmask = None
+            inet = len(words) >= 4 and words[2] == 'inet'
+        elif inet and len(words) >= 2:
+            if words[0] == 'address':
+                addresses.append(words[1])
+            elif words[0] == 'netmask':
+                netmask = words[1]
+    collect()
+    return networks
+
+
 def existing_bridges() -> set[str]:
     bridges: set[str] = set()
     for line in command_output("ip", "-o", "link", "show").splitlines():
@@ -25,15 +66,13 @@ def existing_bridges() -> set[str]:
     for path in Path("/etc/pve/lxc").glob("*.conf"):
         text = path.read_text(encoding="utf-8", errors="ignore")
         bridges.update(re.findall(r"(?:^|,)bridge=([^,\s]+)", text, re.MULTILINE))
-    interface_paths = [Path("/etc/network/interfaces")]
-    interface_paths.extend(Path("/etc/network/interfaces.d").glob("*"))
-    for path in interface_paths:
-        if not path.is_file():
-            continue
+    for path in _interface_files():
         text = path.read_text(encoding="utf-8", errors="ignore")
-        bridges.update(
-            re.findall(r"^(?:auto|iface)\s+(vmbr\d+)\b", text, re.MULTILINE)
-        )
+        for line in text.splitlines():
+            words = line.partition('#')[0].split()
+            if words and words[0] in {'auto', 'iface'}:
+                names = words[1:] if words[0] == 'auto' else words[1:2]
+                bridges.update(name for name in names if re.fullmatch(r'vmbr\d+', name))
     return bridges
 
 
@@ -54,6 +93,8 @@ def existing_networks() -> list[ipaddress.IPv4Network]:
                 networks.append(ipaddress.ip_interface(address).network)
             except ValueError:
                 pass
+    for path in _interface_files():
+        networks.extend(_configured_networks(path.read_text(encoding='utf-8', errors='ignore')))
     return networks
 
 

oci/tests/test_private_network_configured.py

@@ -0,0 +1,74 @@
+"""Down and pending bridge configs must reserve their names and IPv4 subnets."""
+import importlib.util
+import ipaddress
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / 'remote'))
+spec = importlib.util.spec_from_file_location('configured_private_network', ROOT / 'remote/allocate_private_network.py')
+allocator = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(allocator)
+
+
+def deployment():
+    return {'network': {'private_allocation':'automatic', 'private_subnet':'10.77.0.0/24',
+                        'private_bridge':'vmbr10', 'private_host_address':'10.77.0.1/24'}}
+
+
+class ConfiguredPrivateNetworkTests(unittest.TestCase):
+    def test_interface_files_include_pending_main_and_include_fragments(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root/'interfaces').write_text('auto vmbr0\n')
+            (root/'interfaces.new').write_text('auto vmbr10\n')
+            (root/'interfaces.d').mkdir()
+            (root/'interfaces.d/down').write_text('iface vmbr11 inet static\n')
+            (root/'interfaces.d/nested').mkdir()
+            def fake_path(value):
+                return root / Path(value).relative_to('/etc/network')
+            with patch.object(allocator, 'Path', side_effect=fake_path):
+                files = allocator._interface_files()
+            self.assertEqual(set(files), {root/'interfaces',root/'interfaces.new',root/'interfaces.d/down'})
+
+    def test_down_and_pending_subnets_are_occupied_even_with_different_bridge_names(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            down = root/'interfaces'
+            down.write_text('iface vmbr999 inet static\n    address 10.77.0.1\n    netmask 255.255.255.0\n')
+            pending = root/'interfaces.new'
+            pending.write_text('auto vmbr11\niface vmbr11 inet static\n    address 10.77.1.1/24\n')
+            with patch.object(allocator, '_interface_files', return_value=[down,pending]), \
+                 patch.object(allocator, 'command_output', return_value=''), \
+                 patch.object(allocator.Path, 'glob', return_value=[]):
+                bridges = allocator.existing_bridges()
+                occupied = allocator.existing_networks()
+            self.assertIn('vmbr11', bridges)
+            self.assertIn(ipaddress.ip_network('10.77.0.0/24'), occupied)
+            self.assertIn(ipaddress.ip_network('10.77.1.0/24'), occupied)
+            self.assertEqual(allocator.allocate_network(deployment(),bridges,occupied),
+                             ('vmbr12',ipaddress.ip_network('10.77.2.0/24')))
+
+    def test_pending_bridge_name_reserves_name_without_an_address(self):
+        with tempfile.TemporaryDirectory() as directory:
+            pending = Path(directory)/'interfaces.new'
+            pending.write_text('auto vmbr0 vmbr10\niface vmbr10 inet manual\n')
+            with patch.object(allocator, '_interface_files', return_value=[pending]), \
+                 patch.object(allocator, 'command_output', return_value=''), \
+                 patch.object(allocator.Path, 'glob', return_value=[]):
+                bridges = allocator.existing_bridges()
+            self.assertEqual(allocator.allocate_network(deployment(),bridges,[]),
+                             ('vmbr11',ipaddress.ip_network('10.77.1.0/24')))
+
+    def test_ipv6_and_invalid_addresses_do_not_reserve_ipv4_pool(self):
+        text = ('iface vmbr12 inet6 static\n address fd00::1/64\n'
+                'iface vmbr20 inet static\n address invalid\n'
+                'iface vmbr21 inet static\n address 10.77.8.1/24 # comment\n')
+        self.assertEqual(allocator._configured_networks(text), [ipaddress.ip_network('10.77.8.0/24')])
+
+
+if __name__ == '__main__':
+    unittest.main()