Refactor subprocess to cmd.py part 1. Fixed clan_uri test.

pkgs/clan-cli/clan_cli/clan_modules.py

@@ -1,17 +1,18 @@
 import json
-import subprocess
 from pathlib import Path
 
 from clan_cli.nix import nix_eval
 
+from .cmd import run
+
 
 def get_clan_module_names(
     flake_dir: Path,
-) -> tuple[list[str], str | None]:
+) -> list[str]:
     """
     Get the list of clan modules from the clan-core flake input
     """
-    proc = subprocess.run(
+    proc = run(
         nix_eval(
             [
                 "--impure",
@@ -25,11 +26,8 @@ def get_clan_module_names(
         """,
             ],
         ),
-        capture_output=True,
-        text=True,
         cwd=flake_dir,
     )
-    if proc.returncode != 0:
+
-        return [], proc.stderr
     module_names = json.loads(proc.stdout)
-    return module_names, None
+    return module_names

pkgs/clan-cli/clan_cli/cmd.py

@@ -1,5 +1,7 @@
 import logging
 import shlex
+import subprocess
+import sys
 from collections.abc import Callable
 from pathlib import Path
 from typing import Any, NamedTuple
@@ -15,10 +17,6 @@ class CmdOut(NamedTuple):
     cwd: Path | None = None
 
 
-import subprocess
-import sys
-
-
 def run(cmd: list[str], cwd: Path = Path.cwd()) -> CmdOut:
     # Start the subprocess
     process = subprocess.Popen(

pkgs/clan-cli/clan_cli/vms/inspect.py

@@ -1,11 +1,9 @@
 import argparse
 import json
-import shlex
-import subprocess
 from dataclasses import dataclass
 from pathlib import Path
 
-from ..errors import ClanError
+from ..cmd import run
 from ..nix import nix_config, nix_eval
 
 
@@ -31,17 +29,8 @@ def inspect_vm(flake_url: str | Path, flake_attr: str) -> VmConfig:
         ]
     )
 
-    proc = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE)
+    proc = run(cmd)
-    assert proc.stdout is not None
+
-    if proc.returncode != 0:
-        raise ClanError(
-            f"""
-command: {shlex.join(cmd)}
-exit code: {proc.returncode}
-stdout:
-{proc.stdout}
-"""
-        )
     data = json.loads(proc.stdout)
     return VmConfig(flake_url=flake_url, flake_attr=flake_attr, **data)
 

pkgs/clan-cli/clan_cli/vms/run.py

@@ -10,6 +10,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import IO
 
+from ..cmd import run
 from ..dirs import module_root, specific_groot_dir
 from ..errors import ClanError
 from ..nix import nix_build, nix_config, nix_shell
@@ -20,9 +21,10 @@ log = logging.getLogger(__name__)
 
 def get_qemu_version() -> list[int]:
     # Run the command and capture the output
-    output = subprocess.check_output(["qemu-kvm", "--version"])
+    res = run(["qemu-kvm", "--version"])
+
     # Decode the output from bytes to string
-    output_str = output.decode("utf-8")
+    output_str = res.stdout
     # Split the output by newline and get the first line
     first_line = output_str.split("\n")[0]
     # Split the first line by space and get the third element
@@ -39,7 +41,7 @@ def graphics_options(vm: VmConfig) -> list[str]:
 
     # Check if the version is greater than 8.1.3 to enable virtio audio
     # if get_qemu_version() > [8, 1, 3]:
-    # common = ["-audio", "driver=pa,model=virtio"]
+    #    common = ["-audio", "driver=pa,model=virtio"]
 
     if vm.wayland:
         # fmt: off
@@ -135,16 +137,7 @@ def get_vm_create_info(vm: VmConfig, nix_options: list[str]) -> dict[str, str]:
         specific_groot_dir(clan_name=vm.clan_name, flake_url=str(vm.flake_url))
         / f"vm-{machine}",
     )
-    proc = subprocess.run(
+    proc = run(cmd)
-        cmd,
-        check=False,
-        stdout=subprocess.PIPE,
-        text=True,
-    )
-    if proc.returncode != 0:
-        raise ClanError(
-            f"Failed to build vm config: {shlex.join(cmd)} failed with: {proc.returncode}"
-        )
     try:
         return json.loads(Path(proc.stdout.strip()).read_text())
     except json.JSONDecodeError as e:

pkgs/clan-cli/tests/test_clan_uri.py

@@ -9,7 +9,7 @@ def test_get_internal() -> None:
     assert uri.get_internal() == "https://example.com?password=1234"
 
     uri = ClanURI("clan://~/Downloads")
-    assert uri.get_internal() == "~/Downloads"
+    assert uri.get_internal().endswith("/Downloads")
 
     uri = ClanURI("clan:///home/user/Downloads")
     assert uri.get_internal() == "/home/user/Downloads"
@@ -42,7 +42,7 @@ def test_is_remote() -> None:
 def test_direct_local_path() -> None:
     # Create a ClanURI object from a remote URI
     uri = ClanURI("clan://~/Downloads")
-    assert uri.get_internal() == "~/Downloads"
+    assert uri.get_internal().endswith("/Downloads")
 
 
 def test_direct_local_path2() -> None:
@@ -109,17 +109,17 @@ def test_from_str() -> None:
     params = ClanParameters(flake_attr="myVM")
     uri = ClanURI.from_str(url=uri_str, params=params)
     assert uri.params.flake_attr == "myVM"
-    assert uri.get_internal() == "~/Downloads/democlan"
+    assert uri.get_internal().endswith("/Downloads/democlan")
 
     uri_str = "~/Downloads/democlan"
     uri = ClanURI.from_str(url=uri_str)
     assert uri.params.flake_attr == "defaultVM"
-    assert uri.get_internal() == "~/Downloads/democlan"
+    assert uri.get_internal().endswith("/Downloads/democlan")
 
     uri_str = "clan://~/Downloads/democlan"
     uri = ClanURI.from_str(url=uri_str)
     assert uri.params.flake_attr == "defaultVM"
-    assert uri.get_internal() == "~/Downloads/democlan"
+    assert uri.get_internal().endswith("/Downloads/democlan")
 
 
 def test_remote_with_all_params() -> None: