Adapt tests for IPv6 hosts

- add brackets around IP in URI when necessary
- ignore tests not suporting IPv6

Signed-off-by: BenjiReis <[email protected]>

conftest.py

@@ -8,6 +8,7 @@
 
 from lib.common import wait_for, vm_image, is_uuid
 from lib.common import setup_formatted_and_mounted_disk, teardown_formatted_and_mounted_disk
+from lib.netutil import is_ipv6
 from lib.pool import Pool
 from lib.vm import VM
 
@@ -162,6 +163,17 @@ def xfail_on_xcpng_8_3(host, request):
     if host.xcp_version >= version.parse("8.3"):
         request.node.add_marker(pytest.mark.xfail)
 
+@pytest.fixture(scope='session')
+def host_no_ipv6(host):
+    if is_ipv6(host.hostname_or_ip):
+        pytest.skip(f"This test requires an IPv4 XCP-ng")
+
+@pytest.fixture(scope='function')
+def xfail_on_xcpng_ipv6(host, request):
+    """ Test that is relevant but expected to fail in current state of XCP-ng 8.3 in IPv6. """
+    if is_ipv6(host.hostname_or_ip):
+        request.node.add_marker(pytest.mark.xfail)
+
 @pytest.fixture(scope='session')
 def local_sr_on_hostA1(hostA1):
     """ A local SR on the pool's master. """

lib/commands.py

@@ -4,6 +4,7 @@
 
 import lib.config as config
 
+from lib.netutil import wrap_ip
 
 class BaseCommandFailed(Exception):
     __slots__ = 'returncode', 'stdout', 'cmd'
