From 36e784470cdb71c1ff010edf33889639e7c8ddd1 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Thu, 26 Mar 2026 16:38:01 +0100 Subject: [PATCH] f2: turn Nix.daemon into a fixture that way we can parametrize it over the list of protocols we have Change-Id: Ie3fc267ade9c2ca74c163347f4fa58f09a317f40 --- tests/functional2/daemon/test_trust.py | 6 +- .../functional2/store/test_optimise_store.py | 6 +- tests/functional2/testlib/fixtures/nix.py | 122 ++++++++++-------- 3 files changed, 76 insertions(+), 58 deletions(-) diff --git a/tests/functional2/daemon/test_trust.py b/tests/functional2/daemon/test_trust.py index a74e8be22..03a54f86e 100644 --- a/tests/functional2/daemon/test_trust.py +++ b/tests/functional2/daemon/test_trust.py @@ -1,7 +1,7 @@ import pytest import json -from testlib.fixtures.nix import Nix +from testlib.fixtures.nix import Nix, NixDaemon @pytest.mark.parametrize( @@ -13,9 +13,9 @@ from testlib.fixtures.nix import Nix (["*"], ["--force-untrusted"], False), ], ) -def test_trust(nix: Nix, trusted: list[str], flags: list[str], expected: bool): +def test_trust(nix: Nix, daemon: NixDaemon, trusted: list[str], flags: list[str], expected: bool): nix.settings.add_xp_feature("nix-command", "daemon-trust-override") - with nix.daemon(flags, settings={"trusted-users": trusted}) as inner: + with daemon(nix, flags, settings={"trusted-users": trusted}) as inner: trusted = json.loads(inner.nix(["store", "ping", "--json"]).run().ok().stdout) assert trusted["trusted"] == expected diff --git a/tests/functional2/store/test_optimise_store.py b/tests/functional2/store/test_optimise_store.py index c284dcf24..8f909c261 100644 --- a/tests/functional2/store/test_optimise_store.py +++ b/tests/functional2/store/test_optimise_store.py @@ -1,7 +1,7 @@ from pathlib import Path from testlib.fixtures.file_helper import with_files -from testlib.fixtures.nix import Nix +from testlib.fixtures.nix import Nix, NixDaemon from testlib.utils import get_global_asset @@ -40,7 +40,7 @@ class TestOptimizeStore: def test_optimise_store(self, nix: Nix): self._test_optimise_store(nix) - def test_optimise_store_daemon(self, nix: Nix): + def test_optimise_store_daemon(self, nix: Nix, daemon: NixDaemon): nix.settings.auto_optimise_store = True - with nix.daemon([], {"trusted-users": "*"}) as inner: + with daemon(nix, [], {"trusted-users": "*"}) as inner: self._test_optimise_store(inner) diff --git a/tests/functional2/testlib/fixtures/nix.py b/tests/functional2/testlib/fixtures/nix.py index 45e8a1f29..774920f80 100644 --- a/tests/functional2/testlib/fixtures/nix.py +++ b/tests/functional2/testlib/fixtures/nix.py @@ -4,7 +4,7 @@ import dataclasses import sys from functools import partialmethod from pathlib import Path -from typing import Any +from typing import Any, Literal, get_args from collections.abc import Callable, Generator import shutil import subprocess @@ -172,57 +172,6 @@ class Nix: ) -> Command: return self.nix_cmd([nix_exe, *cmd], flake=flake, cwd=cwd) - @contextlib.contextmanager - def daemon( - self, - args: list[str] | None = None, - settings: dict[str, _NixSettingValue] | None = None, - **kwargs, - ) -> "Nix": - daemon = copy.deepcopy(self) - daemon.logger = self.logger.getChild("daemon") - daemon.settings["allowed-users"] = ["*"] - daemon.settings["trusted-users"] = [] - daemon.settings.store = f"local?root={self.env.dirs.test_root}" - daemon.settings.update(settings) - - sockets_dir = Path(daemon.env.dirs.nix_state_dir) / "daemon-socket" - sockets = [sockets_dir / "socket"] - for p in sockets: - p.unlink(missing_ok=True) - - proc = daemon.nix(args or [], nix_exe="nix-daemon", **kwargs).start() - - def log_daemon_result(result: CommandResult | None, level: int): - if result: - daemon.logger.log(level, "daemon exited with code %i", result.rc) - daemon.logger.log(level, "stdout: %s", result.stdout_s) - daemon.logger.log(level, "stderr: %s", result.stderr_s) - else: - daemon.logger.error("daemon exited unexpectedly") - - # wait for daemon to come up. this may take a while under load. - # we only test the *last* socket in the list because that's the - # last one the daemon creates, once it's there the daemon is up - while not sockets[-1].exists(): - if status := proc.wait(0.01): - log_daemon_result(status, logging.ERROR) - raise RuntimeError("daemon exited during startup") - - inner = copy.deepcopy(self) - inner.settings.store = f"unix://{sockets[-1]}" # missing multi socket support - - try: - timeout, level = 1, logging.ERROR - yield inner - # 5 seconds should be enough to wait for a *graceful* exit. - timeout, level = 5, logging.DEBUG - finally: - result = proc.terminate(timeout) - if not result: - result = proc.kill() - log_daemon_result(result, level) - # Mark each of these as correct as they are not ClassVars, but we also don't want to turn off RUF045 nix_build = partialmethod(nix, nix_exe="nix-build") # noqa: RUF045 nix_shell = partialmethod(nix, nix_exe="nix-shell") # noqa: RUF045 @@ -341,6 +290,75 @@ def nix(tmp_path: Path, env: ManagedEnv, logger: logging.Logger) -> Generator[Ni cmd.run().ok() +type NixDaemon = Callable[..., contextlib.AbstractAsyncContextManager[Nix]] + +type NixDaemonProtocol = Literal["legacy-combined"] + +daemon_protocols: list[NixDaemonProtocol] = get_args(NixDaemonProtocol.__value__) + + +# paramterize every daemon tests to run using all supported nix protocols +@pytest.fixture(params=daemon_protocols) +def daemon(request: pytest.FixtureRequest) -> NixDaemon: + default_protocol = request.param + + @contextlib.contextmanager + def wrapper( + nix: Nix, + args: list[str] | None = None, + settings: dict[str, _NixSettingValue] | None = None, + protocol: NixDaemonProtocol | None = None, + **kwargs, + ) -> contextlib.AbstractAsyncContextManager[Nix]: + protocol = protocol or default_protocol + + daemon = copy.deepcopy(nix) + daemon.logger = nix.logger.getChild("daemon") + daemon.settings["allowed-users"] = ["*"] + daemon.settings["trusted-users"] = [] + daemon.settings.store = f"local?root={nix.env.dirs.test_root}" + daemon.settings.update(settings) + + sockets_dir = Path(daemon.env.dirs.nix_state_dir) / "daemon-socket" + sockets = [sockets_dir / "socket"] + for p in sockets: + p.unlink(missing_ok=True) + + proc = daemon.nix(args or [], nix_exe="nix-daemon", **kwargs).start() + + def log_daemon_result(result: CommandResult | None, level: int): + if result: + daemon.logger.log(level, "daemon exited with code %i", result.rc) + daemon.logger.log(level, "stdout: %s", result.stdout_s) + daemon.logger.log(level, "stderr: %s", result.stderr_s) + else: + daemon.logger.error("daemon exited unexpectedly") + + # wait for daemon to come up. this may take a while under load. + # we only test the *last* socket in the list because that's the + # last one the daemon creates, once it's there the daemon is up + while not sockets[-1].exists(): + if status := proc.wait(0.01): + log_daemon_result(status, logging.ERROR) + raise RuntimeError("daemon exited during startup") + + inner = copy.deepcopy(nix) + inner.settings.store = f"unix://{sockets[-1]}" # missing multi socket support + + try: + timeout, level = 1, logging.ERROR + yield inner + # 5 seconds should be enough to wait for a *graceful* exit. + timeout, level = 5, logging.DEBUG + finally: + result = proc.terminate(timeout) + if not result: + result = proc.kill() + log_daemon_result(result, level) + + return wrapper + + @pytest.fixture def enable_diverted_store(nix: Nix): """