Replace manual process handling by local_cmd()

Author: guzu

conftest.py

@@ -12,6 +12,7 @@
 
 import lib.config as global_config
 from lib import pxe
+from lib import commands
 from lib.common import (
     callable_marker,
     DiskDevName,
@@ -202,7 +203,8 @@ 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"),
+            # TODO: put that in lib/netutils ?
+            wait_for(lambda: commands.local_cmd(['nc', '-zw5', host_vm.ip, '22'], check=False),
                      "Wait for ssh up on nested host", retry_delay_secs=5)
 
             hostname_or_ip = host_vm.ip

lib/host.py

@@ -551,9 +551,9 @@ 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(lambda: commands.local_cmd(['ping', '-c1', self.hostname_or_ip], check=False),
                      "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(lambda: commands.local_cmd(['nc', '-zw5', self.hostname_or_ip, '22'], check=False),
                      "Wait for ssh up on host", timeout_secs=10 * 60, retry_delay_secs=5)
             wait_for(self.is_enabled, "Wait for XAPI to be ready", timeout_secs=30 * 60)
 

lib/xo.py

@@ -1,5 +1,6 @@
 import json
-import subprocess
+
+from lib import commands
 
 from typing import Any, Dict, Literal, Union, overload
 
@@ -13,29 +14,28 @@ 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) -> int:
     ...
 @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[int, 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)
+
     if simple_output:
-        output = res.stdout.decode().strip()
+        output = res.stdout
         if use_json:
             return json.loads(output)
         return output
-    return res
+
+    # XXX: doesn't seem to be used as by default simple_output is true
+    return res.returncode
 
 def xo_object_exists(uuid):
     lst = json.loads(xo_cli('--list-objects', {'uuid': 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,7 @@
 
 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()
+    encrypted_password = local_cmd(['openssl', 'passwd', '-6', password]).output.strip()
     if target_hostname is None:
         target_hostname = "xcp-ng-" + "".join(
             random.choice(string.ascii_lowercase) for i in range(5)
@@ -70,7 +67,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).returncode == 0
 
 def is_ssh_up(ip):
     try:

tests/fs_diff/test_fs_diff.py

@@ -1,7 +1,9 @@
+
 import pytest
 
 import os
-import subprocess
+
+from lib.commands import local_cmd
 
 # Requirements:
 # - 2 XCP-ng host of same version
@@ -14,13 +16,7 @@ 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()
-
-    if process.returncode != 0:
-        print(stdout.decode())
-
-    assert process.returncode == 0
+    res = local_cmd([fsdiff, "--reference-host", f"{hosts[0]}",
+                             "--test-host", f"{hosts[1]}",
+                             "--json-output"])
+    assert res.returncode == 0

tests/misc/test_access_links.py

@@ -1,7 +1,6 @@
 import pytest
 
 import hashlib
-import subprocess
 
 from lib import commands
 

tests/network/test_vif_allowed_ip.py

@@ -3,12 +3,16 @@
 import ipaddress
 import os
 
+from lib.commands import local_cmd
+
 # Requirements:
 # - one XCP-ng host (--host) >= 8.2 (>= 8.3 for the CIDR tests) with no SDN controller configured
 # - a VM (--vm)
 
 def ip_responsive(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).returncode == 0
 
 @pytest.mark.small_vm
 @pytest.mark.usefixtures("host_no_sdn_controller")

tests/packages/bugtool/test_bugtool.py

@@ -1,7 +1,5 @@
 import pytest
 
-import subprocess
-
 from lib.host import Host
 
 # This smoke test runs xen-bugtool and verifies that the archive it generates

tests/packages/netdata/test_netdata.py

@@ -1,7 +1,6 @@
 import pytest
 
-import subprocess
-
+from lib.commands import local_cmd
 from lib.netutil import wrap_ip
 
 # This test installs netdata and netdata-ui packages, verifies the service status,
@@ -17,13 +16,9 @@ def __get_headers(host, port, path=None):
         url = f"http://{wrap_ip(host.hostname_or_ip)}:{port}"
         if path is not None:
             url += f"/{path}"
-        process = subprocess.Popen(
-            ["curl", "-XGET", "-k", "-I", "-s", url],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE
-        )
-        stdout, _ = process.communicate()
-        return stdout.decode().splitlines()
+
+        res = local_cmd(["curl", "-XGET", "-k", "-I", "-s", url])
+        return res.output.splitlines()
 
     # Verify the ActiveState for the netdata service
     def test_netdata_service(self, host):