@@ -157,10 +158,11 @@ def scp(hostname_or_ip, src, dest, check=True, suppress_fingerprint_warnings=Tru
         # Based on https://unix.stackexchange.com/a/365976/257493
         opts = '-o "StrictHostKeyChecking no" -o "LogLevel ERROR" -o "UserKnownHostsFile /dev/null"'
 
+    ip = wrap_ip(hostname_or_ip)
     if local_dest:
-        src = 'root@{}:{}'.format(hostname_or_ip, src)
+        src = 'root@{}:{}'.format(ip, src)
     else:
-        dest = 'root@{}:{}'.format(hostname_or_ip, dest)
+        dest = 'root@{}:{}'.format(ip, dest)
 
     command = "scp {} {} {}".format(opts, src, dest)
     res = subprocess.run(

lib/host.py

@@ -10,6 +10,7 @@
 
 from lib.common import _param_get, safe_split, to_xapi_bool, wait_for, wait_for_not
 from lib.common import prefix_object_name
+from lib.netutil import wrap_ip
 from lib.sr import SR
 from lib.vm import VM
 from lib.xo import xo_cli, xo_object_exists
@@ -142,7 +143,7 @@ def _get_xensource_inventory(self):
     def xo_get_server_id(self, store=True):
         servers = xo_cli('server.getAll', use_json=True)
         for server in servers:
-            if server['host'] == self.hostname_or_ip:
+            if server['host'] == wrap_ip(self.hostname_or_ip):
                 if store:
                     self.xo_srv_id = server['id']
                 return server['id']
@@ -154,7 +155,7 @@ def xo_server_remove(self):
         else:
             servers = xo_cli('server.getAll', use_json=True)
             for server in servers:
-                if server['host'] == self.hostname_or_ip:
+                if server['host'] == wrap_ip(self.hostname_or_ip):
                     xo_cli('server.remove', {'id': server['id']})
 
     def xo_server_add(self, username, password, label=None, unregister_first=True):
@@ -166,7 +167,7 @@ def xo_server_add(self, username, password, label=None, unregister_first=True):
         xo_srv_id = xo_cli(
             'server.add',
             {
-                'host': self.hostname_or_ip,
+                'host': wrap_ip(self.hostname_or_ip),
                 'username': username,
                 'password': password,
                 'allowUnauthorized': 'true',
@@ -178,7 +179,7 @@ def xo_server_add(self, username, password, label=None, unregister_first=True):
     def xo_server_status(self):
         servers = xo_cli('server.getAll', use_json=True)
         for server in servers:
-            if server['host'] == self.hostname_or_ip:
+            if server['host'] == wrap_ip(self.hostname_or_ip):
                 return server['status']
         return None
 

lib/netutil.py

@@ -0,0 +1,12 @@
+import socket
+
+def is_ipv6(ip):
+    try:
+        socket.inet_pton(socket.AF_INET6, ip)
+        return True
+    except Exception:
+        return False
+
+def wrap_ip(ip):
+    """ Wrap an IP between brackets if and only if it's an IPv6. """
+    return f"[{ip}]" if is_ipv6(ip) else ip

tests/misc/test_basic_without_ssh.py

@@ -75,6 +75,7 @@ def test_checkpoint(self, imported_vm):
     # Live migration tests
     # We want to test storage migration (memory+disks) and live migration without storage migration (memory only).
     # The order will depend on the initial location of the VM: a local SR or a shared SR.
+    @pytest.mark.usefixtures("hostA2")
     def test_live_migrate(self, imported_vm, existing_shared_sr):
         def live_migrate(vm, dest_host, dest_sr, check_vdis=False):
             vm.migrate(dest_host, dest_sr)

tests/misc/test_file_server.py

@@ -1,6 +1,8 @@
 import pytest
 import subprocess
 
+from lib.netutil import wrap_ip
+
 # These tests are meant to test an host fileserver behavior.
 #
 # Requirements:
@@ -9,23 +11,24 @@
 
 def test_fileserver_redirect_https(host):
     path = "/path/to/dir/file.txt"
+    ip = wrap_ip(host.hostname_or_ip)
     process = subprocess.Popen(
-        ["curl", "-i", "http://" + host.hostname_or_ip + path],
+        ["curl", "-i", "http://" + ip + path],
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE
     )
     stdout, _ = process.communicate()
     lines = stdout.decode().splitlines()
     assert lines[0].strip() == "HTTP/1.1 301 Moved Permanently"
-    assert lines[2].strip() == "location:https://" + host.hostname_or_ip + path
+    assert lines[2].strip() == "location:https://" + ip + path
 
 @pytest.mark.usefixtures("host_at_least_8_3")
 class TestHSTS:
     HSTS_HEADER = "strict-transport-security:max-age=63072000"
 
     def __get_header(host):
         process = subprocess.Popen(
-            ["curl", "-XGET", "-k", "-I", "https://" + host.hostname_or_ip],
+            ["curl", "-XGET", "-k", "-I", "https://" + wrap_ip(host.hostname_or_ip)],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE
         )

tests/storage/moosefs/test_moosefs_sr.py

@@ -31,6 +31,7 @@ def test_create_moosefs_sr_without_mfsmount(self, host, moosefs_device_config):
             sr.destroy()
             assert False, "MooseFS SR creation should failed!"
 
+    @pytest.mark.usefixtures("host_no_ipv6")
     def test_create_and_destroy_sr(self, moosefs_device_config, pool_with_moosefs_enabled):
         # Create and destroy tested in the same test to leave the host as unchanged as possible
         master = pool_with_moosefs_enabled.master
@@ -41,7 +42,7 @@ def test_create_and_destroy_sr(self, moosefs_device_config, pool_with_moosefs_en
         vm.destroy(verify=True)
         sr.destroy(verify=True)
 
-@pytest.mark.usefixtures("moosefs_sr")
+@pytest.mark.usefixtures("moosefs_sr", "host_no_ipv6")
 class TestMooseFSSR:
     def test_vdi_is_not_open(self, vdi_on_moosefs_sr):
         assert not vdi_is_open(vdi_on_moosefs_sr)

tests/storage/moosefs/test_moosefs_sr_crosspool_migration.py

@@ -12,7 +12,7 @@
 
 @pytest.mark.small_vm # run with a small VM to test the features
 @pytest.mark.big_vm # and ideally on a big VM to test it scales
-@pytest.mark.usefixtures("hostB1", "local_sr_on_hostB1")
+@pytest.mark.usefixtures("hostB1", "local_sr_on_hostB1", "host_no_ipv6")
 class Test:
     def test_cold_crosspool_migration(self, host, hostB1, vm_on_moosefs_sr, moosefs_sr, local_sr_on_hostB1):
         cold_migration_then_come_back(vm_on_moosefs_sr, host, moosefs_sr, hostB1, local_sr_on_hostB1)

tests/storage/moosefs/test_moosefs_sr_intrapool_migration.py

@@ -12,7 +12,7 @@
 
 @pytest.mark.small_vm # run with a small VM to test the features
 @pytest.mark.big_vm # and ideally on a big VM to test it scales
-@pytest.mark.usefixtures("hostA2", "local_sr_on_hostA2")
+@pytest.mark.usefixtures("hostA2", "local_sr_on_hostA2", "host_no_ipv6")
 class Test:
     def test_live_intrapool_shared_migration(self, host, hostA2, vm_on_moosefs_sr, moosefs_sr):
         live_storage_migration_then_come_back(vm_on_moosefs_sr, host, moosefs_sr, hostA2, moosefs_sr)

tests/xen/test_xtf.py

@@ -8,7 +8,7 @@
 # - host: XCP-ng host >= 8.2, with Xen booted with the hvm_fep command line parameter
 #         See https://xenbits.xen.org/docs/xtf/
 
-@pytest.mark.usefixtures("host_with_hvm_fep", "host_with_dynamically_disabled_ept_sp")
+@pytest.mark.usefixtures("host_with_hvm_fep", "host_with_dynamically_disabled_ept_sp", "host_no_ipv6")
 class TestXtf:
     _common_skips = [
         'test-hvm32-umip',