group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true
+ strategy:
+ matrix:
+ distro:
+ - fedora
+
steps:
- uses: actions/checkout@v2
- uses: ./
+ - name: Configure test distribution
+ run: |
+ mkdir -p mkosi.default.d
+
+ tee mkosi.default.d/test-distribution.conf <<- EOF
+ [Distribution]
+ Distribution=${{ matrix.distro }}
+ EOF
+
- name: Install dependencies
run: sudo apt-get update && sudo apt-get install --no-install-recommends python3-pexpect python3-pytest
- name: Run integration tests
- run: sudo python3 -m pytest tests
+ run: sudo python3 -m pytest -m integration tests
integration-test:
runs-on: ubuntu-20.04
)
from .backend import (
+ _FILE,
ARG_DEBUG,
Distribution,
ManifestFormat,
return True
-def run_shell(args: MkosiArgs) -> None:
+def run_shell_cmdline(args: MkosiArgs) -> List[str]:
if args.output_format in (OutputFormat.directory, OutputFormat.subvolume):
target = f"--directory={args.output}"
else:
cmdline += ["--"]
cmdline += args.cmdline
+ return cmdline
+
+
+def run_shell(args: MkosiArgs) -> None:
with suppress_stacktrace():
- run(cmdline, stdout=sys.stdout, stderr=sys.stderr)
+ run(run_shell_cmdline(args), stdout=sys.stdout, stderr=sys.stderr)
def find_qemu_binary() -> str:
die("Couldn't find OVMF UEFI variables file.")
-def run_qemu(args: MkosiArgs) -> None:
+@contextlib.contextmanager
+def run_qemu_cmdline(args: MkosiArgs) -> Iterator[List[str]]:
has_kvm = os.path.exists("/dev/kvm")
accel = "kvm" if has_kvm else "tcg"
cmdline += args.cmdline
print_running_cmd(cmdline)
+ yield cmdline
+
+def run_qemu(args: MkosiArgs) -> None:
+ with run_qemu_cmdline(args) as cmdline:
with suppress_stacktrace():
run(cmdline, stdout=sys.stdout, stderr=sys.stderr)
die("Container/VM address not found")
-def run_ssh(args: MkosiArgs) -> None:
+def run_command_image(args: MkosiArgs, commands: Sequence[str], timeout: int, check: bool, stdout: _FILE = sys.stdout, stderr: _FILE = sys.stderr) -> CompletedProcess:
+ if args.verb == "qemu":
+ return run_ssh(args, commands, check, stdout, stderr, timeout)
+ else:
+ cmdline = ["systemd-run", "--quiet", "--wait", "--pipe", "-M", virt_name(args), "/usr/bin/env", *commands]
+ with suppress_stacktrace():
+ return run(cmdline, check=check, stdout=stdout, stderr=stderr, text=True, timeout=timeout)
+
+
+def run_ssh_cmdline(args: MkosiArgs, commands: Optional[Sequence[str]] = None) -> Sequence[str]:
cmd = [
"ssh",
# Silence known hosts file errors/warnings.
cmd += ["-p", f"{args.ssh_port}"]
dev, address = find_address(args)
- cmd += [f"root@{address}{dev}", *args.cmdline]
+ cmd += [f"root@{address}{dev}"]
+ cmd += commands or args.cmdline
+
+ return cmd
+
+def run_ssh(
+ args: MkosiArgs,
+ commands: Optional[Sequence[str]] = None,
+ check: bool = True,
+ stdout: _FILE = sys.stdout,
+ stderr: _FILE = sys.stderr,
+ timeout: Optional[int] = None,
+) -> CompletedProcess:
with suppress_stacktrace():
- run(cmd, stdout=sys.stdout, stderr=sys.stderr)
+ return run(run_ssh_cmdline(args, commands), check=check, stdout=stdout, stderr=stderr, text=True, timeout=timeout)
def run_serve(args: MkosiArgs) -> None:
--- /dev/null
+# SPDX-License-Identifier: LGPL-2.1+
+
+from __future__ import annotations
+
+import contextlib
+import signal
+import subprocess
+from textwrap import dedent
+from typing import Any, Optional, Sequence
+
+import pexpect # type: ignore
+
+from . import (
+ MKOSI_COMMANDS_SUDO,
+ CompletedProcess,
+ build_stuff,
+ check_native,
+ check_output,
+ check_root,
+ init_namespace,
+ load_args,
+ needs_build,
+ parse_args,
+ prepend_to_environ_path,
+ run_command_image,
+ run_qemu_cmdline,
+ run_shell_cmdline,
+ unlink_output,
+)
+from .backend import MkosiArgs, die
+
+
+class Machine:
+ def __init__(self, args: Optional[Sequence[str]] = None, debug: bool = False) -> None:
+ # Remains None until image is built and booted, then receives pexpect process
+ self._serial: Optional[pexpect.spawn] = None
+ self.exit_code: int = -1
+ self.debug = debug
+ self.stack = contextlib.ExitStack()
+ self.args: MkosiArgs
+
+ # We make sure to add the arguments in the machine class itself, rather than typing this for every testing function.
+ tmp = parse_args(args)["default"]
+ tmp.force = 1
+ tmp.autologin = True
+ if tmp.verb == "qemu":
+ tmp.bootable = True
+ tmp.qemu_headless = True
+ tmp.hostonly_initrd = True
+ tmp.network_veth = True
+ tmp.ssh = True
+ elif tmp.verb == "boot":
+ pass
+ else:
+ die("No valid verb was entered.")
+
+ self.args = load_args(tmp)
+
+ @property
+ def serial(self) -> pexpect.spawn:
+ if self._serial is None:
+ raise ValueError(
+ dedent(
+ """\
+ Trying to access serial console before machine boot or after machine shutdown.
+ In order to boot the machine properly, use it as a context manager.
+ Then, a Mkosi image will be booted in the __enter__ method.
+ """
+ )
+ )
+ return self._serial
+
+ def ensure_booted(self) -> None:
+ # Try to access the serial console which will raise an exception if the machine is not currently booted.
+ assert self._serial is not None
+
+ def __enter__(self) -> Machine:
+ if self.args.verb in MKOSI_COMMANDS_SUDO:
+ check_root()
+ unlink_output(self.args)
+
+ if self.args.verb == "build":
+ check_output(self.args)
+
+ if needs_build(self.args):
+ check_root()
+ check_native(self.args)
+ init_namespace(self.args)
+ build_stuff(self.args)
+
+ with contextlib.ExitStack() as stack:
+ prepend_to_environ_path(self.args.extra_search_paths)
+
+ if self.args.verb in ("shell", "boot"):
+ cmdline = run_shell_cmdline(self.args)
+ elif self.args.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.args))
+ else:
+ die("No valid verb was entered.")
+
+ cmd = " ".join(str(x) for x in cmdline)
+
+ # Here we have something equivalent to the command lines used on spawn() and run() from backend.py
+ # We use pexpect to boot an image that we will be able to interact with in the future
+ # Then we tell the process to look for the # sign, which indicates the CLI for that image is active
+ # Once we've build/boot an image the CLI will prompt "root@image ~]# "
+ # Then, when pexpects finds the "#" it means we're ready to interact with the process
+ self._serial = pexpect.spawnu(cmd, logfile=None, timeout=240)
+ self._serial.expect("#")
+ self.stack = stack.pop_all()
+
+ return self
+
+ def run(self, commands: Sequence[str], timeout: int = 900, check: bool = True) -> CompletedProcess:
+ self.ensure_booted()
+
+ process = run_command_image(self.args, commands, timeout, check, subprocess.PIPE, subprocess.PIPE)
+ if self.debug:
+ print(f"Stdout:\n {process.stdout}")
+ print(f"Stderr:\n {process.stderr}")
+
+ return process
+
+ def __exit__(self, *args: Any, **kwargs: Any) -> None:
+ if self._serial:
+ self._serial.kill(signal.SIGTERM)
+ self.exit_code = self._serial.wait()
+ self._serial = None
+ self.stack.__exit__(*args, **kwargs)
--- /dev/null
+# SPDX-License-Identifier: LGPL-2.1+
+
+import os
+from subprocess import TimeoutExpired
+
+import pytest
+
+import mkosi.machine as machine
+from mkosi.backend import MkosiException
+
+
+pytestmark = [
+ pytest.mark.integration,
+ pytest.mark.parametrize("verb", ["boot", "qemu"]),
+ pytest.mark.skipif(os.getuid() != 0, reason="Must be invoked as root.")
+]
+
+
+def test_simple_run(verb: str) -> None:
+ with machine.Machine([verb]) as m:
+ p = m.run(["echo", "This is a test."])
+ assert "This is a test." == p.stdout.strip("\n")
+
+ assert m.exit_code == 0
+
+
+def test_wrong_command(verb: str) -> None:
+ # First tests with argument check = True from mkosi.backend.run(), therefore we see if an exception is raised
+ with machine.Machine([verb]) as m:
+ with pytest.raises(MkosiException):
+ m.run(["NonExisting", "Command"])
+ with pytest.raises(MkosiException):
+ m.run(["ls", "NullDirectory"])
+
+ assert m.exit_code == 0
+
+ # Second group of tests with check = False to see if stderr and returncode have the expected values
+ with machine.Machine([verb]) as m:
+ result = m.run(["NonExisting", "Command"], check=False)
+ assert result.returncode in (203, 127)
+
+ result = m.run(["ls", "-"], check=False)
+ assert result.returncode == 2
+ assert "No such file or directory" in result.stderr
+
+ assert m.exit_code == 0
+
+
+def test_infinite_command(verb: str) -> None:
+ with machine.Machine([verb]) as m:
+ with pytest.raises(TimeoutExpired):
+ m.run(["tail", "-f", "/dev/null"], 2)
+
+ assert m.exit_code == 0
+
+
+def test_before_boot(verb: str) -> None:
+ m = machine.Machine([verb])
+ with pytest.raises(AssertionError):
+ m.run(["ls"])
+
+
+def test_after_shutdown(verb: str) -> None:
+ with machine.Machine([verb]) as m:
+ pass
+
+ with pytest.raises(AssertionError):
+ m.run(["ls"])
+ assert m.exit_code == 0
projects / thirdparty / mkosi.git / commitdiff
