yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

clan-cli: cmd.run now has its options extracted to a dataclass

Browse Source
This commit is contained in:
Qubasa
2024-11-28 15:26:37 +01:00
parent d672c08074
commit d2719f3179
31 changed files with 218 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pkgs/clan-cli/clan_cli/api/directory.py
View File

@@ -5,7 +5,7 @@ from pathlib import Path
from typing import Any, Literal from typing import Any, Literal
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
from clan_cli.nix import nix_shell, run_no_stdout from clan_cli.nix import nix_shell, run_no_output
from . import API from . import API
@@ -154,7 +154,7 @@ def show_block_devices(options: BlockDeviceOptions) -> Blockdevices:
"PATH,NAME,RM,SIZE,RO,MOUNTPOINTS,TYPE,ID-LINK", "PATH,NAME,RM,SIZE,RO,MOUNTPOINTS,TYPE,ID-LINK",
], ],
) )
proc = run_no_stdout(cmd, needs_user_terminal=True) proc = run_no_output(cmd, needs_user_terminal=True)
res = proc.stdout.strip() res = proc.stdout.strip()
blk_info: dict[str, Any] = json.loads(res) blk_info: dict[str, Any] = json.loads(res)

4
pkgs/clan-cli/clan_cli/api/mdns_discovery.py
View File

@@ -2,7 +2,7 @@ import argparse
import re import re
from dataclasses import dataclass from dataclasses import dataclass
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_output
from clan_cli.nix import nix_shell from clan_cli.nix import nix_shell
from . import API from . import API
@@ -100,7 +100,7 @@ def show_mdns() -> DNSInfo:
"--terminate", "--terminate",
], ],
) )
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
data = parse_avahi_output(proc.stdout) data = parse_avahi_output(proc.stdout)
return data return data

6
pkgs/clan-cli/clan_cli/api/modules.py
View File

@@ -5,7 +5,7 @@ from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from typing import Any, TypedDict, get_args, get_type_hints from typing import Any, TypedDict, get_args, get_type_hints
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_output
from clan_cli.errors import ClanCmdError, ClanError from clan_cli.errors import ClanCmdError, ClanError
from clan_cli.inventory import Inventory, load_inventory_json, set_inventory from clan_cli.inventory import Inventory, load_inventory_json, set_inventory
from clan_cli.inventory.classes import Service from clan_cli.inventory.classes import Service
@@ -150,7 +150,7 @@ def get_modules(base_path: str) -> dict[str, str]:
] ]
) )
try: try:
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()
except ClanCmdError as e: except ClanCmdError as e:
msg = "clanInternals might not have inventory.modules attributes" msg = "clanInternals might not have inventory.modules attributes"
@@ -171,7 +171,7 @@ def get_module_interface(base_path: str, module_name: str) -> dict[str, Any]:
""" """
cmd = nix_eval([f"{base_path}#clanInternals.moduleSchemas.{module_name}", "--json"]) cmd = nix_eval([f"{base_path}#clanInternals.moduleSchemas.{module_name}", "--json"])
try: try:
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()
except ClanCmdError as e: except ClanCmdError as e:
msg = "clanInternals might not have moduleSchemas attributes" msg = "clanInternals might not have moduleSchemas attributes"

14
pkgs/clan-cli/clan_cli/clan/create.py
View File

