from pathlib import Path import shlex import textwrap from typing import cast, TYPE_CHECKING from beartype import beartype from test_driver.machine import Machine from test_driver.errors import RequestedAssertionFailed if TYPE_CHECKING: global machine machine = cast(Machine, ...) assert machine.shell is not None ls = "eza -lah --color=always --group-directories-first" @beartype def run_log(machine: Machine, *commands: str, timeout: int | None = 60) -> str: output = "" for command in commands: with machine.nested(f"must succeed: {command}"): (status, out) = machine.execute(f"{command} | tee /dev/stderr", timeout=timeout) if status != 0: machine.log(f"output: {out}") raise RequestedAssertionFailed( f"command `{command}` failed (exit code {status})", ) output += out return output @beartype def get_config_file() -> str: machine.wait_for_unit("gotosocial.service") gotosocial_pid = int(machine.get_unit_property("gotosocial.service", "MainPID")) cmdline = machine.succeed(f"cat /proc/{gotosocial_pid}/cmdline") cmdline_args = cmdline.split("\0") config_file_idx = cmdline_args.index("--config-path") + 1 config_file = Path(cmdline_args[config_file_idx]) machine.log(f"copying from VM: {config_file=}") machine.copy_from_vm(config_file.as_posix()) config_file_path = machine.out_dir / config_file.name with open(config_file_path, "r") as f: data = f.read() config_file_path.unlink() return data @beartype def dynix_append(option: str, value: str): machine.succeed(f''' dynix append {shlex.quote(option)} {shlex.quote(value)} '''.strip()) @beartype def do_apply(): expr = textwrap.dedent(""" (import { }).config.dynamicism.applyDynamicConfiguration { } """).strip() machine.succeed(rf""" nix run --show-trace --log-format raw-with-logs --impure -E {shlex.quote(expr)} """.strip()) machine.wait_for_unit("default.target") machine.wait_for_unit("install-dynix.service") dynix_out = machine.succeed("dynix --version") assert "dynix" in dynix_out, f"dynix not in {dynix_out=}" machine.log("REBUILDING configuration inside VM") machine.succeed("env PAGER= nixos-rebuild switch --log-format raw-with-logs --fallback") machine.wait_for_unit("gotosocial.service") # Make sure the config before any dynamic changes is what we expect. config_text = get_config_file() lines = config_text.splitlines() try: application_name = next(line for line in lines if line.startswith("application-name:")) except StopIteration: raise AssertionError(f"no 'application-name:' found in config file: {textwrap.indent(config_text, " ")}") assert "gotosocial-for-machine" in application_name, f"'gotosocial-for-machine' should be in {application_name=}" try: host = next(line for line in lines if line.startswith("host:")) except StopIteration: raise AssertionError(f"no 'host:' found in config file: {textwrap.indent(config_text, " ")}") assert "gotosocial-machine" in host, f"'gotosocial-machine' should be in {host=}" new_app_name = "yay!" dynix_append("services.gotosocial.settings.application-name", f'"{new_app_name}"') do_apply() config_text = get_config_file() lines = config_text.splitlines() try: application_name = next(line for line in lines if line.startswith("application-name:")) except StopIteration: raise AssertionError(f"no 'application-name:' found in config file: {textwrap.indent(config_text, " ")}") assert new_app_name in application_name, f"'{new_app_name}' should be in {application_name=}" machine.log("REBUILDING configuration inside VM") machine.succeed("env PAGER= nixos-rebuild switch --log-format raw-with-logs --fallback") machine.wait_for_unit("gotosocial.service") config_text = get_config_file() lines = config_text.splitlines() try: application_name = next(line for line in lines if line.startswith("application-name:")) except StopIteration: raise AssertionError(f"no 'application-name:' found in config file: {textwrap.indent(config_text, " ")}") assert "gotosocial-for-machine" in application_name, f"'gotosocial-for-machine' should be in {application_name=}"