fix flakey tests

tests/distccd/test-script.py

@@ -49,6 +49,19 @@ parser.add_argument("--log-level", type=str)
 #parser.add_argument("--stats", action="store_true")
 #parser.add_argument("--stats-port", type=int)
 
+@beartype
+def parse_systemd_exec(prop: str) -> str:
+    # idk why, but systemd exec lines are secretly DBus dictionaries,
+    # which `systemctl show -p` represents as an equals-delimited, semicolon-separated,
+    # key-value pair, all in curly braces. e.g.:
+    # { path=/nix/store/… ; argv[]=foobar ; ignore_errors=no ; … }
+    inside = prop.removeprefix('{ ').removesuffix(' }')
+    pairs = inside.split(';')
+    # FIXME: don't assume `path` is the first one.
+    # In case systemd ever changes that.
+    _key, path = pairs[0].split('=')
+    return path.strip()
+
 @beartype
 def get_cli_args() -> argparse.Namespace:
     machine.wait_for_unit("distccd.service")
@@ -56,15 +69,10 @@ def get_cli_args() -> argparse.Namespace:
     machine.log(f"{mainpid=}")
     pidtext = machine.succeed(f"pgrep -P {mainpid}")
     machine.log(f"{pidtext=}")
-    pid = int(pidtext.splitlines()[0])
-    machine.log(f"{pid=}")
-    execstart = machine.get_unit_property("distccd.service", "ExecStart")
-    print(f"{execstart=}")
 
-    cmdline = machine.succeed(f"cat /proc/{pid}/cmdline")
-    cmdline_args = cmdline.split("\0")
+    pid = int(pidtext.splitlines()[0])
+    cmdline_args = machine.succeed(rf"cat /proc/{pid}/cmdline | tr '\0' '\n'").splitlines()
     machine.log(f"{cmdline_args=}")
-    print(f"{cmdline_args=}")
 
     args, rest = parser.parse_known_args(cmdline_args)
     return args
@@ -106,6 +114,11 @@ def run_all_tests(machine: Machine, *, use_daemon: bool):
     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")
+
     # Config should have our initial values.
     args = get_cli_args()
     assert args.jobs == 12, f'{args.jobs=} != 12'
@@ -158,10 +171,6 @@ machine.reboot()
 
 machine.wait_for_unit("default.target")
 machine.wait_for_unit("install-dynix.service")
-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")
 try:
     run_all_tests(machine, use_daemon=True)
 except Exception as e:

tests/gotosocial/test-script.py

@@ -41,15 +41,34 @@ def run_log(machine: Machine, *commands: str, timeout: int | None = 60) -> str:
 
     return output
 
+@beartype
+def parse_systemd_exec(prop: str) -> str:
+    # idk why, but systemd exec lines are secretly DBus dictionaries,
+    # which `systemctl show -p` represents as an equals-delimited, semicolon-separated,
+    # key-value pair, all in curly braces. e.g.:
+    # { path=/nix/store/… ; argv[]=foobar ; ignore_errors=no ; … }
+    inside = prop.removeprefix('{ ').removesuffix(' }')
+
+    as_dict = dict()
+    for pair in inside.split(';'):
+        k, v = pair.split('=', maxsplit=1)
+        # They have whitespace around the equals but I don't think that's guaranteed.
+        as_dict[k.strip()] = v.strip()
+
+    # If argv is non-empty, use that entirely.
+    if argv := as_dict["argv[]"]:
+        return argv
+    # Otherwise, just use the path, I guess?
+    return as_dict["path"]
+
 @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")
 
+    execstart = machine.get_unit_property("gotosocial.service", "ExecStart")
+    cmdline_args = parse_systemd_exec(execstart).split()
     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=}")

tests/harmonia/test-script.py

@@ -43,11 +43,13 @@ def run_log(machine: Machine, *commands: str, timeout: int | None = 60) -> str:
 @beartype
 def get_config_file() -> dict[str, Any]:
     machine.wait_for_unit("harmonia.service")
-    pid = int(machine.get_unit_property("harmonia.service", "MainPID"))
-    env_lines: list[str] = machine.succeed(f"cat /proc/{pid}/environ").replace("\0", "\n").splitlines()
-    pairs: list[list[str]] = [line.split("=", maxsplit=1) for line in env_lines]
-    env = dict(pairs)
 
+    # FIXME: this doesn't work if any of the environment variables have spaces,
+    # but idk what else to do.
+    systemd_environ = machine.get_unit_property("harmonia.service", "Environment").split(" ")
+
+    pairs: list[list[str]] = [elem.split("=", maxsplit=1) for elem in systemd_environ]
+    env = dict(pairs)
     config_file = Path(env["CONFIG_FILE"])
 
     machine.log(f"copying from VM: {config_file=}")
@@ -105,8 +107,6 @@ def run_all_tests(machine: Machine, *, use_daemon: bool):
     run_log(machine, "systemctl start dynix-daemon.service")
     machine.wait_for_unit("dynix-daemon.service")
 
-    machine.log("Checking initial harmonia.service conditions")
-
     # Config should have our initial values.
     config_toml = get_config_file()
     assert int(config_toml['workers']) == 4, f"{config_toml['workers']=} != 4"
@@ -127,8 +127,6 @@ def run_all_tests(machine: Machine, *, use_daemon: bool):
     machine.log("Testing that workers, but not max_connection_rate, changed")
     # Workers, but not max connection rate, should have changed.
     config_toml = get_config_file()
-    from pprint import pformat
-    machine.log(pformat(config_toml))
     assert int(config_toml['workers']) == new_workers, f"{config_toml['workers']=} != {new_workers}"
     assert int(config_toml['max_connection_rate']) == 256, f"{config_toml['max_connection_rate']=} != 256"