@@ -5,7 +5,7 @@ from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from clan_cli.api import API from clan_cli.api import API
from clan_cli.cmd import CmdOut, run from clan_cli.cmd import CmdOut, RunOpts, run
from clan_cli.dirs import TemplateType, clan_templates from clan_cli.dirs import TemplateType, clan_templates
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
from clan_cli.inventory import Inventory, init_inventory from clan_cli.inventory import Inventory, init_inventory
@@ -61,10 +61,10 @@ def create_clan(options: CreateOptions) -> CreateClanResponse:
template_url, template_url,
] ]
) )
flake_init = run(command, cwd=directory) flake_init = run(command, RunOpts(cwd=directory))
flake_update = run( flake_update = run(
nix_shell(["nixpkgs#nix"], ["nix", "flake", "update"]), cwd=directory nix_shell(["nixpkgs#nix"], ["nix", "flake", "update"]), RunOpts(cwd=directory)
) )
if options.initial: if options.initial:
@@ -81,14 +81,18 @@ def create_clan(options: CreateOptions) -> CreateClanResponse:
response.git_add = run(git_command(directory, "add", ".")) response.git_add = run(git_command(directory, "add", "."))
# check if username is set # check if username is set
has_username = run(git_command(directory, "config", "user.name"), check=False) has_username = run(
git_command(directory, "config", "user.name"), RunOpts(check=False)
)
response.git_config_username = None response.git_config_username = None
if has_username.returncode != 0: if has_username.returncode != 0:
response.git_config_username = run( response.git_config_username = run(
git_command(directory, "config", "user.name", "clan-tool") git_command(directory, "config", "user.name", "clan-tool")
) )
has_username = run(git_command(directory, "config", "user.email"), check=False) has_username = run(
git_command(directory, "config", "user.email"), RunOpts(check=False)
)
if has_username.returncode != 0: if has_username.returncode != 0:
response.git_config_email = run( response.git_config_email = run(
git_command(directory, "config", "user.email", "clan@example.com") git_command(directory, "config", "user.email", "clan@example.com")

4
pkgs/clan-cli/clan_cli/clan/show.py
View File

@@ -5,7 +5,7 @@ from pathlib import Path
from urllib.parse import urlparse from urllib.parse import urlparse
from clan_cli.api import API from clan_cli.api import API
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_output
from clan_cli.errors import ClanCmdError, ClanError from clan_cli.errors import ClanCmdError, ClanError
from clan_cli.inventory import Meta from clan_cli.inventory import Meta
from clan_cli.nix import nix_eval from clan_cli.nix import nix_eval
@@ -24,7 +24,7 @@ def show_clan_meta(uri: str | Path) -> Meta:
res = "{}" res = "{}"
try: try:
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()
except ClanCmdError as e: except ClanCmdError as e:
msg = "Evaluation failed on meta attribute" msg = "Evaluation failed on meta attribute"

96
pkgs/clan-cli/clan_cli/cmd.py
View File

@@ -241,57 +241,65 @@ if os.environ.get("CLAN_CLI_PERF"):
TIME_TABLE = TimeTable() TIME_TABLE = TimeTable()
@dataclass
class RunOpts:
input: bytes | None = None
stdout: IO[bytes] | None = None
stderr: IO[bytes] | None = None
env: dict[str, str] | None = None
cwd: Path | None = None
log: Log = Log.STDERR
prefix: str | None = None
msg_color: MsgColor | None = None
check: bool = True
error_msg: str | None = None
needs_user_terminal: bool = False
timeout: float = math.inf
shell: bool = False
def run( def run(
cmd: list[str], cmd: list[str],
*, options: RunOpts | None = None,
input: bytes | None = None, # noqa: A002
stdout: IO[bytes] | None = None,
stderr: IO[bytes] | None = None,
env: dict[str, str] | None = None,
cwd: Path | None = None,
log: Log = Log.STDERR,
prefix: str | None = None,
msg_color: MsgColor | None = None,
check: bool = True,
error_msg: str | None = None,
needs_user_terminal: bool = False,
timeout: float = math.inf,
shell: bool = False,
) -> CmdOut: ) -> CmdOut:
if cwd is None: if options is None:
cwd = Path.cwd() options = RunOpts()
if options.cwd is None:
options.cwd = Path.cwd()
if prefix is None: if options.prefix is None:
prefix = "$" options.prefix = "$"
if input: if options.input:
if any(not ch.isprintable() for ch in input.decode("ascii", "replace")): if any(not ch.isprintable() for ch in options.input.decode("ascii", "replace")):
filtered_input = "<<binary_blob>>" filtered_input = "<<binary_blob>>"
else: else:
filtered_input = input.decode("ascii", "replace") filtered_input = options.input.decode("ascii", "replace")
print_trace( print_trace(
f"$: echo '{filtered_input}' | {indent_command(cmd)}", cmdlog, prefix f"$: echo '{filtered_input}' | {indent_command(cmd)}",
cmdlog,
options.prefix,
) )
elif cmdlog.isEnabledFor(logging.DEBUG): elif cmdlog.isEnabledFor(logging.DEBUG):
print_trace(f"$: {indent_command(cmd)}", cmdlog, prefix) print_trace(f"$: {indent_command(cmd)}", cmdlog, options.prefix)
start = timeit.default_timer() start = timeit.default_timer()
with ExitStack() as stack: with ExitStack() as stack:
stdin = subprocess.PIPE if input is not None else None stdin = subprocess.PIPE if options.input is not None else None
process = stack.enter_context( process = stack.enter_context(
subprocess.Popen( subprocess.Popen(
cmd, cmd,
cwd=str(cwd), cwd=str(options.cwd),
env=env, env=options.env,
stdin=stdin, stdin=stdin,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
start_new_session=not needs_user_terminal, start_new_session=not options.needs_user_terminal,
shell=shell, shell=options.shell,
) )
) )
if needs_user_terminal: if options.needs_user_terminal:
# we didn't allocate a new session, so we can't terminate the process group # we didn't allocate a new session, so we can't terminate the process group
stack.enter_context(terminate_process(process)) stack.enter_context(terminate_process(process))
else: else:
@@ -299,13 +307,13 @@ def run(
stdout_buf, stderr_buf = handle_io( stdout_buf, stderr_buf = handle_io(
process, process,
log, options.log,
prefix=prefix, prefix=options.prefix,
msg_color=msg_color, msg_color=options.msg_color,
timeout=timeout, timeout=options.timeout,
input_bytes=input, input_bytes=options.input,
stdout=stdout, stdout=options.stdout,
stderr=stderr, stderr=options.stderr,
) )
process.wait() process.wait()
@@ -317,20 +325,20 @@ def run(
cmd_out = CmdOut( cmd_out = CmdOut(
stdout=stdout_buf, stdout=stdout_buf,
stderr=stderr_buf, stderr=stderr_buf,
cwd=cwd, cwd=options.cwd,
env=env, env=options.env,
command_list=cmd, command_list=cmd,
returncode=process.returncode, returncode=process.returncode,
msg=error_msg, msg=options.error_msg,
) )
if check and process.returncode != 0: if options.check and process.returncode != 0:
raise ClanCmdError(cmd_out) raise ClanCmdError(cmd_out)
return cmd_out return cmd_out
def run_no_stdout( def run_no_output(
cmd: list[str], cmd: list[str],
*, *,
env: dict[str, str] | None = None, env: dict[str, str] | None = None,
@@ -344,16 +352,17 @@ def run_no_stdout(
shell: bool = False, shell: bool = False,
) -> CmdOut: ) -> CmdOut:
""" """
Like run, but automatically suppresses stdout, if not in DEBUG log level. Like run, but automatically suppresses all output, if not in DEBUG log level.
If in DEBUG log level the stdout of commands will be shown. If in DEBUG log level the stdout of commands will be shown.
""" """
if cwd is None: if cwd is None:
cwd = Path.cwd() cwd = Path.cwd()
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
return run(cmd, env=env, log=log, check=check, error_msg=error_msg) return run(cmd, RunOpts(env=env, log=log, check=check, error_msg=error_msg))
log = Log.NONE log = Log.NONE
return run( return run(
cmd, cmd,
RunOpts(
env=env, env=env,
log=log, log=log,
check=check, check=check,
@@ -361,4 +370,5 @@ def run_no_stdout(
error_msg=error_msg, error_msg=error_msg,
needs_user_terminal=needs_user_terminal, needs_user_terminal=needs_user_terminal,
shell=shell, shell=shell,
),
) )

4
pkgs/clan-cli/clan_cli/facts/generate.py
View File

@@ -8,7 +8,7 @@ from collections.abc import Callable
from pathlib import Path from pathlib import Path
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
from clan_cli.cmd import run from clan_cli.cmd import RunOpts, run
from clan_cli.completions import ( from clan_cli.completions import (
add_dynamic_completer, add_dynamic_completer,
complete_machines, complete_machines,
@@ -103,7 +103,7 @@ def generate_service_facts(
cmd = ["bash", "-c", generator] cmd = ["bash", "-c", generator]
run( run(
cmd, cmd,
env=env, RunOpts(env=env),
) )
files_to_commit = [] files_to_commit = []
# store secrets # store secrets

6
pkgs/clan-cli/clan_cli/flash/automount.py
View File

@@ -4,7 +4,7 @@ from collections.abc import Generator
from contextlib import contextmanager from contextlib import contextmanager
from pathlib import Path from pathlib import Path
from clan_cli.cmd import Log, run from clan_cli.cmd import Log, RunOpts, run
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -30,11 +30,11 @@ def pause_automounting(devices: list[Path]) -> Generator[None, None, None]:
str_devs = [str(dev) for dev in devices] str_devs = [str(dev) for dev in devices]
cmd = ["sudo", str(inhibit_path), "enable", *str_devs] cmd = ["sudo", str(inhibit_path), "enable", *str_devs]
result = run(cmd, log=Log.BOTH, check=False, needs_user_terminal=True) result = run(cmd, RunOpts(log=Log.BOTH, check=False, needs_user_terminal=True))
if result.returncode != 0: if result.returncode != 0:
log.error("Failed to inhibit automounting") log.error("Failed to inhibit automounting")
yield None yield None
cmd = ["sudo", str(inhibit_path), "disable", *str_devs] cmd = ["sudo", str(inhibit_path), "disable", *str_devs]
result = run(cmd, log=Log.BOTH, check=False) result = run(cmd, RunOpts(log=Log.BOTH, check=False))
if result.returncode != 0: if result.returncode != 0:
log.error("Failed to re-enable automounting") log.error("Failed to re-enable automounting")

4
pkgs/clan-cli/clan_cli/flash/flash.py
View File

@@ -9,7 +9,7 @@ from tempfile import TemporaryDirectory
from typing import Any from typing import Any
from clan_cli.api import API from clan_cli.api import API
from clan_cli.cmd import Log, run from clan_cli.cmd import Log, RunOpts, run
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
from clan_cli.facts.generate import generate_facts from clan_cli.facts.generate import generate_facts
from clan_cli.facts.secret_modules import SecretStoreBase from clan_cli.facts.secret_modules import SecretStoreBase
@@ -156,7 +156,9 @@ def flash_machine(
) )
run( run(
cmd, cmd,
RunOpts(
log=Log.BOTH, log=Log.BOTH,
error_msg=f"Failed to flash {machine}", error_msg=f"Failed to flash {machine}",
needs_user_terminal=True, needs_user_terminal=True,
),
) )

6
pkgs/clan-cli/clan_cli/flash/list.py
View File

@@ -4,7 +4,7 @@ import os
from pathlib import Path from pathlib import Path
from clan_cli.api import API from clan_cli.api import API
from clan_cli.cmd import Log, run from clan_cli.cmd import Log, RunOpts, run
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
from clan_cli.nix import nix_build from clan_cli.nix import nix_build
@@ -14,7 +14,7 @@ log = logging.getLogger(__name__)
@API.register @API.register
def list_possible_languages() -> list[str]: def list_possible_languages() -> list[str]:
cmd = nix_build(["nixpkgs#glibcLocales"]) cmd = nix_build(["nixpkgs#glibcLocales"])
result = run(cmd, log=Log.STDERR, error_msg="Failed to find glibc locales") result = run(cmd, RunOpts(log=Log.STDERR, error_msg="Failed to find glibc locales"))
locale_file = Path(result.stdout.strip()) / "share" / "i18n" / "SUPPORTED" locale_file = Path(result.stdout.strip()) / "share" / "i18n" / "SUPPORTED"
if not locale_file.exists(): if not locale_file.exists():
@@ -40,7 +40,7 @@ def list_possible_languages() -> list[str]:
@API.register @API.register
def list_possible_keymaps() -> list[str]: def list_possible_keymaps() -> list[str]:
cmd = nix_build(["nixpkgs#kbd"]) cmd = nix_build(["nixpkgs#kbd"])
result = run(cmd, log=Log.STDERR, error_msg="Failed to find kbdinfo") result = run(cmd, RunOpts(log=Log.STDERR, error_msg="Failed to find kbdinfo"))
keymaps_dir = Path(result.stdout.strip()) / "share" / "keymaps" keymaps_dir = Path(result.stdout.strip()) / "share" / "keymaps"
if not keymaps_dir.exists(): if not keymaps_dir.exists():

11
pkgs/clan-cli/clan_cli/git.py
View File

@@ -1,6 +1,6 @@
from pathlib import Path from pathlib import Path
from .cmd import Log, run from .cmd import Log, RunOpts, run
from .errors import ClanError from .errors import ClanError
from .locked_open import locked_open from .locked_open import locked_open
from .nix import run_cmd from .nix import run_cmd
@@ -67,8 +67,10 @@ def _commit_file_to_git(
run( run(
cmd, cmd,
RunOpts(
log=Log.BOTH, log=Log.BOTH,
error_msg=f"Failed to add {file_path} file to git index", error_msg=f"Failed to add {file_path} file to git index",
),
) )
# check if there is a diff # check if there is a diff
@@ -77,7 +79,7 @@ def _commit_file_to_git(
["git", "-C", str(repo_dir), "diff", "--cached", "--exit-code", "--"] ["git", "-C", str(repo_dir), "diff", "--cached", "--exit-code", "--"]
+ [str(file_path) for file_path in file_paths], + [str(file_path) for file_path in file_paths],
) )
result = run(cmd, check=False, cwd=repo_dir) result = run(cmd, RunOpts(check=False, cwd=repo_dir))
# if there is no diff, return # if there is no diff, return
if result.returncode == 0: if result.returncode == 0:
return return
@@ -98,5 +100,8 @@ def _commit_file_to_git(
) )
run( run(
cmd, error_msg=f"Failed to commit {file_paths} to git repository {repo_dir}" cmd,
RunOpts(
error_msg=f"Failed to commit {file_paths} to git repository {repo_dir}"
),
) )

4
pkgs/clan-cli/clan_cli/inventory/__init__.py
View File

@@ -18,7 +18,7 @@ from pathlib import Path
from typing import Any from typing import Any
from clan_cli.api import API, dataclass_to_dict, from_dict from clan_cli.api import API, dataclass_to_dict, from_dict
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_output
from clan_cli.errors import ClanCmdError, ClanError from clan_cli.errors import ClanCmdError, ClanError
from clan_cli.git import commit_file from clan_cli.git import commit_file
from clan_cli.nix import nix_eval from clan_cli.nix import nix_eval
@@ -78,7 +78,7 @@ def load_inventory_eval(flake_dir: str | Path) -> Inventory:
] ]
) )
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
try: try:
res = proc.stdout.strip() res = proc.stdout.strip()

6
pkgs/clan-cli/clan_cli/machines/create.py
View File

@@ -9,7 +9,7 @@ from tempfile import TemporaryDirectory
from clan_cli.api import API from clan_cli.api import API
from clan_cli.clan.create import git_command from clan_cli.clan.create import git_command
from clan_cli.clan_uri import FlakeId from clan_cli.clan_uri import FlakeId
from clan_cli.cmd import Log, run from clan_cli.cmd import Log, RunOpts, run
from clan_cli.completions import add_dynamic_completer, complete_tags from clan_cli.completions import add_dynamic_completer, complete_tags
from clan_cli.dirs import TemplateType, clan_templates, get_clan_flake_toplevel_or_env from clan_cli.dirs import TemplateType, clan_templates, get_clan_flake_toplevel_or_env
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
@@ -103,7 +103,7 @@ def create_machine(opts: CreateOptions) -> None:
# Check if debug logging is enabled # Check if debug logging is enabled
is_debug_enabled = log.isEnabledFor(logging.DEBUG) is_debug_enabled = log.isEnabledFor(logging.DEBUG)
log_flag = Log.BOTH if is_debug_enabled else Log.NONE log_flag = Log.BOTH if is_debug_enabled else Log.NONE
run(command, log=log_flag, cwd=tmpdirp) run(command, RunOpts(log=log_flag, cwd=tmpdirp))
validate_directory(tmpdirp) validate_directory(tmpdirp)
@@ -126,7 +126,7 @@ def create_machine(opts: CreateOptions) -> None:
shutil.copytree(src, dst, ignore_dangling_symlinks=True, copy_function=log_copy) shutil.copytree(src, dst, ignore_dangling_symlinks=True, copy_function=log_copy)
run(git_command(clan_dir, "add", f"machines/{machine_name}"), cwd=clan_dir) run(git_command(clan_dir, "add", f"machines/{machine_name}"), RunOpts(cwd=clan_dir))
inventory = load_inventory_json(clan_dir) inventory = load_inventory_json(clan_dir)

8
pkgs/clan-cli/clan_cli/machines/hardware.py
View File

@@ -7,7 +7,7 @@ from pathlib import Path
from clan_cli.api import API from clan_cli.api import API
from clan_cli.clan_uri import FlakeId from clan_cli.clan_uri import FlakeId
from clan_cli.cmd import run, run_no_stdout from clan_cli.cmd import RunOpts, run, run_no_output
from clan_cli.completions import add_dynamic_completer, complete_machines from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.dirs import specific_machine_dir from clan_cli.dirs import specific_machine_dir
from clan_cli.errors import ClanCmdError, ClanError from clan_cli.errors import ClanCmdError, ClanError
@@ -71,7 +71,7 @@ def show_machine_deployment_target(clan_dir: Path, machine_name: str) -> str | N
"--json", "--json",
] ]
) )
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()
target_host = json.loads(res) target_host = json.loads(res)
@@ -93,7 +93,7 @@ def show_machine_hardware_platform(clan_dir: Path, machine_name: str) -> str | N
"--json", "--json",
] ]
) )
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()
host_platform = json.loads(res) host_platform = json.loads(res)
@@ -160,7 +160,7 @@ def generate_machine_hardware_info(opts: HardwareGenerateOptions) -> HardwareCon
*config_command, *config_command,
], ],
) )
out = run(cmd, needs_user_terminal=True) out = run(cmd, RunOpts(needs_user_terminal=True))
if out.returncode != 0: if out.returncode != 0:
log.error(out) log.error(out)
msg = f"Failed to inspect {opts.machine}. Address: {opts.target_host}" msg = f"Failed to inspect {opts.machine}. Address: {opts.target_host}"

