Replace manual process handling by local_cmd()

.github/workflows/code-checkers.yml

@@ -38,7 +38,7 @@ jobs:
       - uses: ./.github/actions/uv-setup/
       - name: Create a dummy data.py
         run: cp data.py-dist data.py
-      - run: ruff check lib/ tests/
+      - run: ruff check conftest.py lib/ tests/
 
   flake8:
     runs-on: ubuntu-latest

conftest.py

@@ -4,7 +4,6 @@
 
 import itertools
 import logging
-import os
 import tempfile
 
 import git
@@ -13,19 +12,17 @@
 import lib.config as global_config
 from lib import pxe
 from lib.common import (
-    callable_marker,
     DiskDevName,
     HostAddress,
+    callable_marker,
     is_uuid,
     prefix_object_name,
-    setup_formatted_and_mounted_disk,
     shortened_nodeid,
-    teardown_formatted_and_mounted_disk,
     vm_image,
     wait_for,
 )
-from lib.netutil import is_ipv6
 from lib.host import Host
+from lib.netutil import is_ipv6, wait_for_ssh
 from lib.pool import Pool
 from lib.sr import SR
 from lib.vm import VM, vm_cache_key_from_def
@@ -34,7 +31,9 @@
 # Import package-scoped fixtures. Although we need to define them in a separate file so that we can
 # then import them in individual packages to fix the buggy package scope handling by pytest, we also
 # need to import them in the global conftest.py so that they are recognized as fixtures.
-from pkgfixtures import formatted_and_mounted_ext4_disk, sr_disk_wiped
+# Linters might be confused by that and see these imports as unused. The `noqa` suppresses the
+# false-positive warnings.
+from pkgfixtures import formatted_and_mounted_ext4_disk, sr_disk_wiped  # noqa
 
 from typing import Dict, Generator, Iterable
 
@@ -202,8 +201,7 @@ def setup_host(hostname_or_ip, *, config=None):
             assert len(ips) == 1
             host_vm.ip = ips[0]
 
-            wait_for(lambda: not os.system(f"nc -zw5 {host_vm.ip} 22"),
-                     "Wait for ssh up on nested host", retry_delay_secs=5)
+            wait_for_ssh(host_vm.ip, host_desc="nested host")
 
             hostname_or_ip = host_vm.ip
 
@@ -321,7 +319,7 @@ def xfail_on_xcpng_8_3(host, request):
 @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.skip("This test requires an IPv4 XCP-ng")
 
 @pytest.fixture(scope="session")
 def shared_sr(host):
@@ -442,9 +440,7 @@ def vm_ref(request):
             logging.info(">> No VM specified on CLI, and no default found in test definition. Using global default.")
             ref = 'mini-linux-x86_64-bios'
 
-    if is_uuid(ref):
-        return ref
-    elif ref.startswith('http'):
+    if is_uuid(ref) or ref.startswith('http'):
         return ref
     else:
         return vm_image(ref)

lib/commands.py

@@ -5,7 +5,6 @@
 
 import lib.config as config
 from lib.common import HostAddress
-from lib.netutil import wrap_ip
 
 from typing import List, Literal, Union, overload
 
