Use new auto assign logic for ip addresses

We no longer want users to pick ip addresses but just specify the
network and then they get automatically an ip address. The exception is
manually setting an ip address in which case we can use the adminapi to
get an ip address and lock it so others do not use it.

Author: kofrezo

tests/conftest.py

@@ -1,32 +1,27 @@
 """igvm - Shared functions
 
-Copyright (c) 2020 InnoGames GmbH
+Copyright (c) 2025 InnoGames GmbH
 """
 
 from datetime import datetime, timezone
-from math import ceil, log
 from re import match
-from time import sleep
 from shlex import quote
+from time import sleep
 from typing import Optional
 
 import boto3
 from adminapi.dataset import Query
 from adminapi.exceptions import DatasetError
-from adminapi.filters import Any, Not, Regexp
+from adminapi.filters import Any, Regexp
 from botocore.exceptions import ClientError
 from libvirt import VIR_DOMAIN_RUNNING
 
-from igvm.exceptions import IGVMTestError
 from igvm.hypervisor import Hypervisor
 from igvm.settings import HYPERVISOR_ATTRIBUTES, AWS_RETURN_CODES
 from tests import (
     IGVM_LOCKED_TIMEOUT,
     JENKINS_EXECUTOR,
-    PYTEST_XDIST_WORKER,
-    PYTEST_XDIST_WORKER_COUNT,
     VM_HOSTNAME_PATTERN,
-    VM_NET,
 )
 
 
@@ -70,14 +65,6 @@ def clean_all(route_network, datacenter_type, vm_hostname=None):
     # Remove all connected Serveradmin objects.
     clean_serveradmin({'hostname': Regexp(pattern)})
 
-    # Try to remove VMs with the same IP in any case because we use custom
-    # logic to assign them and we want to avoid IP address conflicts.
-    # Index 1 is usually used for the test's subject VM,
-    # 2 might be used for testing IP change.
-    for ip_attr in ('ipv4', 'ipv6'):
-        ips = [get_next_address(VM_NET, i, ip_attr) for i in [1, 2]]
-        clean_serveradmin({ip_attr: Any(*ips)})
-
 
 def clean_hv(hv, pattern):
     # We never know what happened on the HV, so always refresh
@@ -207,30 +194,3 @@ def _wait_for_state_reached(ec2_client, instance_id: str, state: str,
         if not any(error in str(e) for error in
                    ['InvalidInstanceID', 'IncorrectInstanceState']):
             raise
-
-
-def get_next_address(vm_net, index, ip_attr):
-    non_vm_hosts = list(Query({
-        'project_network': vm_net,
-        'servertype': Not('vm'),
-    }, [ip_attr]))
-    offset = 1 if len(non_vm_hosts) > 0 else 0
-    subnet_levels = ceil(log(PYTEST_XDIST_WORKER_COUNT + offset, 2))
-    project_network = Query({'hostname': vm_net}, [ip_attr]).get()
-    try:
-        subnets = list(project_network[ip_attr].subnets(subnet_levels))
-    except ValueError:
-        raise IGVMTestError(
-            'Can\'t split {} into enough subnets '
-            'for {} parallel tests'.format(
-                vm_net, PYTEST_XDIST_WORKER_COUNT,
-            )
-        )
-    if len(non_vm_hosts) > subnets[0].num_addresses:
-        raise IGVMTestError(
-            'Can\'t split {} into enough subnets '
-            'for {} parallel tests'.format(
-                vm_net, PYTEST_XDIST_WORKER_COUNT,
-            )
-        )
-    return subnets[PYTEST_XDIST_WORKER + 1][index]

tests/test_integration.py

@@ -58,7 +58,6 @@
 from tests.conftest import (
     clean_all,
     cmd,
-    get_next_address,
 )
 
 basicConfig(level=INFO)
@@ -114,7 +113,7 @@ def setUp(self):
         self.vm_obj['environment'] = 'testing'
         self.vm_obj['hostname'] = VM_HOSTNAME
         self.vm_obj['hypervisor'] = None
-        self.vm_obj['intern_ip'] = get_next_address(VM_NET, 1, self.ip_attr)
+        self.vm_obj['project_network'] = VM_NET
         self.vm_obj['memory'] = 2048
         self.vm_obj['no_monitoring'] = True
         self.vm_obj['num_cpu'] = 2
@@ -688,15 +687,11 @@ def test_reject_out_of_sync_serveradmin(self):
             vm_migrate(VM_HOSTNAME)
 
     def test_new_address(self):
-        # We don't have a way to ask for new IP address from Serveradmin
-        # and lock it for us. The method below will usually work fine.
-        # When it starts failing, we must develop retry method.
-        new_address = get_next_address(VM_NET, 2, 'ipv4')
-
+        new_address = Query({'hostname': VM_NET}, ['intern_ip']).get_free_ip_addr(lock=True)
         change_address(VM_HOSTNAME, new_address, offline=True)
 
-        obj = Query({'hostname': VM_HOSTNAME}, ['intern_ip']).get()
-        self.assertEqual(obj['intern_ip'], new_address)
+        obj = Query({'hostname': VM_HOSTNAME}, ['ipv4']).get()
+        self.assertEqual(obj['ipv4'], new_address)
         with _get_vm(VM_HOSTNAME) as vm:
             vm.run(cmd('ip a | grep {}', new_address))
         self.check_vm_present()