4
pkgs/clan-cli/clan_cli/machines/install.py
View File

@@ -9,7 +9,7 @@ from tempfile import TemporaryDirectory
from clan_cli.api import API from clan_cli.api import API
from clan_cli.clan_uri import FlakeId from clan_cli.clan_uri import FlakeId
from clan_cli.cmd import Log, run from clan_cli.cmd import Log, RunOpts, run
from clan_cli.completions import ( from clan_cli.completions import (
add_dynamic_completer, add_dynamic_completer,
complete_machines, complete_machines,
@@ -119,7 +119,7 @@ def install_machine(opts: InstallOptions) -> None:
["nixpkgs#nixos-anywhere"], ["nixpkgs#nixos-anywhere"],
cmd, cmd,
), ),
log=Log.BOTH, RunOpts(log=Log.BOTH),
) )

6
pkgs/clan-cli/clan_cli/machines/list.py
View File

@@ -6,7 +6,7 @@ from pathlib import Path
from typing import Literal from typing import Literal
from clan_cli.api import API from clan_cli.api import API
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_output
from clan_cli.completions import add_dynamic_completer, complete_tags from clan_cli.completions import add_dynamic_completer, complete_tags
from clan_cli.errors import ClanCmdError, ClanError from clan_cli.errors import ClanCmdError, ClanError
from clan_cli.inventory import Machine, load_inventory_eval, set_inventory from clan_cli.inventory import Machine, load_inventory_eval, set_inventory
@@ -69,7 +69,7 @@ def list_nixos_machines(flake_url: str | Path) -> list[str]:
"--json", "--json",
] ]
) )
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
try: try:
res = proc.stdout.strip() res = proc.stdout.strip()
@@ -123,7 +123,7 @@ def check_machine_online(
], ],
) )
try: try:
proc = run_no_stdout(cmd, needs_user_terminal=True) proc = run_no_output(cmd, needs_user_terminal=True)
if proc.returncode != 0: if proc.returncode != 0:
return "Offline" return "Offline"
except ClanCmdError: except ClanCmdError:

