dynix/tests/gotosocial/test-script.py

# SPDX-FileCopyrightText: 2026 Qyriad <qyriad@qyriad.me>
#
# SPDX-License-Identifier: EUPL-1.1

from pathlib import Path
import shlex
import textwrap
from typing import Any, 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"
testing_client = "/root/.nix-profile/libexec/dynix-testing-client.py"

ANSI_RESET = "\x1b[0m"
ANSI_BOLD = "\x1b[1m"
ANSI_NOBOLD = "\x1b[22m"
ANSI_RED = "\x1b[31m"
ANSI_GREEN = "\x1b[32m"

@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()

    machine.logger.info(f"{ANSI_GREEN}INFO{ANSI_RESET}: got config file:")
    machine.logger.info(textwrap.indent(data, "  "))

    return data

@beartype
def dynix_append_cli(option: str, value: Any):
    value = f'"{value}"' if isinstance(value, str) else value
    machine.succeed(f'''
        dynix append {shlex.quote(option)} {shlex.quote(value)}
    '''.strip())

    expr = textwrap.dedent("""
        (import <nixpkgs/nixos> { }).config.dynamicism.applyDynamicConfiguration { }
    """).strip()
    machine.succeed(textwrap.dedent(rf"""
        nix run --show-trace --log-format raw-with-logs --impure -E {shlex.quote(expr)}
    """).strip())


@beartype
def dynix_append_daemon(option: str, value: Any):
    import json
    payload = json.dumps(dict(
        action="append",
        args=dict(
            name=option,
            value=value,
        ),
    ))

    machine.succeed(f"echo '{payload}' | {testing_client} /run/user/0/dynix.sock")

@beartype
def run_all_tests(machine: Machine, *, use_daemon: bool):
    dynix_append = dynix_append_daemon if use_daemon else dynix_append_cli

    dynix_out = machine.succeed("dynix --version")
    assert "dynix" in dynix_out, f"dynix not in {dynix_out=}"

    machine.succeed("systemctl start user@0.service")
    machine.wait_for_unit("user@0.service")
    machine.succeed("systemctl start dynix-daemon.service")
    machine.wait_for_unit("dynix-daemon.service")

    machine.log("REBUILDING configuration inside VM")
    machine.succeed("env PAGER= nixos-rebuild switch --log-format raw-with-logs --no-reexec --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}')

    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 --no-reexec --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=}"

machine.start(allow_reboot=True)
machine.wait_for_unit("default.target")
machine.wait_for_unit("install-dynix.service")
try:
    run_all_tests(machine, use_daemon=False)
except Exception as e:
    machine.logger.error(f"{ANSI_RED}ERROR{ANSI_RESET} during {ANSI_BOLD}CLI{ANSI_RESET} tests: {e}")
    raise

machine.reboot()

machine.wait_for_unit("install-dynix.service")
try:
    run_all_tests(machine, use_daemon=True)
except Exception as e:
    machine.logger.error(f"{ANSI_RED}ERROR{ANSI_RESET} during {ANSI_BOLD}daemon{ANSI_RESET} tests: {e}")
    raise