Adapt tests for IPv6 hosts

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
 from lib.xo import xo_cli
@@ -176,6 +177,11 @@ 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='session')
 def local_sr_on_hostA1(hostA1):
     """ A local SR on the pool's master. """
@@ -375,7 +381,8 @@ def second_network(request, host):
     pif_uuid = host.xe('pif-list', {'host-uuid': host.uuid, 'network-uuid': network_uuid}, minimal=True)
     if not pif_uuid:
         pytest.fail("The provided --second-network UUID doesn't exist or doesn't have a PIF on master host")
-    ip = host.xe('pif-param-get', {'uuid': pif_uuid, 'param-name': 'IP'})
+    ipv6 = (host.xe('pif-param-get', {'uuid': pif_uuid, 'param-name': 'primary-address-type'}) == "IPv6")
+    ip = host.xe('pif-param-get', {'uuid': pif_uuid, 'param-name': 'IPv6' if ipv6 else 'IP'})
     if not ip:
         pytest.fail("The provided --second-network has a PIF but no IP")
     if network_uuid == host.management_network():

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
@@ -147,7 +148,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']
@@ -159,7 +160,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):
@@ -171,7 +172,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',
@@ -183,7 +184,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/migration/test_host_evacuate.py

@@ -111,5 +111,8 @@ def test_host_evacuate_with_network_not_present(self, host, hostA2, second_netwo
             pif_uuid = host.xe('pif-list', {'host-uuid': hostA2.uuid, 'network-uuid': second_network}, minimal=True)
             args['uuid'] = pif_uuid
             logging.info(f"Re-add and plug PIF {pif_uuid}")
+            if ipv6:
+                # Default is IPv4
+                host.xe('pif-set-primary-address-type', {'uuid': pif_uuid, 'primary_address_type': 'ipv6'})
             host.xe(reconfigure_method, args)
             host.xe('pif-plug', {'uuid': pif_uuid})

tests/misc/test_file_server.py

@@ -2,6 +2,8 @@
 import re
 import subprocess
 
+from lib.netutil import wrap_ip
+
 # These tests are meant to test an host fileserver behavior.
 #
 # Requirements:
@@ -14,15 +16,16 @@ def _header_equal(header, name, value):
 
 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 _header_equal(lines[2], "location", "https://" + host.hostname_or_ip + path)
+    assert _header_equal(lines[2], "location", "https://" + ip + path)
 
 @pytest.mark.usefixtures("host_at_least_8_3")
 class TestHSTS:
@@ -31,7 +34,7 @@ class TestHSTS:
 
     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,8 @@ def test_create_moosefs_sr_without_mfsmount(self, host, moosefs_device_config):
             sr.destroy()
             assert False, "MooseFS SR creation should failed!"
 
+    # MooseFS doesn't support IPv6
+    @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 +43,8 @@ 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")
+# MooseFS doesn't support IPv6
+@pytest.mark.usefixtures("moosefs_sr", "host_no_ipv6")
 class TestMooseFSSR:
     @pytest.mark.quicktest
     def test_quicktest(self, 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") # MooseFS doesn't support 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") # MooseFS doesn't support 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,8 @@
 # - 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")
+# XTF git repo is not available in IPv6
+@pytest.mark.usefixtures("host_with_hvm_fep", "host_with_dynamically_disabled_ept_sp", "host_no_ipv6")
 class TestXtf:
     _common_skips = [
         # UMIP requires hardware support, that is a recent enough CPU