{
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, ... }:
{
environment.systemPackages = [ pkgs.sd-switch ];
};
testScript = ''
sd_switch = "${config.sd-switch.command}";
path = "/run/systemd/system"
def install_service(source, name, template_arg=None, tag="1", desc_tag="1", reload_tag="1"):
source_root = "${./.}/services"
machine.succeed(
f"mkdir -p {path}/multi-user.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}/multi-user.target.wants/{name}.service"
)
def install_timer(source, name, tag="1"):
source_root = "${./.}/services"
machine.succeed(
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():
machine.succeed(
f"for svc in {path}/*.service {path}/*.timer; do systemctl stop ''${{svc##*/}}; done"
f" && rm -rv {path}"
)
start_all()
machine.wait_for_unit("multi-user.target")
with subtest("no old, empty new"):
actual = machine.succeed(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")
actual = machine.succeed(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")
cleanup()
with subtest("no old, timer in new"):
install_timer("timer-successful", "sd-timer-1")
actual = machine.succeed(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")
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 = machine.succeed(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")
machine.wait_for_unit("sd-switch-2.service")
machine.wait_for_unit("sd-switch-3.service")
machine.wait_for_unit("sd-switch-4.service")
machine.wait_for_unit("sd-switch-5.service")
machine.wait_for_unit("sd-switch-6.service")
# Get main PID of services that should stay running so we can confirm
# they still run after the switch.
sd_switch_1_pid = machine.succeed("systemctl show -P MainPID sd-switch-1.service").strip()
sd_switch_4_pid = machine.succeed("systemctl show -P MainPID sd-switch-4.service").strip()
sd_switch_5_pid = machine.succeed("systemctl show -P MainPID sd-switch-5.service").strip()
sd_switch_6_pid = machine.succeed("systemctl 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 = machine.succeed(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.
machine.succeed(f"systemctl show -P MainPID sd-switch-1.service | grep {sd_switch_1_pid}").strip()
# Verify that X-Restart-If-Changed=false really worked.
machine.succeed(f"systemctl show -P MainPID sd-switch-4.service | grep {sd_switch_4_pid}").strip()
# Verify that sd-switch-5.service really is left running.
machine.succeed(f"systemctl show -P MainPID sd-switch-5.service | grep {sd_switch_5_pid}").strip()
# Verify that sd-switch-6.service really is left running.
machine.succeed(f"systemctl 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 = machine.succeed(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 = machine.fail("systemctl is-system-running").strip()
expected = "degraded"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
machine.succeed(f"mv {path} /tmp/old-path")
install_service("successful", "sd-switch-1")
actual = machine.succeed(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")
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("missing old/new directories"):
actual = machine.fail(f"{sd_switch} --old-units /no/such/old/dir --new-units /var/empty 2>&1").strip()
expected = "/no/such/old/dir is not a directory"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
actual = machine.fail(f"{sd_switch} --old-units /var/empty --new-units /no/such/new/dir 2>&1").strip()
expected = "/no/such/new/dir is not a directory"
assert actual == expected, f"expected output '{expected}', but got '{actual}'"
'';
};
}