10
pkgs/clan-cli/clan_cli/machines/machines.py
View File

@@ -9,7 +9,7 @@ from tempfile import NamedTemporaryFile
from typing import TYPE_CHECKING, Any, Literal from typing import TYPE_CHECKING, Any, Literal
from clan_cli.clan_uri import FlakeId from clan_cli.clan_uri import FlakeId
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_output
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
from clan_cli.facts import public_modules as facts_public_modules from clan_cli.facts import public_modules as facts_public_modules
from clan_cli.facts import secret_modules as facts_secret_modules from clan_cli.facts import secret_modules as facts_secret_modules
@@ -70,7 +70,7 @@ class Machine:
attr = f'(builtins.getFlake "{self.flake}").nixosConfigurations.{self.name}.pkgs.hostPlatform.system' attr = f'(builtins.getFlake "{self.flake}").nixosConfigurations.{self.name}.pkgs.hostPlatform.system'
output = self._eval_cache.get(attr) output = self._eval_cache.get(attr)
if output is None: if output is None:
output = run_no_stdout( output = run_no_output(
nix_eval(["--impure", "--expr", attr]) nix_eval(["--impure", "--expr", attr])
).stdout.strip() ).stdout.strip()
self._eval_cache[attr] = output self._eval_cache[attr] = output
@@ -243,7 +243,7 @@ class Machine:
config_json.flush() config_json.flush()
file_info = json.loads( file_info = json.loads(
run_no_stdout( run_no_output(
nix_eval( nix_eval(
[ [
"--impure", "--impure",
@@ -288,9 +288,9 @@ class Machine:
args += nix_options + self.nix_options args += nix_options + self.nix_options
if method == "eval": if method == "eval":
output = run_no_stdout(nix_eval(args)).stdout.strip() output = run_no_output(nix_eval(args)).stdout.strip()
return output return output
return Path(run_no_stdout(nix_build(args)).stdout.strip()) return Path(run_no_output(nix_build(args)).stdout.strip())
def eval_nix( def eval_nix(
self, self,

9
pkgs/clan-cli/clan_cli/machines/update.py
View File

@@ -7,7 +7,7 @@ import sys
from clan_cli.api import API from clan_cli.api import API
from clan_cli.clan_uri import FlakeId from clan_cli.clan_uri import FlakeId
from clan_cli.cmd import MsgColor, run from clan_cli.cmd import MsgColor, RunOpts, run
from clan_cli.colors import AnsiColor from clan_cli.colors import AnsiColor
from clan_cli.completions import ( from clan_cli.completions import (
add_dynamic_completer, add_dynamic_completer,
@@ -64,7 +64,10 @@ def upload_sources(machine: Machine) -> str:
path, path,
] ]
) )
run(cmd, env=env, error_msg="failed to upload sources", prefix=machine.name) run(
cmd,
RunOpts(env=env, error_msg="failed to upload sources", prefix=machine.name),
)
return path return path
# Slow path: we need to upload all sources to the remote machine # Slow path: we need to upload all sources to the remote machine
@@ -78,7 +81,7 @@ def upload_sources(machine: Machine) -> str:
flake_url, flake_url,
] ]
) )
proc = run(cmd, env=env, error_msg="failed to upload sources") proc = run(cmd, RunOpts(env=env, error_msg="failed to upload sources"))
try: try:
return json.loads(proc.stdout)["path"] return json.loads(proc.stdout)["path"]

4
pkgs/clan-cli/clan_cli/nix/__init__.py
View File

@@ -4,7 +4,7 @@ import tempfile
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from clan_cli.cmd import run, run_no_stdout from clan_cli.cmd import run, run_no_output
from clan_cli.dirs import nixpkgs_flake, nixpkgs_source from clan_cli.dirs import nixpkgs_flake, nixpkgs_source
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
@@ -63,7 +63,7 @@ def nix_add_to_gcroots(nix_path: Path, dest: Path) -> None:
def nix_config() -> dict[str, Any]: def nix_config() -> dict[str, Any]:
cmd = nix_command(["show-config", "--json"]) cmd = nix_command(["show-config", "--json"])
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
data = json.loads(proc.stdout) data = json.loads(proc.stdout)
config = {} config = {}
for key, value in data.items(): for key, value in data.items():

4
pkgs/clan-cli/clan_cli/secrets/import_sops.py
View File

@@ -3,7 +3,7 @@ import json
import sys import sys
from pathlib import Path from pathlib import Path
from clan_cli.cmd import run from clan_cli.cmd import RunOpts, run
from clan_cli.completions import ( from clan_cli.completions import (
add_dynamic_completer, add_dynamic_completer,
complete_groups, complete_groups,
@@ -32,7 +32,7 @@ def import_sops(args: argparse.Namespace) -> None:
cmd += ["--output-type", "json", "--decrypt", args.sops_file] cmd += ["--output-type", "json", "--decrypt", args.sops_file]
cmd = nix_shell(["nixpkgs#sops"], cmd) cmd = nix_shell(["nixpkgs#sops"], cmd)
res = run(cmd, error_msg=f"Could not import sops file {file}") res = run(cmd, RunOpts(error_msg=f"Could not import sops file {file}"))
secrets = json.loads(res.stdout) secrets = json.loads(res.stdout)
for k, v in secrets.items(): for k, v in secrets.items():
k = args.prefix + k k = args.prefix + k

80
pkgs/clan-cli/clan_cli/secrets/sops.py
View File

@@ -1,22 +1,22 @@
import dataclasses import dataclasses
import enum import enum
import functools
import io import io
import json import json
import logging import logging
import os import os
import shutil import shutil
import subprocess import subprocess
import sys
from collections.abc import Iterable, Sequence from collections.abc import Iterable, Sequence
from contextlib import suppress from contextlib import suppress
from pathlib import Path from pathlib import Path
from tempfile import NamedTemporaryFile from tempfile import NamedTemporaryFile
from typing import IO, Protocol from typing import IO
import clan_cli.cmd
from clan_cli.api import API from clan_cli.api import API
from clan_cli.cmd import Log, RunOpts, run
from clan_cli.dirs import user_config_dir from clan_cli.dirs import user_config_dir
from clan_cli.errors import ClanError, CmdOut from clan_cli.errors import ClanError
from clan_cli.nix import nix_shell from clan_cli.nix import nix_shell
from .folders import sops_machines_folder, sops_users_folder from .folders import sops_machines_folder, sops_users_folder
@@ -163,10 +163,10 @@ class ExitStatus(enum.IntEnum): # see: cmd/sops/codes/codes.go
return ExitStatus(code) if code in ExitStatus else None return ExitStatus(code) if code in ExitStatus else None
class Executer(Protocol): # class Executer(Protocol):
def __call__( # def __call__(
self, cmd: list[str], *, env: dict[str, str] | None = None # self, cmd: list[str], *, env: dict[str, str] | None = None
) -> CmdOut: ... # ) -> CmdOut: ...
class Operation(enum.StrEnum): class Operation(enum.StrEnum):
@@ -176,11 +176,11 @@ class Operation(enum.StrEnum):
UPDATE_KEYS = "updatekeys" UPDATE_KEYS = "updatekeys"
def run( def sops_run(
call: Operation, call: Operation,
secret_path: Path, secret_path: Path,
public_keys: Iterable[tuple[str, KeyType]], public_keys: Iterable[tuple[str, KeyType]],
executer: Executer, run_opts: RunOpts | None = None,
) -> tuple[int, str]: ) -> tuple[int, str]:
"""Call the sops binary for the given operation.""" """Call the sops binary for the given operation."""
# louis(2024-11-19): I regrouped the call into the sops binary into this # louis(2024-11-19): I regrouped the call into the sops binary into this
@@ -235,7 +235,12 @@ def run(
sops_cmd.append(str(secret_path)) sops_cmd.append(str(secret_path))
cmd = nix_shell(["nixpkgs#sops"], sops_cmd) cmd = nix_shell(["nixpkgs#sops"], sops_cmd)
p = executer(cmd, env=environ) opts = (
dataclasses.replace(run_opts, env=environ)
if run_opts
else RunOpts(env=environ)
)
p = run(cmd, opts)
return p.returncode, p.stdout return p.returncode, p.stdout
@@ -254,7 +259,7 @@ def get_public_age_key(privkey: str) -> str:
def generate_private_key(out_file: Path | None = None) -> tuple[str, str]: def generate_private_key(out_file: Path | None = None) -> tuple[str, str]:
cmd = nix_shell(["nixpkgs#age"], ["age-keygen"]) cmd = nix_shell(["nixpkgs#age"], ["age-keygen"])
try: try:
proc = clan_cli.cmd.run(cmd) proc = run(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()
pubkey = None pubkey = None
private_key = None private_key = None
@@ -355,10 +360,13 @@ def ensure_admin_public_key(flake_dir: Path) -> SopsKey:
def update_keys(secret_path: Path, keys: Iterable[tuple[str, KeyType]]) -> list[Path]: def update_keys(secret_path: Path, keys: Iterable[tuple[str, KeyType]]) -> list[Path]:
secret_path = secret_path / "secret" secret_path = secret_path / "secret"
error_msg = f"Could not update keys for {secret_path}" error_msg = f"Could not update keys for {secret_path}"
executer = functools.partial(
clan_cli.cmd.run, log=clan_cli.cmd.Log.BOTH, error_msg=error_msg rc, _ = sops_run(
Operation.UPDATE_KEYS,
secret_path,
keys,
RunOpts(log=Log.BOTH, error_msg=error_msg),
) )
rc, _ = run(Operation.UPDATE_KEYS, secret_path, keys, executer)
was_modified = ExitStatus.parse(rc) != ExitStatus.FILE_HAS_NOT_BEEN_MODIFIED was_modified = ExitStatus.parse(rc) != ExitStatus.FILE_HAS_NOT_BEEN_MODIFIED
return [secret_path] if was_modified else [] return [secret_path] if was_modified else []
@@ -372,20 +380,19 @@ def encrypt_file(
folder.mkdir(parents=True, exist_ok=True) folder.mkdir(parents=True, exist_ok=True)
if not content: if not content:
# Don't use our `run` here, because it breaks editor integration. # Use direct stdout / stderr, as else it breaks editor integration.
# We never need this in our UI. # We never need this in our UI. TUI only.
def executer(cmd: list[str], *, env: dict[str, str] | None = None) -> CmdOut: rc, _ = sops_run(
return CmdOut( Operation.EDIT,
stdout="", secret_path,
stderr="", pubkeys,
cwd=Path.cwd(), RunOpts(
env=env, stdout=sys.stdout.buffer,
command_list=cmd, stderr=sys.stderr.buffer,
returncode=subprocess.run(cmd, env=env, check=False).returncode, check=False,
msg=None, log=Log.NONE,
),
) )
rc, _ = run(Operation.EDIT, secret_path, pubkeys, executer)
status = ExitStatus.parse(rc) status = ExitStatus.parse(rc)
if rc == 0 or status == ExitStatus.FILE_HAS_NOT_BEEN_MODIFIED: if rc == 0 or status == ExitStatus.FILE_HAS_NOT_BEEN_MODIFIED:
return return
@@ -418,8 +425,12 @@ def encrypt_file(
else: else:
msg = f"Invalid content type: {type(content)}" msg = f"Invalid content type: {type(content)}"
raise ClanError(msg) raise ClanError(msg)
executer = functools.partial(clan_cli.cmd.run, log=clan_cli.cmd.Log.BOTH) sops_run(
run(Operation.ENCRYPT, Path(source.name), pubkeys, executer) Operation.ENCRYPT,
Path(source.name),
pubkeys,
RunOpts(log=Log.BOTH),
)
# atomic copy of the encrypted file # atomic copy of the encrypted file
with NamedTemporaryFile(dir=folder, delete=False) as dest: with NamedTemporaryFile(dir=folder, delete=False) as dest:
shutil.copyfile(source.name, dest.name) shutil.copyfile(source.name, dest.name)
@@ -432,10 +443,13 @@ def encrypt_file(
def decrypt_file(secret_path: Path) -> str: def decrypt_file(secret_path: Path) -> str:
# decryption uses private keys from the environment or default paths: # decryption uses private keys from the environment or default paths:
no_public_keys_needed: list[tuple[str, KeyType]] = [] no_public_keys_needed: list[tuple[str, KeyType]] = []
executer = functools.partial(
clan_cli.cmd.run, error_msg=f"Could not decrypt {secret_path}" _, stdout = sops_run(
Operation.DECRYPT,
secret_path,
no_public_keys_needed,
RunOpts(error_msg=f"Could not decrypt {secret_path}"),
) )
_, stdout = run(Operation.DECRYPT, secret_path, no_public_keys_needed, executer)
return stdout return stdout

4
pkgs/clan-cli/clan_cli/ssh/host.py
View File

@@ -9,7 +9,7 @@ from pathlib import Path
from shlex import quote from shlex import quote
from typing import IO, Any from typing import IO, Any
from clan_cli.cmd import CmdOut, Log, MsgColor from clan_cli.cmd import CmdOut, Log, MsgColor, RunOpts
from clan_cli.cmd import run as local_run from clan_cli.cmd import run as local_run
from clan_cli.colors import AnsiColor from clan_cli.colors import AnsiColor
from clan_cli.ssh.host_key import HostKeyCheck from clan_cli.ssh.host_key import HostKeyCheck
@@ -68,6 +68,7 @@ class Host:
) -> CmdOut: ) -> CmdOut:
res = local_run( res = local_run(
cmd, cmd,
RunOpts(
shell=shell, shell=shell,
stdout=stdout, stdout=stdout,
prefix=self.command_prefix, prefix=self.command_prefix,
@@ -81,6 +82,7 @@ class Host:
error_msg=error_msg, error_msg=error_msg,
msg_color=msg_color, msg_color=msg_color,
needs_user_terminal=needs_user_terminal, needs_user_terminal=needs_user_terminal,
),
) )
return res return res

4
pkgs/clan-cli/clan_cli/ssh/upload.py
View File

@@ -2,7 +2,7 @@ import tarfile
from pathlib import Path from pathlib import Path
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
from clan_cli.cmd import Log from clan_cli.cmd import Log, RunOpts
from clan_cli.cmd import run as run_local from clan_cli.cmd import run as run_local
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
from clan_cli.ssh.host import Host from clan_cli.ssh.host import Host
@@ -77,8 +77,10 @@ def upload(
with tar_path.open("rb") as f: with tar_path.open("rb") as f:
run_local( run_local(
cmd, cmd,
RunOpts(
input=f.read(), input=f.read(),
log=Log.BOTH, log=Log.BOTH,
prefix=host.command_prefix, prefix=host.command_prefix,
needs_user_terminal=True, needs_user_terminal=True,
),
) )

4
pkgs/clan-cli/clan_cli/state/list.py
View File

@@ -3,7 +3,7 @@ import json
import logging import logging
from pathlib import Path from pathlib import Path
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_output
from clan_cli.completions import ( from clan_cli.completions import (
add_dynamic_completer, add_dynamic_completer,
complete_machines, complete_machines,
@@ -31,7 +31,7 @@ def list_state_folders(machine: str, service: None | str = None) -> None:
res = "{}" res = "{}"
try: try:
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()
except ClanCmdError as e: except ClanCmdError as e:
msg = "Clan might not have meta attributes" msg = "Clan might not have meta attributes"

4
pkgs/clan-cli/clan_cli/tags.py
View File

@@ -2,7 +2,7 @@ import json
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_output
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
from clan_cli.nix import nix_eval from clan_cli.nix import nix_eval
@@ -18,7 +18,7 @@ def list_tagged_machines(flake_url: str | Path) -> dict[str, Any]:
"--json", "--json",
] ]
) )
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
try: try:
res = proc.stdout.strip() res = proc.stdout.strip()

7
pkgs/clan-cli/clan_cli/vars/generate.py
View File

@@ -8,7 +8,7 @@ from pathlib import Path
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
from clan_cli.cmd import run from clan_cli.cmd import RunOpts, run
from clan_cli.completions import ( from clan_cli.completions import (
add_dynamic_completer, add_dynamic_completer,
complete_machines, complete_machines,
@@ -192,10 +192,7 @@ def execute_generator(
cmd = bubblewrap_cmd(generator.final_script, tmpdir) cmd = bubblewrap_cmd(generator.final_script, tmpdir)
else: else:
cmd = ["bash", "-c", generator.final_script] cmd = ["bash", "-c", generator.final_script]
run( run(cmd, RunOpts(env=env))
cmd,
env=env,
)
files_to_commit = [] files_to_commit = []
# store secrets # store secrets
files = generator.files files = generator.files

9
pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py
View File

@@ -6,7 +6,7 @@ from itertools import chain
from pathlib import Path from pathlib import Path
from typing import override from typing import override
from clan_cli.cmd import Log, run from clan_cli.cmd import Log, RunOpts, run
from clan_cli.machines.machines import Machine from clan_cli.machines.machines import Machine
from clan_cli.nix import nix_shell from clan_cli.nix import nix_shell
from clan_cli.vars.generate import Generator, Var from clan_cli.vars.generate import Generator, Var
@@ -50,8 +50,7 @@ class SecretStore(SecretStoreBase):
str(self.entry_dir(generator, var.name)), str(self.entry_dir(generator, var.name)),
], ],
), ),
input=value, RunOpts(input=value, check=True),
check=True,
) )
return None # we manage the files outside of the git repo return None # we manage the files outside of the git repo
@@ -88,7 +87,7 @@ class SecretStore(SecretStoreBase):
self.entry_prefix, self.entry_prefix,
], ],
), ),
check=False, RunOpts(check=False),
) )
.stdout.strip() .stdout.strip()
.encode() .encode()
@@ -116,7 +115,7 @@ class SecretStore(SecretStoreBase):
str(symlink), str(symlink),
], ],
), ),
check=False, RunOpts(check=False),
) )
.stdout.strip() .stdout.strip()
.encode() .encode()

