clan-cli: deprecate nix_shell() in favor of run_cmd()

This commit is contained in:
DavHau
2024-07-16 13:48:55 +07:00
parent c9de01b9d0
commit 37e6ca7a30
24 changed files with 104 additions and 116 deletions

View File

@@ -3,7 +3,7 @@ from pathlib import Path
import pytest import pytest
from clan_cli.custom_logger import setup_logging from clan_cli.custom_logger import setup_logging
from clan_cli.nix import nix_shell from clan_cli.nix import run_cmd
pytest_plugins = [ pytest_plugins = [
"temporary_dir", "temporary_dir",
@@ -27,12 +27,12 @@ def pytest_sessionstart(session: pytest.Session) -> None:
@pytest.fixture @pytest.fixture
def git_repo(tmp_path: Path) -> Path: def git_repo(tmp_path: Path) -> Path:
# initialize a git repository # initialize a git repository
cmd = nix_shell(["nixpkgs#git"], ["git", "init"]) cmd = run_cmd(["git"], ["git", "init"])
subprocess.run(cmd, cwd=tmp_path, check=True) subprocess.run(cmd, cwd=tmp_path, check=True)
# set user.name and user.email # set user.name and user.email
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.name", "test"]) cmd = run_cmd(["git"], ["git", "config", "user.name", "test"])
subprocess.run(cmd, cwd=tmp_path, check=True) subprocess.run(cmd, cwd=tmp_path, check=True)
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.email", "test@test.test"]) cmd = run_cmd(["git"], ["git", "config", "user.email", "test@test.test"])
subprocess.run(cmd, cwd=tmp_path, check=True) subprocess.run(cmd, cwd=tmp_path, check=True)
# return the path to the git repository # return the path to the git repository
return tmp_path return tmp_path

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 run_cmd, run_no_stdout
from . import API from . import API
@@ -117,7 +117,7 @@ def show_block_devices() -> Blockdevices:
Abstract api method to show block devices. Abstract api method to show block devices.
It must return a list of block devices. It must return a list of block devices.
""" """
cmd = nix_shell(["nixpkgs#util-linux"], ["lsblk", "--json"]) cmd = run_cmd(["util-linux"], ["lsblk", "--json"])
proc = run_no_stdout(cmd) proc = run_no_stdout(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()

View File

@@ -3,7 +3,7 @@ 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_stdout
from clan_cli.nix import nix_shell from clan_cli.nix import run_cmd
from . import API from . import API
@@ -89,8 +89,8 @@ def parse_avahi_output(output: str) -> DNSInfo:
@API.register @API.register
def show_mdns() -> DNSInfo: def show_mdns() -> DNSInfo:
cmd = nix_shell( cmd = run_cmd(
["nixpkgs#avahi"], ["avahi"],
[ [
"avahi-browse", "avahi-browse",
"--all", "--all",

View File

@@ -10,7 +10,7 @@ from clan_cli.inventory import Inventory, InventoryMeta
from ..cmd import CmdOut, run from ..cmd import CmdOut, run
from ..errors import ClanError from ..errors import ClanError
from ..nix import nix_command, nix_shell from ..nix import nix_command, run_cmd
default_template_url: str = "git+https://git.clan.lol/clan/clan-core" default_template_url: str = "git+https://git.clan.lol/clan/clan-core"
minimal_template_url: str = "git+https://git.clan.lol/clan/clan-core#templates.minimal" minimal_template_url: str = "git+https://git.clan.lol/clan/clan-core#templates.minimal"
@@ -64,21 +64,19 @@ def create_clan(options: CreateOptions) -> CreateClanResponse:
out = run(command, cwd=directory) out = run(command, cwd=directory)
## Begin: setup git ## Begin: setup git
command = nix_shell(["nixpkgs#git"], ["git", "init"]) command = run_cmd(["git"], ["git", "init"])
out = run(command, cwd=directory) out = run(command, cwd=directory)
cmd_responses["git init"] = out cmd_responses["git init"] = out
command = nix_shell(["nixpkgs#git"], ["git", "add", "."]) command = run_cmd(["git"], ["git", "add", "."])
out = run(command, cwd=directory) out = run(command, cwd=directory)
cmd_responses["git add"] = out cmd_responses["git add"] = out
command = nix_shell(["nixpkgs#git"], ["git", "config", "user.name", "clan-tool"]) command = run_cmd(["git"], ["git", "config", "user.name", "clan-tool"])
out = run(command, cwd=directory) out = run(command, cwd=directory)
cmd_responses["git config"] = out cmd_responses["git config"] = out
command = nix_shell( command = run_cmd(["git"], ["git", "config", "user.email", "clan@example.com"])
["nixpkgs#git"], ["git", "config", "user.email", "clan@example.com"]
)
out = run(command, cwd=directory) out = run(command, cwd=directory)
cmd_responses["git config"] = out cmd_responses["git config"] = out
## End: setup git ## End: setup git

View File

@@ -19,7 +19,7 @@ from ..errors import ClanError
from ..git import commit_files from ..git import commit_files
from ..machines.inventory import get_all_machines, get_selected_machines from ..machines.inventory import get_all_machines, get_selected_machines
from ..machines.machines import Machine from ..machines.machines import Machine
from ..nix import nix_shell from ..nix import run_cmd
from .check import check_secrets from .check import check_secrets
from .public_modules import FactStoreBase from .public_modules import FactStoreBase
from .secret_modules import SecretStoreBase from .secret_modules import SecretStoreBase
@@ -39,10 +39,10 @@ def read_multiline_input(prompt: str = "Finish with Ctrl-D") -> str:
def bubblewrap_cmd(generator: str, facts_dir: Path, secrets_dir: Path) -> list[str]: def bubblewrap_cmd(generator: str, facts_dir: Path, secrets_dir: Path) -> list[str]:
# fmt: off # fmt: off
return nix_shell( return run_cmd(
[ [
"nixpkgs#bash", "bash",
"nixpkgs#bubblewrap", "bubblewrap",
], ],
[ [
"bwrap", "bwrap",

View File

@@ -3,7 +3,7 @@ import subprocess
from pathlib import Path from pathlib import Path
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 run_cmd
from . import SecretStoreBase from . import SecretStoreBase
@@ -16,8 +16,8 @@ class SecretStore(SecretStoreBase):
self, service: str, name: str, value: bytes, groups: list[str] self, service: str, name: str, value: bytes, groups: list[str]
) -> Path | None: ) -> Path | None:
subprocess.run( subprocess.run(
nix_shell( run_cmd(
["nixpkgs#pass"], ["pass"],
["pass", "insert", "-m", f"machines/{self.machine.name}/{name}"], ["pass", "insert", "-m", f"machines/{self.machine.name}/{name}"],
), ),
input=value, input=value,
@@ -27,8 +27,8 @@ class SecretStore(SecretStoreBase):
def get(self, service: str, name: str) -> bytes: def get(self, service: str, name: str) -> bytes:
return subprocess.run( return subprocess.run(
nix_shell( run_cmd(
["nixpkgs#pass"], ["pass"],
["pass", "show", f"machines/{self.machine.name}/{name}"], ["pass", "show", f"machines/{self.machine.name}/{name}"],
), ),
check=True, check=True,
@@ -49,8 +49,8 @@ class SecretStore(SecretStoreBase):
hashes = [] hashes = []
hashes.append( hashes.append(
subprocess.run( subprocess.run(
nix_shell( run_cmd(
["nixpkgs#git"], ["git"],
[ [
"git", "git",
"-C", "-C",
@@ -68,8 +68,8 @@ class SecretStore(SecretStoreBase):
if symlink.is_symlink(): if symlink.is_symlink():
hashes.append( hashes.append(
subprocess.run( subprocess.run(
nix_shell( run_cmd(
["nixpkgs#git"], ["git"],
[ [
"git", "git",
"-C", "-C",

View File

@@ -7,7 +7,7 @@ from tempfile import TemporaryDirectory
from ..cmd import Log, run from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines from ..completions import add_dynamic_completer, complete_machines
from ..machines.machines import Machine from ..machines.machines import Machine
from ..nix import nix_shell from ..nix import run_cmd
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -25,8 +25,8 @@ def upload_secrets(machine: Machine) -> None:
ssh_cmd = host.ssh_cmd() ssh_cmd = host.ssh_cmd()
run( run(
nix_shell( run_cmd(
["nixpkgs#rsync"], ["rsync"],
[ [
"rsync", "rsync",
"-e", "-e",

View File

@@ -19,7 +19,7 @@ from .completions import add_dynamic_completer, complete_machines
from .errors import ClanError from .errors import ClanError
from .facts.secret_modules import SecretStoreBase from .facts.secret_modules import SecretStoreBase
from .machines.machines import Machine from .machines.machines import Machine
from .nix import nix_shell from .nix import run_cmd
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -82,8 +82,8 @@ def flash_machine(
disko_install.extend(["--option", "dry-run", "true"]) disko_install.extend(["--option", "dry-run", "true"])
disko_install.extend(extra_args) disko_install.extend(extra_args)
cmd = nix_shell( cmd = run_cmd(
["nixpkgs#disko"], ["disko"],
disko_install, disko_install,
) )
run(cmd, log=Log.BOTH, error_msg=f"Failed to flash {machine}") run(cmd, log=Log.BOTH, error_msg=f"Failed to flash {machine}")

View File

@@ -9,7 +9,7 @@ from clan_cli.errors import ClanError
from ..cmd import run, run_no_stdout from ..cmd import run, run_no_stdout
from ..completions import add_dynamic_completer, complete_machines from ..completions import add_dynamic_completer, complete_machines
from ..nix import nix_config, nix_eval, nix_shell from ..nix import nix_config, nix_eval, run_cmd
from .types import machine_name_type from .types import machine_name_type
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -98,12 +98,12 @@ def generate_machine_hardware_info(
Generate hardware information for a machine Generate hardware information for a machine
and place the resulting *.nix file in the machine's directory. and place the resulting *.nix file in the machine's directory.
""" """
cmd = nix_shell( cmd = run_cmd(
[ [
"nixpkgs#openssh", "openssh",
"nixpkgs#sshpass", "sshpass",
# Provides nixos-generate-config on non-NixOS systems # Provides nixos-generate-config on non-NixOS systems
"nixpkgs#nixos-install-tools", "nixos-install-tools",
], ],
[ [
*(["sshpass", "-p", f"{password}"] if password else []), *(["sshpass", "-p", f"{password}"] if password else []),

View File

@@ -12,7 +12,7 @@ from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines from ..completions import add_dynamic_completer, complete_machines
from ..facts.generate import generate_facts from ..facts.generate import generate_facts
from ..machines.machines import Machine from ..machines.machines import Machine
from ..nix import nix_shell from ..nix import run_cmd
from ..ssh.cli import is_ipv6, is_reachable, qrcode_scan from ..ssh.cli import is_ipv6, is_reachable, qrcode_scan
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -81,8 +81,8 @@ def install_nixos(
cmd.append(target_host) cmd.append(target_host)
run( run(
nix_shell( run_cmd(
["nixpkgs#nixos-anywhere"], ["nixos-anywhere"],
cmd, cmd,
), ),
log=Log.BOTH, log=Log.BOTH,

View File

@@ -100,19 +100,6 @@ def nix_metadata(flake_url: str | Path) -> dict[str, Any]:
return data return data
def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
# we cannot use nix-shell inside the nix sandbox
# in our tests we just make sure we have all the packages
if os.environ.get("IN_NIX_SANDBOX") or os.environ.get("CLAN_NO_DYNAMIC_DEPS"):
return cmd
return [
*nix_command(["shell", "--inputs-from", f"{nixpkgs_flake()!s}"]),
*packages,
"-c",
*cmd,
]
# lazy loads list of allowed and static programs # lazy loads list of allowed and static programs
class Programs: class Programs:
allowed_programs = None allowed_programs = None
@@ -135,7 +122,7 @@ class Programs:
return program in cls.static_programs return program in cls.static_programs
# Alternative implementation of nix_shell() to replace nix_shell() at some point # Alternative implementation of run_cmd() to replace run_cmd() at some point
# Features: # Features:
# - allow list for programs (need to be specified in allowed-programs.json) # - allow list for programs (need to be specified in allowed-programs.json)
# - be abe to compute a closure of all deps for testing # - be abe to compute a closure of all deps for testing
@@ -143,7 +130,9 @@ class Programs:
def run_cmd(programs: list[str], cmd: list[str]) -> list[str]: def run_cmd(programs: list[str], cmd: list[str]) -> list[str]:
for program in programs: for program in programs:
if not Programs.is_allowed(program): if not Programs.is_allowed(program):
raise ValueError(f"Program not allowed: {program}") raise ValueError(
f"Program {program} is not allowed because it is not part of clan_cli/nix/allowed-programs.json."
)
if os.environ.get("IN_NIX_SANDBOX"): if os.environ.get("IN_NIX_SANDBOX"):
return cmd return cmd
missing_packages = [ missing_packages = [

View File

@@ -2,10 +2,13 @@
"age", "age",
"bash", "bash",
"bubblewrap", "bubblewrap",
"disko",
"e2fsprogs", "e2fsprogs",
"git", "git",
"gnupg",
"mypy", "mypy",
"nix", "nix",
"nixos-anywhere",
"openssh", "openssh",
"qemu", "qemu",
"rsync", "rsync",

View File

@@ -11,7 +11,7 @@ from ..completions import (
complete_users, complete_users,
) )
from ..errors import ClanError from ..errors import ClanError
from ..nix import nix_shell from ..nix import run_cmd
from .secrets import encrypt_secret, sops_secrets_folder from .secrets import encrypt_secret, sops_secrets_folder
@@ -28,7 +28,7 @@ def import_sops(args: argparse.Namespace) -> None:
if args.input_type: if args.input_type:
cmd += ["--input-type", args.input_type] cmd += ["--input-type", args.input_type]
cmd += ["--output-type", "json", "--decrypt", args.sops_file] cmd += ["--output-type", "json", "--decrypt", args.sops_file]
cmd = nix_shell(["nixpkgs#sops"], cmd) cmd = run_cmd(["sops"], cmd)
res = run(cmd, error_msg=f"Could not import sops file {file}") res = run(cmd, error_msg=f"Could not import sops file {file}")
secrets = json.loads(res.stdout) secrets = json.loads(res.stdout)

View File

@@ -12,7 +12,7 @@ from typing import IO
from ..cmd import Log, run from ..cmd import Log, run
from ..dirs import user_config_dir from ..dirs import user_config_dir
from ..errors import ClanError from ..errors import ClanError
from ..nix import nix_shell from ..nix import run_cmd
from .folders import sops_machines_folder, sops_users_folder from .folders import sops_machines_folder, sops_users_folder
@@ -23,7 +23,7 @@ class SopsKey:
def get_public_key(privkey: str) -> str: def get_public_key(privkey: str) -> str:
cmd = nix_shell(["nixpkgs#age"], ["age-keygen", "-y"]) cmd = run_cmd(["age"], ["age-keygen", "-y"])
try: try:
res = subprocess.run( res = subprocess.run(
cmd, input=privkey, stdout=subprocess.PIPE, text=True, check=True cmd, input=privkey, stdout=subprocess.PIPE, text=True, check=True
@@ -36,7 +36,7 @@ def get_public_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 = run_cmd(["age"], ["age-keygen"])
try: try:
proc = run(cmd) proc = run(cmd)
res = proc.stdout.strip() res = proc.stdout.strip()
@@ -125,8 +125,8 @@ def update_keys(secret_path: Path, keys: list[str]) -> list[Path]:
with sops_manifest(keys) as manifest: with sops_manifest(keys) as manifest:
secret_path = secret_path / "secret" secret_path = secret_path / "secret"
time_before = secret_path.stat().st_mtime time_before = secret_path.stat().st_mtime
cmd = nix_shell( cmd = run_cmd(
["nixpkgs#sops"], ["sops"],
[ [
"sops", "sops",
"--config", "--config",
@@ -152,7 +152,7 @@ def encrypt_file(
if not content: if not content:
args = ["sops", "--config", str(manifest)] args = ["sops", "--config", str(manifest)]
args.extend([str(secret_path)]) args.extend([str(secret_path)])
cmd = nix_shell(["nixpkgs#sops"], args) cmd = run_cmd(["sops"], args)
# Don't use our `run` here, because it breaks editor integration. # Don't use our `run` here, because it breaks editor integration.
# We never need this in our UI. # We never need this in our UI.
p = subprocess.run(cmd, check=False) p = subprocess.run(cmd, check=False)
@@ -180,7 +180,7 @@ def encrypt_file(
# we pass an empty manifest to pick up existing configuration of the user # we pass an empty manifest to pick up existing configuration of the user
args = ["sops", "--config", str(manifest)] args = ["sops", "--config", str(manifest)]
args.extend(["-i", "--encrypt", str(f.name)]) args.extend(["-i", "--encrypt", str(f.name)])
cmd = nix_shell(["nixpkgs#sops"], args) cmd = run_cmd(["sops"], args)
run(cmd, log=Log.BOTH) run(cmd, log=Log.BOTH)
# atomic copy of the encrypted file # atomic copy of the encrypted file
with NamedTemporaryFile(dir=folder, delete=False) as f2: with NamedTemporaryFile(dir=folder, delete=False) as f2:
@@ -195,8 +195,8 @@ def encrypt_file(
def decrypt_file(secret_path: Path) -> str: def decrypt_file(secret_path: Path) -> str:
with sops_manifest([]) as manifest: with sops_manifest([]) as manifest:
cmd = nix_shell( cmd = run_cmd(
["nixpkgs#sops"], ["sops"],
["sops", "--config", str(manifest), "--decrypt", str(secret_path)], ["sops", "--config", str(manifest), "--decrypt", str(secret_path)],
) )
res = run(cmd, error_msg=f"Could not decrypt {secret_path}") res = run(cmd, error_msg=f"Could not decrypt {secret_path}")

View File

@@ -6,7 +6,7 @@ import socket
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from ..nix import nix_shell from ..nix import run_cmd
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -18,13 +18,13 @@ def ssh(
ssh_args: list[str] = [], ssh_args: list[str] = [],
torify: bool = False, torify: bool = False,
) -> None: ) -> None:
packages = ["nixpkgs#openssh"] packages = ["openssh"]
if torify: if torify:
packages.append("nixpkgs#tor") packages.append("tor")
password_args = [] password_args = []
if password: if password:
packages.append("nixpkgs#sshpass") packages.append("sshpass")
password_args = [ password_args = [
"sshpass", "sshpass",
"-p", "-p",
@@ -48,15 +48,15 @@ def ssh(
if torify: if torify:
cmd_args.insert(0, "torify") cmd_args.insert(0, "torify")
cmd = nix_shell(packages, cmd_args) cmd = run_cmd(packages, cmd_args)
subprocess.run(cmd) subprocess.run(cmd)
def qrcode_scan(picture_file: str) -> str: def qrcode_scan(picture_file: str) -> str:
return ( return (
subprocess.run( subprocess.run(
nix_shell( run_cmd(
["nixpkgs#zbar"], ["zbar"],
[ [
"zbarimg", "zbarimg",
"--quiet", "--quiet",

View File

@@ -19,7 +19,7 @@ from ..errors import ClanError
from ..git import commit_files from ..git import commit_files
from ..machines.inventory import get_all_machines, get_selected_machines from ..machines.inventory import get_all_machines, get_selected_machines
from ..machines.machines import Machine from ..machines.machines import Machine
from ..nix import nix_shell from ..nix import run_cmd
from .check import check_secrets from .check import check_secrets
from .public_modules import FactStoreBase from .public_modules import FactStoreBase
from .secret_modules import SecretStoreBase from .secret_modules import SecretStoreBase
@@ -39,10 +39,10 @@ def read_multiline_input(prompt: str = "Finish with Ctrl-D") -> str:
def bubblewrap_cmd(generator: str, generator_dir: Path) -> list[str]: def bubblewrap_cmd(generator: str, generator_dir: Path) -> list[str]:
# fmt: off # fmt: off
return nix_shell( return run_cmd(
[ [
"nixpkgs#bash", "bash",
"nixpkgs#bubblewrap", "bubblewrap",
], ],
[ [
"bwrap", "bwrap",

View File

@@ -3,7 +3,7 @@ import subprocess
from pathlib import Path from pathlib import Path
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 run_cmd
from . import SecretStoreBase from . import SecretStoreBase
@@ -16,8 +16,8 @@ class SecretStore(SecretStoreBase):
self, service: str, name: str, value: bytes, groups: list[str] self, service: str, name: str, value: bytes, groups: list[str]
) -> Path | None: ) -> Path | None:
subprocess.run( subprocess.run(
nix_shell( run_cmd(
["nixpkgs#pass"], ["pass"],
["pass", "insert", "-m", f"machines/{self.machine.name}/{name}"], ["pass", "insert", "-m", f"machines/{self.machine.name}/{name}"],
), ),
input=value, input=value,
@@ -27,8 +27,8 @@ class SecretStore(SecretStoreBase):
def get(self, service: str, name: str) -> bytes: def get(self, service: str, name: str) -> bytes:
return subprocess.run( return subprocess.run(
nix_shell( run_cmd(
["nixpkgs#pass"], ["pass"],
["pass", "show", f"machines/{self.machine.name}/{name}"], ["pass", "show", f"machines/{self.machine.name}/{name}"],
), ),
check=True, check=True,
@@ -49,8 +49,8 @@ class SecretStore(SecretStoreBase):
hashes = [] hashes = []
hashes.append( hashes.append(
subprocess.run( subprocess.run(
nix_shell( run_cmd(
["nixpkgs#git"], ["git"],
[ [
"git", "git",
"-C", "-C",
@@ -68,8 +68,8 @@ class SecretStore(SecretStoreBase):
if symlink.is_symlink(): if symlink.is_symlink():
hashes.append( hashes.append(
subprocess.run( subprocess.run(
nix_shell( run_cmd(
["nixpkgs#git"], ["git"],
[ [
"git", "git",
"-C", "-C",

View File

@@ -7,7 +7,7 @@ from tempfile import TemporaryDirectory
from ..cmd import Log, run from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines from ..completions import add_dynamic_completer, complete_machines
from ..machines.machines import Machine from ..machines.machines import Machine
from ..nix import nix_shell from ..nix import run_cmd
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -25,8 +25,8 @@ def upload_secrets(machine: Machine) -> None:
ssh_cmd = host.ssh_cmd() ssh_cmd = host.ssh_cmd()
run( run(
nix_shell( run_cmd(
["nixpkgs#rsync"], ["rsync"],
[ [
"rsync", "rsync",
"-e", "-e",

View File

@@ -13,7 +13,7 @@ from ..dirs import module_root, user_cache_dir, vm_state_dir
from ..errors import ClanError from ..errors import ClanError
from ..facts.generate import generate_facts from ..facts.generate import generate_facts
from ..machines.machines import Machine from ..machines.machines import Machine
from ..nix import nix_shell from ..nix import run_cmd
from .inspect import VmConfig, inspect_vm from .inspect import VmConfig, inspect_vm
from .qemu import qemu_command from .qemu import qemu_command
from .virtiofsd import start_virtiofsd from .virtiofsd import start_virtiofsd
@@ -82,8 +82,8 @@ def prepare_disk(
file_name: str = "disk.img", file_name: str = "disk.img",
) -> Path: ) -> Path:
disk_img = directory / file_name disk_img = directory / file_name
cmd = nix_shell( cmd = run_cmd(
["nixpkgs#qemu"], ["qemu"],
[ [
"qemu-img", "qemu-img",
"create", "create",
@@ -170,11 +170,11 @@ def run_vm(
qga_socket_file=qga_socket_file, qga_socket_file=qga_socket_file,
) )
packages = ["nixpkgs#qemu"] packages = ["qemu"]
env = os.environ.copy() env = os.environ.copy()
if vm.graphics and not vm.waypipe: if vm.graphics and not vm.waypipe:
packages.append("nixpkgs#virt-viewer") packages.append("virt-viewer")
remote_viewer_mimetypes = module_root() / "vms" / "mimetypes" remote_viewer_mimetypes = module_root() / "vms" / "mimetypes"
env["XDG_DATA_DIRS"] = ( env["XDG_DATA_DIRS"] = (
f"{remote_viewer_mimetypes}:{env.get('XDG_DATA_DIRS', '')}" f"{remote_viewer_mimetypes}:{env.get('XDG_DATA_DIRS', '')}"
@@ -185,7 +185,7 @@ def run_vm(
start_virtiofsd(virtiofsd_socket), start_virtiofsd(virtiofsd_socket),
): ):
run( run(
nix_shell(packages, qemu_cmd.args), run_cmd(packages, qemu_cmd.args),
env=env, env=env,
log=Log.BOTH, log=Log.BOTH,
error_msg=f"Could not start vm {machine}", error_msg=f"Could not start vm {machine}",

View File

@@ -6,7 +6,7 @@ from collections.abc import Iterator
from pathlib import Path from pathlib import Path
from ..errors import ClanError from ..errors import ClanError
from ..nix import nix_shell from ..nix import run_cmd
@contextlib.contextmanager @contextlib.contextmanager
@@ -14,8 +14,8 @@ def start_virtiofsd(socket_path: Path) -> Iterator[None]:
sandbox = "namespace" sandbox = "namespace"
if shutil.which("newuidmap") is None: if shutil.which("newuidmap") is None:
sandbox = "none" sandbox = "none"
virtiofsd = nix_shell( virtiofsd = run_cmd(
["nixpkgs#virtiofsd"], ["virtiofsd"],
[ [
"virtiofsd", "virtiofsd",
"--socket-path", "--socket-path",

View File

@@ -5,7 +5,7 @@ import time
from collections.abc import Iterator from collections.abc import Iterator
from ..errors import ClanError from ..errors import ClanError
from ..nix import nix_shell from ..nix import run_cmd
VMADDR_CID_HYPERVISOR = 2 VMADDR_CID_HYPERVISOR = 2
@@ -25,8 +25,8 @@ def start_waypipe(cid: int | None, title_prefix: str) -> Iterator[None]:
if cid is None: if cid is None:
yield yield
return return
waypipe = nix_shell( waypipe = run_cmd(
["nixpkgs#waypipe"], ["waypipe"],
[ [
"waypipe", "waypipe",
"--vsock", "--vsock",

View File

@@ -4,7 +4,7 @@ from pathlib import Path
import pytest import pytest
from clan_cli.custom_logger import setup_logging from clan_cli.custom_logger import setup_logging
from clan_cli.nix import nix_shell from clan_cli.nix import run_cmd
pytest_plugins = [ pytest_plugins = [
"temporary_dir", "temporary_dir",
@@ -32,12 +32,12 @@ def pytest_sessionstart(session: pytest.Session) -> None:
@pytest.fixture @pytest.fixture
def git_repo(tmp_path: Path) -> Path: def git_repo(tmp_path: Path) -> Path:
# initialize a git repository # initialize a git repository
cmd = nix_shell(["nixpkgs#git"], ["git", "init"]) cmd = run_cmd(["git"], ["git", "init"])
subprocess.run(cmd, cwd=tmp_path, check=True) subprocess.run(cmd, cwd=tmp_path, check=True)
# set user.name and user.email # set user.name and user.email
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.name", "test"]) cmd = run_cmd(["git"], ["git", "config", "user.name", "test"])
subprocess.run(cmd, cwd=tmp_path, check=True) subprocess.run(cmd, cwd=tmp_path, check=True)
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.email", "test@test.test"]) cmd = run_cmd(["git"], ["git", "config", "user.email", "test@test.test"])
subprocess.run(cmd, cwd=tmp_path, check=True) subprocess.run(cmd, cwd=tmp_path, check=True)
# return the path to the git repository # return the path to the git repository
return tmp_path return tmp_path

View File

@@ -10,7 +10,7 @@ from clan_cli.clan_uri import FlakeId
from clan_cli.facts.secret_modules.password_store import SecretStore from clan_cli.facts.secret_modules.password_store import SecretStore
from clan_cli.machines.facts import machine_get_fact from clan_cli.machines.facts import machine_get_fact
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 run_cmd
from clan_cli.ssh import HostGroup from clan_cli.ssh import HostGroup
@@ -38,14 +38,10 @@ def test_upload_secret(
""" """
) )
subprocess.run( subprocess.run(
nix_shell( run_cmd(["gnupg"], ["gpg", "--batch", "--gen-key", str(gpg_key_spec)]),
["nixpkgs#gnupg"], ["gpg", "--batch", "--gen-key", str(gpg_key_spec)]
),
check=True, check=True,
) )
subprocess.run( subprocess.run(run_cmd(["pass"], ["pass", "init", "test@local"]), check=True)
nix_shell(["nixpkgs#pass"], ["pass", "init", "test@local"]), check=True
)
cli.run(["facts", "generate", "vm1"]) cli.run(["facts", "generate", "vm1"])
store = SecretStore( store = SecretStore(

View File

@@ -40,6 +40,7 @@ def run_vm_in_thread(machine_name: str) -> None:
# wait for qmp socket to exist # wait for qmp socket to exist
def wait_vm_up(state_dir: Path) -> None: def wait_vm_up(state_dir: Path) -> None:
socket_file = state_dir / "qga.sock" socket_file = state_dir / "qga.sock"
# don't lower this value -> trouble if test machine is slow
timeout: float = 100 timeout: float = 100
while True: while True:
if timeout <= 0: if timeout <= 0:
@@ -55,7 +56,8 @@ def wait_vm_up(state_dir: Path) -> None:
# wait for vm to be down by checking if qga socket is down # wait for vm to be down by checking if qga socket is down
def wait_vm_down(state_dir: Path) -> None: def wait_vm_down(state_dir: Path) -> None:
socket_file = state_dir / "qga.sock" socket_file = state_dir / "qga.sock"
timeout: float = 300 # don't lower this value -> trouble if test machine is slow
timeout: float = 100
while socket_file.exists(): while socket_file.exists():
if timeout <= 0: if timeout <= 0:
raise TimeoutError( raise TimeoutError(