From 4da9a53b6606a008cc8de79e35a96515c8f2fd88 Mon Sep 17 00:00:00 2001
From: eldritch horrors <pennae@lix.systems>
Date: Sun, 3 May 2026 16:32:26 +0200
Subject: [PATCH] f2: add test for daemon socket connection

tests both order of sockets tried and that all sockets work as expected.

Change-Id: I5aeb506e159564a0b14b6fbbe5df1e5ccdb437d5
---
 tests/functional2/daemon/test_connect.py | 97 ++++++++++++++++++++++++
 1 file changed, 97 insertions(+)
 create mode 100644 tests/functional2/daemon/test_connect.py

diff --git a/tests/functional2/daemon/test_connect.py b/tests/functional2/daemon/test_connect.py
new file mode 100644
index 000000000..464c80a39
--- /dev/null
+++ b/tests/functional2/daemon/test_connect.py
@@ -0,0 +1,97 @@
+import pytest
+import itertools
+
+from testlib.fixtures.nix import Nix
+
+
+@pytest.fixture(autouse=True)
+def setup(nix: Nix):
+    nix.settings.add_xp_feature("nix-command")
+
+
+def _observe_socket_order(nix: Nix, uri: str) -> list[str]:
+    cmd = nix.nix(["store", "ping", "--store", uri, "--debug"]).run()
+    cmd.expect(1)
+    return [
+        line.removeprefix("skipping socket ").split(":")[0]
+        for line in cmd.stderr_s.splitlines()
+        if line.startswith("skipping socket ")
+    ]
+
+
+def test_connection_order_plain(nix: Nix):
+    assert _observe_socket_order(nix, f"unix://{nix.env.dirs.home}/socket") == [
+        str(nix.env.dirs.home / "socket")
+    ]
+
+
+def test_connection_order_any(nix: Nix):
+    assert _observe_socket_order(nix, f"unix://{nix.env.dirs.home}?protocol=any") == [
+        str(nix.env.dirs.home / "socket")
+    ]
+
+
+def test_connection_protocol_invalid(nix: Nix):
+    cmd = nix.nix(["store", "ping", "--store", "unix:///dev/null?protocol=invalid"]).run()
+    cmd.expect(1)
+    assert "unknown daemon protocol invalid" in cmd.stderr_s
+
+
+def test_connection_protocol_any_not_standalone(nix: Nix):
+    cmd = nix.nix(["store", "ping", "--store", "unix:///dev/null?protocol=any,any"]).run()
+    cmd.expect(1)
+    assert "unknown daemon protocol any" in cmd.stderr_s
+
+
+_protocols: dict[str, str] = {"legacy-combined": "."}
+
+
+@pytest.mark.parametrize(
+    ("protos", "sockets"),
+    [
+        (list(ps), [_protocols[p] for p in ps])
+        for r in range(1, len(_protocols) + 1)
+        for ps in itertools.permutations(_protocols, r)
+    ],
+)
+def test_connection_order_specific_unix(nix: Nix, protos: list[str], sockets: list[str]):
+    assert _observe_socket_order(
+        nix, f"unix://{nix.env.dirs.home}?protocol={','.join(protos)}"
+    ) == [str(nix.env.dirs.home / s) for s in sockets]
+
+
+# this once can't handle legacy-combined sockets because those don't search paths
+@pytest.mark.parametrize(
+    ("protos", "sockets"),
+    [
+        (list(ps), [_protocols[p] for p in ps])
+        for r in range(1, len(_protocols) + 1)
+        for ps in itertools.permutations(
+            {t: p for t, p in _protocols.items() if t != "legacy-combined"}, r
+        )
+    ],
+)
+def test_connection_order_specific_daemon_modern(nix: Nix, protos: list[str], sockets: list[str]):
+    assert _observe_socket_order(nix, f"daemon?protocol={','.join(protos)}") == [
+        str(nix.env.dirs.nix_state_dir / "daemon-socket" / s) for s in sockets
+    ]
+
+
+# this once can't handle legacy-combined sockets because those don't search paths
+@pytest.mark.parametrize(
+    ("protos", "sockets"),
+    [
+        (list(ps), [_protocols[p] for p in ps])
+        for r in range(1, len(_protocols) + 1)
+        for ps in itertools.permutations(
+            {t: p for t, p in _protocols.items() if t != "legacy-combined"}, r
+        )
+    ],
+)
+def test_connection_order_specific_daemon_modern_relocated(
+    nix: Nix, protos: list[str], sockets: list[str]
+):
+    nix.env["LIX_DAEMON_SOCKET_DIR"] = str(nix.env.dirs.home)
+    assert _observe_socket_order(nix, f"daemon?protocol={','.join(protos)}") == [
+        str(nix.env.dirs.home / s) for s in sockets
+    ]