5
pkgs/clan-cli/clan_cli/vms/run.py
View File

@@ -13,7 +13,7 @@ from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
from clan_cli.cmd import CmdOut, Log, handle_io, run from clan_cli.cmd import CmdOut, Log, RunOpts, handle_io, run
from clan_cli.completions import add_dynamic_completer, complete_machines from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.dirs import module_root, user_cache_dir, vm_state_dir from clan_cli.dirs import module_root, user_cache_dir, vm_state_dir
from clan_cli.errors import ClanCmdError, ClanError from clan_cli.errors import ClanCmdError, ClanError
@@ -109,8 +109,7 @@ def prepare_disk(
) )
run( run(
cmd, cmd,
log=Log.BOTH, RunOpts(log=Log.BOTH, error_msg=f"Could not create disk image at {disk_img}"),
error_msg=f"Could not create disk image at {disk_img}",
) )
return disk_img return disk_img

1
pkgs/clan-cli/tests/test_api_dataclass_compat.py
View File

@@ -119,6 +119,7 @@ def test_all_dataclasses() -> None:
# - API includes Type Generic wrappers, that are not known in the init file. # - API includes Type Generic wrappers, that are not known in the init file.
excludes = [ excludes = [
"api/__init__.py", "api/__init__.py",
"cmd.py", # We don't want the UI to have access to the cmd module anyway
] ]
cli_path = Path("clan_cli").resolve() cli_path = Path("clan_cli").resolve()

