Merge pull request #572 from xcp-ng/srt/xo-cli

the PR is instroducing some xo_cli changes:

-  use TOOLS variable in data.py to get the path of xo-cli command-line : it permits to use custom path. the default is to use xo_cli as currently
- add logging.debug() when xo-cli is invoked in order to trace it (by using local_cmd)
- simplify xo_cli function and unify simple_output and use_json arguments

Author: semarie

data.py-dist

@@ -116,6 +116,7 @@ OTHER_GUEST_TOOLS = {
 # Tools
 TOOLS: dict[str, str] = {
 #    "iso-remaster": "/home/user/src/xcpng/xcp/scripts/iso-remaster/iso-remaster.sh",
+#    "xo-cli": "xo-cli",
 }
 
 # Values can be either full URLs or only partial URLs that will be automatically appended to DEF_VM_URL

lib/xo.py

@@ -1,50 +1,32 @@
 import json
-import subprocess
 
+from data import TOOLS
+from lib.commands import local_cmd
 from lib.typing import JSONType
 
-from typing import Dict, Literal, overload
+from typing import Literal, overload
 
-# TODO: either:
-#   * replace simple_output and use_json by a single output type
-#   * make sure that simple_output=False and use_json=True are not being used together
-
-@overload
-def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: Literal[True] = True,
-           use_json: Literal[False] = False) -> str:
-    ...
-@overload
-def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: Literal[True] = True,
-           use_json: Literal[True]) -> JSONType:
-    ...
 @overload
-def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: Literal[False],
-           use_json: bool = False) -> subprocess.CompletedProcess[bytes]:
+def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_json: Literal[False] = False) -> str:
     ...
 @overload
-def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: bool = True,
-           use_json: bool = False) -> subprocess.CompletedProcess[bytes] | JSONType | str:
+def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_json: Literal[True]) -> JSONType:
     ...
-def xo_cli(
-    action: str, args: dict[str, str] = {}, check: bool = True, simple_output: bool = True, use_json: bool = False
-) -> subprocess.CompletedProcess[bytes] | JSONType | str:
-    run_array = ['xo-cli', action]
+
+def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_json: bool = False) -> JSONType | str:
+    cmd = [TOOLS.get('xo-cli', '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
-    )
-    if simple_output:
-        output = res.stdout.decode().strip()
-        if use_json:
-            return json.loads(output)
-        return output
-    return res
+        cmd += ['--json']
+    cmd += ["%s=%s" % (key, value) for key, value in args.items()]
+
+    res = local_cmd(cmd, check=check)
+
+    if use_json:
+        return json.loads(res.stdout)
+
+    return res.stdout
 
 def xo_object_exists(uuid: str) -> bool:
-    lst = json.loads(xo_cli('--list-objects', {'uuid': uuid}))
+    lst = xo_cli('list-objects', {'uuid': uuid}, use_json=True)
+    assert isinstance(lst, list)
     return len(lst) > 0