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,10 @@ 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}.")
+ self.hosts = [h for h in self.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 a second XCP-ng pool with version >= {version_str}")
+
+ # 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)