return True
-def run_shell_cmdline(config: MkosiConfig, pipe: bool = False, commands: Optional[Sequence[str]] = None) -> List[str]:
+def run_shell_setup(config: MkosiConfig, pipe: bool = False, commands: Optional[Sequence[str]] = None) -> List[str]:
if config.output_format in (OutputFormat.directory, OutputFormat.subvolume):
target = f"--directory={config.output}"
else:
def run_shell(config: MkosiConfig) -> None:
- run(run_shell_cmdline(config, pipe=not sys.stdout.isatty()), stdout=sys.stdout, stderr=sys.stderr)
+ run(run_shell_setup(config, pipe=not sys.stdout.isatty()), stdout=sys.stdout, stderr=sys.stderr)
def find_qemu_binary(config: MkosiConfig) -> str:
@contextlib.contextmanager
-def run_qemu_cmdline(config: MkosiConfig) -> Iterator[List[str]]:
+def run_qemu_setup(config: MkosiConfig) -> Iterator[List[str]]:
accel = "kvm" if config.qemu_kvm else "tcg"
firmware, fw_supports_sb = find_qemu_firmware(config)
def run_qemu(config: MkosiConfig) -> None:
- with run_qemu_cmdline(config) as cmdline:
+ with run_qemu_setup(config) as cmdline:
run(cmdline, stdout=sys.stdout, stderr=sys.stderr)
return ["systemd-run", "--quiet", "--wait", "--pipe", "-M", machine_name(config), "/usr/bin/env", *commands]
-def run_ssh_cmdline(config: MkosiConfig, commands: Optional[Sequence[str]] = None) -> List[str]:
+def run_ssh_setup(config: MkosiConfig, commands: Optional[Sequence[str]] = None) -> List[str]:
cmd = [
"ssh",
# Silence known hosts file errors/warnings.
def run_ssh(config: MkosiConfig) -> CompletedProcess:
- return run(run_ssh_cmdline(config), stdout=sys.stdout, stderr=sys.stderr)
+ return run(run_ssh_setup(config), stdout=sys.stdout, stderr=sys.stderr)
def run_serve(config: MkosiConfig) -> None:
parse_args,
parse_boolean,
prepend_to_environ_path,
- run_qemu_cmdline,
- run_shell_cmdline,
- run_ssh_cmdline,
+ run_qemu_setup,
+ run_shell_setup,
+ run_ssh_setup,
run_systemd_cmdline,
unlink_output,
)
prepend_to_environ_path(self.config.extra_search_paths)
if self.config.verb == Verb.boot:
- cmdline = run_shell_cmdline(self.config)
+ cmdline = run_shell_setup(self.config)
elif self.config.verb == Verb.qemu:
# We must keep the temporary file opened at run_qemu_cmdline accessible, hence the context stack.
- cmdline = stack.enter_context(run_qemu_cmdline(self.config))
+ cmdline = stack.enter_context(run_qemu_setup(self.config))
else:
die("No valid verb was entered.")
stderr: Union[int, TextIO] = subprocess.PIPE if capture_output else sys.stderr
if self.config.verb == Verb.qemu:
- cmdline = run_ssh_cmdline(self.config, commands)
+ cmdline = run_ssh_setup(self.config, commands)
elif self.config.verb == Verb.boot:
cmdline = run_systemd_cmdline(self.config, commands)
else:
- cmdline = run_shell_cmdline(self.config, pipe=True, commands=commands)
+ cmdline = run_shell_setup(self.config, pipe=True, commands=commands)
# The retry logic only applies when running commands against a VM.