Test a joining host has synced certs with joined pool

- Create Host.join_pool and Pool.eject_host methods

Signed-off-by: BenjiReis <benjamin.reis@vates.fr>

conftest.py

@@ -82,38 +82,27 @@ def pytest_configure(config):
     global_config.ignore_ssh_banner = config.getoption('--ignore-ssh-banner')
     global_config.ssh_output_max_lines = int(config.getoption('--ssh-output-max-lines'))
 
-def host_data(hostname_or_ip):
-    # read from data.py
-    from data import HOST_DEFAULT_USER, HOST_DEFAULT_PASSWORD, HOSTS
-    if hostname_or_ip in HOSTS:
-        h_data = HOSTS[hostname_or_ip]
-        return h_data
-    else:
-        return {'user': HOST_DEFAULT_USER, 'password': HOST_DEFAULT_PASSWORD}
-
 def setup_host(hostname_or_ip):
     logging.info(">>> Connect host %s" % hostname_or_ip)
     pool = Pool(hostname_or_ip)
     h = pool.master
     # XO connection
-    h_data = host_data(hostname_or_ip)
-    skip_xo_config = h_data.get('skip_xo_config', False)
-    if not skip_xo_config:
-        h.xo_server_add(h_data['user'], h_data['password'])
+    if not h.skip_xo_config:
+        h.xo_server_add(h.user, h.password)
     else:
         h.xo_get_server_id(store=True)
     wait_for(h.xo_server_connected, timeout_secs=10)
-    return h, skip_xo_config
+    return h
 
 @pytest.fixture(scope='session')
 def hosts(request):
     # a list of master hosts, each from a different pool
     hostname_list = request.param.split(',')
     host_list = [setup_host(hostname_or_ip) for hostname_or_ip in hostname_list]
-    yield [tup[0] for tup in host_list]
+    yield host_list
     # teardown
-    for h, skip_xo_config in host_list:
-        if not skip_xo_config:
+    for h in host_list:
+        if not h.skip_xo_config:
             logging.info("<<< Disconnect host %s" % h)
             h.xo_server_remove()
 

lib/host.py

@@ -13,15 +13,28 @@
 from lib.vm import VM
 from lib.xo import xo_cli, xo_object_exists
 
+def host_data(hostname_or_ip):
+    # read from data.py
+    from data import HOST_DEFAULT_USER, HOST_DEFAULT_PASSWORD, HOSTS
+    if hostname_or_ip in HOSTS:
+        h_data = HOSTS[hostname_or_ip]
+        return h_data
+    else:
+        return {'user': HOST_DEFAULT_USER, 'password': HOST_DEFAULT_PASSWORD}
+
 class Host:
     def __init__(self, pool, hostname_or_ip):
         self.pool = pool
         self.hostname_or_ip = hostname_or_ip
         self.inventory = None
         self.uuid = None
         self.xo_srv_id = None
-        self.user = None
-        self.password = None
+
+        h_data = host_data(self.hostname_or_ip)
+        self.user = h_data['user']
+        self.password = h_data['password']
+        self.skip_xo_config = h_data.get('skip_xo_config', False)
+
         self.saved_packages_list = None
         self.saved_rollback_id = None
         self.inventory = self._get_xensource_inventory()
@@ -345,3 +358,21 @@ def call_plugin(self, plugin_name, function, args=None):
             for k, v in args.items():
                 params['args:%s' % k] = v
         return self.xe('host-call-plugin', params)
+
+    def join_pool(self, pool):
+        master = pool.master
+        self.xe('pool-join', {
+            'master-address': master.hostname_or_ip,
+            'master-username': master.user,
+            'master-password': master.password
+        })
+        wait_for(
+            lambda: self.uuid in pool.hosts_uuids(),
+            f"Wait for joining host {self} to appear in joined pool {master}."
+        )
+        pool.hosts.append(Host(pool, pool.host_ip(self.uuid)))
+        # Do not use `self.is_enabled` since it'd ask the XAPi of hostB1 before the join...
+        wait_for(
+            lambda: master.xe('host-param-get', {'uuid': self.uuid, 'param-name': 'enabled'}),
+            f"Wait for pool {master} to see joined host {self} as enabled."
+        )

lib/pool.py

@@ -2,7 +2,7 @@
 
 import lib.commands as commands
 
-from lib.common import safe_split
+from lib.common import safe_split, wait_for, wait_for_not
 from lib.host import Host
 from lib.sr import SR
 
@@ -136,3 +136,11 @@ def install_custom_uefi_certs(self, auths):
             host.ssh(['secureboot-certs', 'install'] + params)
         finally:
             host.ssh(['rm', '-f'] + list(auths_dict.values()))
+
+    def eject_host(self, host):
+        master = self.master
+        master.xe('pool-eject', {'host-uuid': host.uuid, 'force': True})
+        wait_for_not(lambda: host.uuid in self.hosts_uuids(), "Wait for host {host} to be ejected of pool {master}.")
+        hosts = self.hosts
+        self.hosts = [h for h in hosts if h.uuid != host.uuid]
+        wait_for(host.is_enabled, f"Wait for host {host} to restart in its own pool.", timeout_secs=600)

tests/uefistored/test_cert_inheritance.py

@@ -13,6 +13,8 @@
 # - host(A1): XCP-ng host >= 8.2 (+ updates) (or >= 8.3 for other tests)
 #   with UEFI certs either absent, or present and consistent (state will be saved and restored)
 #   Master of a, at least, 2 hosts pool
+# - hostB1: XCP-ng host >= 8.3
+#   This host will be joined and ejected from pool A, it means its state will be completely reinitialized from scratch
 
 CERT_DIR = "/var/lib/uefistored"
 
@@ -330,3 +332,32 @@ def test_clear_certificates_from_pool(self, host):
             logging.info(f"Check host {h} has no certificate on disk.")
             for key in keys:
                 assert not h.file_exists(f'{CERT_DIR}/{key}.auth')
+
+@pytest.mark.usefixtures("host_at_least_8_3", "pool_without_uefi_certs")
+class TestPoolToDiskCertInheritanceOnJoin:
+    @pytest.fixture(scope='function')
+    def keys_auths_for_joined_host(self, host, hostB1):
+        from packaging import version
+        version_str = "8.3"
+        if not hostB1.xcp_version >= version.parse(version_str):
+            pytest.skip(f"This test requires an second pool XCP-ng >= {version_str} host")
+
+        # Install certs before host join
+        keys = ['PK', 'KEK', 'db', 'dbx']
+        pool_auths = generate_keys(as_dict=True)
+        host.pool.install_custom_uefi_certs([pool_auths[key] for key in keys])
+
+        logging.info(f"> Join host {hostB1} to pool {host} after certificates installed.")
+        hostB1.join_pool(host.pool)
+        joined_host = host.pool.get_host_by_uuid(hostB1.uuid)
+        yield keys, pool_auths, joined_host
+
+        logging.info(f"< Eject host {joined_host} from pool {host}.")
+        # Warning: triggers a reboot of ejected host.
+        host.pool.eject_host(joined_host)
+        host.pool.clear_uefi_certs()
+
+    def test_host_certificates_updated_after_join(self, keys_auths_for_joined_host):
+        keys, pool_auths, joined_host = keys_auths_for_joined_host
+        for key in keys:
+            check_disk_cert_md5sum(joined_host, key, pool_auths[key].auth)