{
config,
lib,
pkgs,
...
}:
{
options = {
sd-switch = {
command = lib.mkOption {
type = lib.types.str;
description = "The base command for running sd-switch.";
};
};
};
config = {
meta.maintainers = [ pkgs.lib.maintainers.rycee ];
nodes.machine =
{ pkgs, ... }:
{
users.users.alice = {
isNormalUser = true;
description = "Alice Foobar";
password = "foobar";
uid = 1000;
};
environment.systemPackages = [ pkgs.sd-switch ];
};
testScript = ''
sd_switch = "${config.sd-switch.command}";
path = "/home/alice/.config/systemd/user"
def install_service(source, name, template_arg=None, tag="1", desc_tag="1", reload_tag="1"):
source_root = "${./.}/services"
wants_name = name
if template_arg:
wants_name += template_arg
succeed_as_alice(
f"mkdir -p {path}/default.target.wants"
f" && install -m644 {source_root}/{source}.service {path}/{name}.service"
f" && sed -i -e 's/@tag@/{tag}/' -e 's/@desc-tag@/{desc_tag}/' -e 's/@reload-tag@/{reload_tag}/' {path}/{name}.service"
f" && ln -s {path}/{name}.service {path}/default.target.wants/{wants_name}.service"
)
def install_timer(source, name, tag="1"):
source_root = "${./.}/services"
succeed_as_alice(
f"mkdir -p {path}/timers.target.wants"
f" && install -m644 {source_root}/{source}.service {path}/{name}.service"
f" && install -m644 {source_root}/{source}.timer {path}/{name}.timer"
f" && sed -i 's/@tag@/{tag}/' {path}/{name}.service"
f" && sed -i 's/@tag@/{tag}/' {path}/{name}.timer"
f" && ln -s {path}/{name}.timer {path}/timers.target.wants/{name}.timer"
)
def cleanup():
succeed_as_alice(
f"for svc in {path}/*.service {path}/*.timer; do systemctl --user stop ''${{svc##*/}}; done"
f" && rm -rv {path}"
)
def login_as_alice():
machine.wait_until_tty_matches("1", "login: ")
machine.send_chars("alice\n")
machine.wait_until_tty_matches("1", "Password: ")
machine.send_chars("foobar\n")
machine.wait_until_tty_matches("1", "alice\@machine")
def logout_alice():
machine.send_chars("exit\n")
def alice_cmd(cmd):
return f"su -l alice --shell /bin/sh -c $'export XDG_RUNTIME_DIR=/run/user/$UID ; {cmd}'"
def succeed_as_alice(cmd):
return machine.succeed(alice_cmd(cmd))
def fail_as_alice(cmd):
return machine.fail(alice_cmd(cmd))
start_all()
machine.wait_for_unit("multi-user.target")
login_as_alice()
with subtest("no old, empty new"):
actual = succeed_as_alice(f"{sd_switch} --new-units /var/empty")
expected = ""
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
with subtest("no old, non-empty new"):
install_service("successful", "sd-switch-1")
install_service("refuse-manual-start", "sd-switch-2")
actual = succeed_as_alice(f"{sd_switch} --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
cleanup()
with subtest("no old, timer in new"):
install_timer("timer-successful", "sd-timer-1")
actual = succeed_as_alice(f"{sd_switch} --new-units {path}").strip()
expected = "Starting units: sd-timer-1.timer"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-timer-1.timer", user="alice")
cleanup()
with subtest("no old, templated service in new"):
install_service("successful", "sd-template-1@", template_arg="hello")
actual = succeed_as_alice(f"{sd_switch} --new-units {path}").strip()
expected = "Starting units: sd-template-1@hello.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-template-1@hello.service", user="alice")
cleanup()
with subtest("unmanaged unit replaced by managed"):
install_service("successful", "sd-switch-1")
install_service("successful", "sd-switch-2")
actual = succeed_as_alice(f"{sd_switch} --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service, sd-switch-2.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
machine.wait_for_unit("sd-switch-2.service", user="alice")
succeed_as_alice(f"rm {path}/default.target.wants/sd-switch-2.service {path}/sd-switch-2.service")
install_service("refuse-manual-start", "sd-switch-2")
actual = succeed_as_alice(f"{sd_switch} --old-units /var/empty --new-units {path}").strip()
expected = (
"Stopping units: sd-switch-1.service\n"
"Keeping old units: sd-switch-2.service\n"
"Starting units: sd-switch-1.service"
)
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
cleanup()
with subtest("unmanaged unit with refuse manual stop replaced by managed"):
install_service("refuse-manual-stop", "sd-switch-1")
actual = succeed_as_alice(f"{sd_switch} --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
succeed_as_alice(f"rm -r {path}")
install_service("successful", "sd-switch-1")
actual = succeed_as_alice(f"{sd_switch} --old-units /var/empty --new-units {path}").strip()
expected = "Keeping old units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
cleanup()
with subtest("unmanaged unit replaced by managed with refuse manual start"):
install_service("successful", "sd-switch-1")
actual = succeed_as_alice(f"{sd_switch} --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
succeed_as_alice(f"rm -r {path}")
install_service("refuse-manual-start", "sd-switch-1")
actual = succeed_as_alice(f"{sd_switch} --old-units /var/empty --new-units {path}").strip()
expected = "Keeping old units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
cleanup()
# Start up some services then switch to a new generation such that
#
# - one service is unchanged,
# we expect it to remain in its current state, running;
#
# - one service is changed only in Unit.Description,
# we expect it to remain in its current state, running;
#
# - one service is changed only in Unit.X-Reload-Triggers,
# we expect it to be reloaded;
#
# - one service is removed,
# we expect it to be stopped;
#
# - one service is changed,
# we expect it to be restarted (through stop and start);
#
# - one service is replaced by a service with `X-RestartIfChanged = false`,
# we expect the old service to keep running;
with subtest("non-empty old, non-empty new"):
install_service("successful", "sd-switch-1")
install_service("successful", "sd-switch-2")
install_service("successful", "sd-switch-3")
install_service("successful", "sd-switch-4")
install_service("successful", "sd-switch-5")
install_service("successful", "sd-switch-6")
actual = succeed_as_alice(f"{sd_switch} --new-units {path}").strip()
expected = (
"Starting units: sd-switch-1.service, sd-switch-2.service,"
" sd-switch-3.service, sd-switch-4.service, sd-switch-5.service,"
" sd-switch-6.service"
)
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
machine.wait_for_unit("sd-switch-2.service", user="alice")
machine.wait_for_unit("sd-switch-3.service", user="alice")
machine.wait_for_unit("sd-switch-4.service", user="alice")
machine.wait_for_unit("sd-switch-5.service", user="alice")
machine.wait_for_unit("sd-switch-6.service", user="alice")
# Get main PID of services that should stay running so we can confirm
# they still run after the switch.
sd_switch_1_pid = succeed_as_alice("systemctl --user show -P MainPID sd-switch-1.service").strip()
sd_switch_4_pid = succeed_as_alice("systemctl --user show -P MainPID sd-switch-4.service").strip()
sd_switch_5_pid = succeed_as_alice("systemctl --user show -P MainPID sd-switch-5.service").strip()
sd_switch_6_pid = succeed_as_alice("systemctl --user show -P MainPID sd-switch-6.service").strip()
machine.succeed(f"mv {path} /tmp/old-path")
install_service("successful", "sd-switch-1")
install_service("successful", "sd-switch-3", tag="changed")
install_service("x-restart-if-changed-false", "sd-switch-4", tag="changed")
install_service("successful", "sd-switch-5", desc_tag="changed")
install_service("successful", "sd-switch-6", reload_tag="changed")
actual = succeed_as_alice(f"{sd_switch} --old-units /tmp/old-path --new-units {path}").strip()
expected = (
"Stopping units: sd-switch-2.service, sd-switch-3.service\n"
"Reloading units: sd-switch-6.service\n"
"Keeping old units: sd-switch-4.service\n"
"Starting units: sd-switch-3.service"
)
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
# Verify that sd-switch-1.service really is left running.
succeed_as_alice(f"systemctl --user show -P MainPID sd-switch-1.service | grep {sd_switch_1_pid}").strip()
# Verify that X-Restart-If-Changed=false really worked.
succeed_as_alice(f"systemctl --user show -P MainPID sd-switch-4.service | grep {sd_switch_4_pid}").strip()
# Verify that sd-switch-5.service really is left running.
succeed_as_alice(f"systemctl --user show -P MainPID sd-switch-5.service | grep {sd_switch_5_pid}").strip()
# Verify that sd-switch-6.service really is left running.
succeed_as_alice(f"systemctl --user show -P MainPID sd-switch-6.service | grep {sd_switch_6_pid}").strip()
cleanup()
machine.succeed("rm -r /tmp/old-path")
# Start a failing service, verify that systemd is degraded, and then switch
# to successful service.
with subtest("handles failed service"):
install_service("failing", "sd-switch-1")
actual = succeed_as_alice(f"{sd_switch} --old-units /var/empty --new-units {path}").strip()
expected = (
"Starting units: sd-switch-1.service\n"
"sd-switch-1.service failed"
)
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
actual = fail_as_alice("systemctl --user is-system-running").strip()
expected = "degraded"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
succeed_as_alice(f"mv {path} /tmp/old-path")
install_service("successful", "sd-switch-1")
actual = succeed_as_alice(f"{sd_switch} --old-units /tmp/old-path --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
cleanup()
machine.succeed("rm -r /tmp/old-path")
with subtest("inside wrong D-Bus session"):
install_service("successful", "sd-switch-1")
actual = succeed_as_alice(f"dbus-run-session -- {sd_switch} --old-units /var/empty --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
cleanup()
with subtest("detects supported language in LC_ALL"):
install_service("successful", "sd-switch-1")
actual = succeed_as_alice(f"env LC_ALL=sv_SE.UTF-8 {sd_switch} --new-units {path}").strip()
expected = "Startar enheter: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
cleanup()
with subtest("falls back to en_US for unsupported language in LC_ALL"):
install_service("successful", "sd-switch-1")
actual = succeed_as_alice(f"env LC_ALL=xx_XX.UTF-8 {sd_switch} --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
cleanup()
# Start a service, then switch to configuration the service marked as
# refuse manual start. Verify that the old service was stopped and no
# attempt to start the new one.
with subtest("handles refuse manual start service"):
install_service("successful", "sd-switch-1")
actual = succeed_as_alice(f"{sd_switch} --old-units /var/empty --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
succeed_as_alice(f"mv {path} /tmp/old-path")
install_service("refuse-manual-start", "sd-switch-1")
install_service("refuse-manual-start", "sd-switch-2")
actual = succeed_as_alice(f"{sd_switch} --old-units /tmp/old-path --new-units {path}").strip()
expected = "Stopping units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
actual = machine.get_unit_info("sd-switch-1.service", user="alice")
expected = "inactive"
assert actual["ActiveState"] == expected, \
f"expected unit state '{expected}', but got '{actual}'"
cleanup()
machine.succeed("rm -r /tmp/old-path")
# Start a service with refuse manual stop, then switch to configuration
# without the service. Verify that the service still is running.
#
# Also include a new service with refuse manual start in the new
# configuration, verify that the service is not started.
#
# Note, this test needs to be last since it leaves a running service.
with subtest("handles refuse manual start/stop service"):
install_service("refuse-manual-stop", "sd-switch-1")
actual = succeed_as_alice(f"{sd_switch} --old-units /var/empty --new-units {path}").strip()
expected = "Starting units: sd-switch-1.service"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
succeed_as_alice(f"mv {path} /tmp/old-path")
install_service("successful", "sd-switch-2")
actual = succeed_as_alice(f"{sd_switch} --old-units /tmp/old-path --new-units {path}").strip()
expected = (
"Keeping old units: sd-switch-1.service\n"
"Starting units: sd-switch-2.service"
)
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.wait_for_unit("sd-switch-1.service", user="alice")
machine.wait_for_unit("sd-switch-2.service", user="alice")
cleanup()
machine.succeed("rm -r /tmp/old-path")
'';
};
}