@@ -190,6 +189,9 @@ def ssh_with_result(hostname_or_ip, cmd, suppress_fingerprint_warnings=True,
     assert False, "unexpected type"
 
 def scp(hostname_or_ip, src, dest, check=True, suppress_fingerprint_warnings=True, local_dest=False):
+    # local import to avoid cyclic import; lib.netutils also import lib.commands
+    from lib.netutil import wrap_ip
+
     opts = '-o "BatchMode yes"'
     if suppress_fingerprint_warnings:
         # Suppress warnings and questions related to host key fingerprints
@@ -244,7 +246,24 @@ def sftp(hostname_or_ip, cmds, check=True, suppress_fingerprint_warnings=True):
 
     return res
 
-def local_cmd(cmd, check=True, decode=True):
+@overload
+def local_cmd(cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[True] = True,
+              decode: Literal[True] = True) -> str:
+    ...
+@overload
+def local_cmd(cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[True] = True,
+              decode: Literal[False]) -> bytes:
+    ...
+@overload
+def local_cmd(cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[False],
+              decode: bool = True) -> LocalCommandResult:
+    ...
+@overload
+def local_cmd(cmd: Union[str, List[str]], *, check: bool = True, simple_output: bool = True,
+              decode: bool = True) -> Union[str, bytes, LocalCommandResult]:
+    ...
+
+def local_cmd(cmd, check=True, simple_output=True, decode=True):
     """ Run a command locally on tester end. """
     logging.debug("[local] %s", (cmd,))
     res = subprocess.run(
@@ -257,17 +276,18 @@ def local_cmd(cmd, check=True, decode=True):
     # get a decoded version of the output in any case, replacing potential errors
     output_for_logs = res.stdout.decode(errors='replace').strip()
 
-    output = res.stdout
-    if decode:
-        output = output.decode()
-
     errorcode_msg = "" if res.returncode == 0 else " - Got error code: %s" % res.returncode
     command = " ".join(cmd)
     logging.debug(f"[local] {command}{errorcode_msg}{_ellide_log_lines(output_for_logs)}")
 
     if res.returncode and check:
         raise LocalCommandFailed(res.returncode, output_for_logs, command)
 
+    output: Union[bytes, str] = res.stdout
+    if decode:
+        output = output.decode()
+    if simple_output:
+        return output.strip()
     return LocalCommandResult(res.returncode, output)
 
 def encode_powershell_command(cmd: str):

lib/efi.py

@@ -460,6 +460,7 @@ def sign_image(self, image: str) -> str:
         Returns path to signed image.
         """
         assert self._owner_cert is not None
+        assert self._owner_cert.key is not None
         if shutil.which('sbsign'):
             signed = get_signed_name(image)
             commands.local_cmd([

lib/host.py

@@ -32,7 +32,7 @@
     wait_for,
     wait_for_not,
 )
-from lib.netutil import wrap_ip
+from lib.netutil import wait_for_ssh, wrap_ip
 from lib.sr import SR
 from lib.vdi import VDI
 from lib.vm import VM
@@ -551,10 +551,7 @@ def reboot(self, verify=False):
                 raise
         if verify:
             wait_for_not(self.is_enabled, "Wait for host down")
-            wait_for(lambda: not os.system(f"ping -c1 {self.hostname_or_ip} > /dev/null 2>&1"),
-                     "Wait for host up", timeout_secs=10 * 60, retry_delay_secs=10)
-            wait_for(lambda: not os.system(f"nc -zw5 {self.hostname_or_ip} 22"),
-                     "Wait for ssh up on host", timeout_secs=10 * 60, retry_delay_secs=5)
+            wait_for_ssh(self.hostname_or_ip)
             wait_for(self.is_enabled, "Wait for XAPI to be ready", timeout_secs=30 * 60)
 
     def management_network(self):

lib/netutil.py

@@ -1,5 +1,20 @@
 import socket
 
+from lib.commands import local_cmd
+from lib.common import wait_for
+
+def wait_for_tcp_port(host, port, port_desc, ping=True, host_desc=None):
+    if host_desc:
+        host_desc = f"({host_desc})"
+    if ping:
+        wait_for(lambda: local_cmd(['ping', '-c1', host], check=False, simple_output=False).returncode == 0,
+                 "Wait for host up (ICMP ping)", timeout_secs=10 * 60, retry_delay_secs=10)
+    wait_for(lambda: local_cmd(['nc', '-zw5', host, str(port)], check=False, simple_output=False).returncode == 0,
+             f"Wait for {port_desc} up on host {host_desc}", timeout_secs=10 * 60, retry_delay_secs=5)
+
+def wait_for_ssh(host, host_desc=None, ping=True):
+    wait_for_tcp_port(host, 22, "SSH", ping, host_desc)
+
 def is_ipv6(ip):
     try:
         socket.inet_pton(socket.AF_INET6, ip)

lib/xo.py

@@ -1,5 +1,7 @@
 import json
-import subprocess
+
+from lib import commands
+from lib.commands import LocalCommandResult
 
 from typing import Any, Dict, Literal, Union, overload
 
@@ -13,28 +15,26 @@ def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple
     ...
 @overload
 def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: Literal[False],
-           use_json: bool = False) -> subprocess.CompletedProcess:
+           use_json: bool = False) -> LocalCommandResult:
     ...
 @overload
 def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: bool = True,
-           use_json: bool = False) -> Union[subprocess.CompletedProcess, Any, str]:
+           use_json: bool = False) -> Union[LocalCommandResult, Any, str]:
     ...
 def xo_cli(action, args={}, check=True, simple_output=True, use_json=False):
     run_array = ['xo-cli', action]
     if use_json:
         run_array += ['--json']
     run_array += ["%s=%s" % (key, value) for key, value in args.items()]
-    res = subprocess.run(
-        run_array,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-        check=check
-    )
+
+    res = commands.local_cmd(run_array, check=check, decode=True, simple_output=False)
+
     if simple_output:
-        output = res.stdout.decode().strip()
+        output = res.stdout.strip()
         if use_json:
             return json.loads(output)
         return output
+
     return res
 
 def xo_object_exists(uuid):

scripts/install_xcpng.py

@@ -6,7 +6,6 @@
 import os
 import random
 import string
-import subprocess
 import sys
 import tempfile
 import time
@@ -17,7 +16,7 @@
 # flake8: noqa: E402
 sys.path.append(f"{os.path.abspath(os.path.dirname(__file__))}/..")
 from lib import pxe
-from lib.commands import SSHCommandFailed, scp, ssh
+from lib.commands import SSHCommandFailed, scp, ssh, local_cmd
 from lib.common import is_uuid, wait_for
 from lib.host import host_data
 from lib.pool import Pool
@@ -27,9 +26,8 @@
 
 def generate_answerfile(directory, installer, hostname_or_ip, target_hostname, action, hdd, netinstall_gpg_check):
     password = host_data(hostname_or_ip)['password']
-    cmd = ['openssl', 'passwd', '-6', password]
-    res = subprocess.run(cmd, stdout=subprocess.PIPE)
-    encrypted_password = res.stdout.decode().strip()
+    algorithm_option = '-6' # SHA512
+    encrypted_password = local_cmd(['openssl', 'passwd', algorithm_option, password])
     if target_hostname is None:
         target_hostname = "xcp-ng-" + "".join(
             random.choice(string.ascii_lowercase) for i in range(5)
@@ -70,7 +68,9 @@ def generate_answerfile(directory, installer, hostname_or_ip, target_hostname, a
             raise Exception(f"Unknown action: `{action}`")
 
 def is_ip_active(ip):
-    return not os.system(f"ping -c 3 -W 10 {ip} > /dev/null 2>&1")
+    # 3 tries with a timeout of 10 sec for each ICMP request
+    return local_cmd(['ping', '-c', '3', '-W', '10', ip],
+                     check=False, simple_output=False).returncode == 0
 
 def is_ssh_up(ip):
     try:
@@ -201,7 +201,7 @@ def main():
             "Waiting for the installation process to complete and the VM to reboot and be up", 3600, 10
         )
         vm_ip_address = get_new_host_ip(mac_address)
-        logging.info('The IP address of the installed XCP-ng is: ' + vm_ip_address)
+        logging.info(f'The IP address of the installed XCP-ng is: {vm_ip_address}')
         wait_for(lambda: is_new_host_ready(vm_ip_address), "Waiting for XAPI to be ready", 600, 10)
         pool2 = Pool(vm_ip_address)
         host2 = pool2.master

tests/fs_diff/test_fs_diff.py

@@ -1,7 +1,8 @@
 import pytest
 
 import os
-import subprocess
+
+from lib.commands import local_cmd
 
 # Requirements:
 # - 2 XCP-ng host of same version
@@ -14,13 +15,11 @@ def test_fs_diff(hosts):
 
     fsdiff = os.path.realpath(f"{os.path.dirname(__file__)}/../../scripts/xcpng-fs-diff.py")
 
-    process = subprocess.Popen(
-        [fsdiff, "--reference-host", f"{hosts[0]}", "--test-host", f"{hosts[1]}", "--json-output"],
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE
-    )
-    stdout, _ = process.communicate()
+    res = local_cmd([fsdiff, "--reference-host", f"{hosts[0]}",
+                             "--test-host", f"{hosts[1]}",
+                             "--json-output"], simple_output=False)
 
-    if process.returncode != 0:
-        print(stdout.decode())
+    if res.returncode != 0:
+        print(res.stdout)
 
-    assert process.returncode == 0
+    assert res.returncode == 0

tests/install/conftest.py

@@ -11,6 +11,7 @@
 from lib.commands import local_cmd
 from lib.common import callable_marker, url_download, wait_for
 from lib.installer import AnswerFile
+from lib.netutil import wait_for_ssh
 
 from typing import Generator, Sequence, Union
 
@@ -292,10 +293,7 @@ def vm_booted_with_installer(host, create_vms, remastered_iso):
             host_vm.ip = ips[0]
 
             # host may not be up if ARP cache was filled
-            wait_for(lambda: local_cmd(["ping", "-c1", host_vm.ip], check=False),
-                     "Wait for host up", timeout_secs=10 * 60, retry_delay_secs=10)
-            wait_for(lambda: local_cmd(["nc", "-zw5", host_vm.ip, "22"], check=False),
-                     "Wait for ssh up on host", timeout_secs=10 * 60, retry_delay_secs=5)
+            wait_for_ssh(host_vm.ip)
 
             yield host_vm
 

tests/install/test.py

@@ -7,6 +7,7 @@
 from lib import commands, installer, pxe
 from lib.common import safe_split, wait_for
 from lib.installer import AnswerFile
+from lib.netutil import wait_for_ssh
 from lib.pif import PIF
 from lib.pool import Pool
 from lib.vdi import VDI
@@ -187,10 +188,7 @@ def _test_firstboot(self, create_vms, mode, *, machine='DEFAULT', is_restore=Fal
             assert len(ips) == 1
             host_vm.ip = ips[0]
 
-            wait_for(
-                lambda: commands.local_cmd(
-                    ["nc", "-zw5", host_vm.ip, "22"], check=False).returncode == 0,
-                "Wait for ssh back up on Host VM", retry_delay_secs=5, timeout_secs=4 * 60)
+            wait_for_ssh(host_vm.ip, host_desc="host VM")
 
             logging.info("Checking installed version (expecting %r %r)",
                          expected_dist, expected_rel)

tests/misc/test_access_links.py

@@ -1,7 +1,6 @@
 import pytest
 
 import hashlib
-import subprocess
 
 from lib import commands
 
@@ -35,10 +34,10 @@ def test_access_links(host, command_id, url_id):
 
     # Verify the download worked by comparing with local download
     # This ensures the content is accessible and identical from both locations
-    local_result = commands.local_cmd(COMMAND)
+    local_result = commands.local_cmd(COMMAND, check=False, simple_output=False)
 
     assert local_result.returncode == 0, (
-        f"Failed to fetch URL locally: {local_result.stderr}"
+        f"Failed to fetch URL locally: {local_result.stdout}"
     )
 
     # Extract checksums