4
pkgs/clan-cli/tests/test_modules.py
View File

@@ -12,7 +12,7 @@ from clan_cli.inventory import (
set_inventory, set_inventory,
) )
from clan_cli.machines.create import CreateOptions, create_machine from clan_cli.machines.create import CreateOptions, create_machine
from clan_cli.nix import nix_eval, run_no_stdout from clan_cli.nix import nix_eval, run_no_output
from fixtures_flakes import FlakeForTest from fixtures_flakes import FlakeForTest
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -97,7 +97,7 @@ def test_add_module_to_inventory(
"--json", "--json",
] ]
) )
proc = run_no_stdout(cmd) proc = run_no_output(cmd)
res = json.loads(proc.stdout.strip()) res = json.loads(proc.stdout.strip())
assert res["machine1"]["authorizedKeys"] == [ssh_key] assert res["machine1"]["authorizedKeys"] == [ssh_key]

2
pkgs/clan-cli/tests/test_vars_deployment.py
View File

@@ -94,7 +94,7 @@ def test_vm_deployment(
).stdout.strip() ).stdout.strip()
assert "no-such-path" not in shared_secret_path assert "no-such-path" not in shared_secret_path
# run nix flake lock # run nix flake lock
cmd.run(["nix", "flake", "lock"], cwd=flake.path) cmd.run(["nix", "flake", "lock"], cmd.RunOpts(cwd=flake.path))
vm1_config = inspect_vm(machine=Machine("m1_machine", FlakeId(str(flake.path)))) vm1_config = inspect_vm(machine=Machine("m1_machine", FlakeId(str(flake.path))))
vm2_config = inspect_vm(machine=Machine("m2_machine", FlakeId(str(flake.path)))) vm2_config = inspect_vm(machine=Machine("m2_machine", FlakeId(str(flake.path))))