Merge pull request 'replace environment variable with nixpkgs directory' (#183) from Mic92-main into main

.gitignore

@@ -1,5 +1,6 @@
 .direnv
 result*
+pkgs/clan-cli/clan_cli/nixpkgs
 
 # python
 __pycache__

pkgs/clan-cli/clan_cli/dirs.py

@@ -24,3 +24,19 @@ def user_config_dir() -> Path:
         return Path(os.path.expanduser("~/Library/Application Support/"))
     else:
         return Path(os.getenv("XDG_CONFIG_HOME", os.path.expanduser("~/.config")))
+
+
+def module_root() -> Path:
+    return Path(__file__).parent
+
+
+def flake_registry() -> Path:
+    return module_root() / "nixpkgs" / "flake-registry.json"
+
+
+def nixpkgs() -> Path:
+    return (module_root() / "nixpkgs" / "path").resolve()
+
+
+def unfree_nixpkgs() -> Path:
+    return module_root() / "nixpkgs" / "unfree"

pkgs/clan-cli/clan_cli/nix.py

@@ -1,10 +1,42 @@
 import os
 
+from .dirs import flake_registry, unfree_nixpkgs
+
 
 def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
-    flake = os.environ.get("CLAN_FLAKE")
+    # we cannot use nix-shell inside the nix sandbox
-    # in unittest we will have all binaries provided
+    # in our tests we just make sure we have all the packages
-    if flake is None:
+    if os.environ.get("IN_NIX_SANDBOX"):
         return cmd
-    wrapped_packages = [f"path:{flake}#{p}" for p in packages]
+    wrapped_packages = [f"nixpkgs#{p}" for p in packages]
-    return ["nix", "shell"] + wrapped_packages + ["-c"] + cmd
+    return (
+        [
+            "nix",
+            "shell",
+            "--extra-experimental-features",
+            "nix-command flakes",
+            "--flake-registry",
+            str(flake_registry()),
+        ]
+        + wrapped_packages
+        + ["-c"]
+        + cmd
+    )
+
+
+def unfree_nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
+    if os.environ.get("IN_NIX_SANDBOX"):
+        return cmd
+    return (
+        [
+            "nix",
+            "shell",
+            "--extra-experimental-features",
+            "nix-command flakes",
+            "-f",
+            str(unfree_nixpkgs()),
+        ]
+        + packages
+        + ["-c"]
+        + cmd
+    )

pkgs/clan-cli/clan_cli/zerotier/__init__.py

@@ -9,7 +9,7 @@ from tempfile import TemporaryDirectory
 from typing import Any, Iterator, Optional
 
 from ..errors import ClanError
-from ..nix import nix_shell
+from ..nix import nix_shell, unfree_nix_shell
 
 
 def try_bind_port(port: int) -> bool:
@@ -87,7 +87,10 @@ def zerotier_controller() -> Iterator[ZerotierController]:
     controller_port = find_free_port(range(10000, 65535))
     if controller_port is None:
         raise ClanError("cannot find a free port for zerotier controller")
-    cmd = nix_shell(["bash", "zerotierone"], ["bash", "-c", "command -v zerotier-one"])
+
+    cmd = unfree_nix_shell(
+        ["bash", "zerotierone"], ["bash", "-c", "command -v zerotier-one"]
+    )
     res = subprocess.run(
         cmd,
         check=True,
@@ -102,7 +105,6 @@ def zerotier_controller() -> Iterator[ZerotierController]:
         raise ClanError(
             f"zerotier-one executable needs to come from /nix/store: {zerotier_exe}"
         )
-
     with TemporaryDirectory() as d:
         tempdir = Path(d)
         home = tempdir / "zerotier-one"

pkgs/clan-cli/default.nix

@@ -48,6 +48,17 @@ let
     chmod -R +w $out
     rm $out/clan_cli/config/jsonschema
     cp -r ${self + /lib/jsonschema} $out/clan_cli/config/jsonschema
+    ln -s ${nixpkgs} $out/clan_cli/nixpkgs
+  '';
+  nixpkgs = runCommand "nixpkgs" { } ''
+    mkdir -p $out/unfree
+    cat > $out/unfree/default.nix <<EOF
+    import "${pkgs.path}" { config = { allowUnfree = true; overlays = []; }; }
+    EOF
+    cat > $out/flake-registry.json <<EOF
+    { "flakes": [{"exact":true,"from":{"id":"nixpkgs", "type": "indirect"},"to": {"path":"${pkgs.path}", "type":"path"}}], "version": 2}
+    EOF
+    ln -s ${pkgs.path} $out/path
   '';
 in
 python3.pkgs.buildPythonPackage {
@@ -56,8 +67,6 @@ python3.pkgs.buildPythonPackage {
   format = "pyproject";
 
   inherit CLAN_OPTIONS_FILE;
-  # This is required for the python tests, where some nix libs depend on nixpkgs
-  CLAN_NIXPKGS = pkgs.path;
 
   nativeBuildInputs = [
     setuptools
@@ -68,13 +77,13 @@ python3.pkgs.buildPythonPackage {
   passthru.tests.clan-pytest = runCommand "clan-tests"
     {
       nativeBuildInputs = [ age zerotierone bubblewrap sops nix openssh rsync stdenv.cc ];
-      CLAN_NIXPKGS = pkgs.path;
     } ''
     export CLAN_OPTIONS_FILE="${CLAN_OPTIONS_FILE}"
     cp -r ${source} ./src
     chmod +w -R ./src
     cd ./src
-    NIX_STATE_DIR=$TMPDIR/nix ${checkPython}/bin/python -m pytest -s ./tests
+
+    NIX_STATE_DIR=$TMPDIR/nix IN_NIX_SANDBOX=1 ${checkPython}/bin/python -m pytest -s ./tests
     touch $out
   '';
   passthru.clan-openapi = runCommand "clan-openapi" { } ''
@@ -84,6 +93,7 @@ python3.pkgs.buildPythonPackage {
     ${checkPython}/bin/python ./bin/gen-openapi --out $out/openapi.json --app-dir . clan_cli.webui.app:app
     touch $out
   '';
+  passthru.nixpkgs = nixpkgs;
 
   passthru.devDependencies = [
     setuptools
@@ -92,11 +102,8 @@ python3.pkgs.buildPythonPackage {
 
   passthru.testDependencies = dependencies ++ testDependencies;
 
-  makeWrapperArgs = [
-    "--set CLAN_FLAKE ${self}"
-  ];
-
   postInstall = ''
+    ln -s ${nixpkgs} $out/${python3.sitePackages}/nixpkgs
     installShellCompletion --bash --name clan \
       <(${argcomplete}/bin/register-python-argcomplete --shell bash clan)
     installShellCompletion --fish --name clan.fish \

pkgs/clan-cli/pyproject.toml

@@ -8,8 +8,8 @@ description = "cLAN CLI tool"
 dynamic = [ "version" ]
 scripts = { clan = "clan_cli:main" }
 
-[tool.setuptools.packages]
+[tool.setuptools.packages.find]
-find = {}
+exclude = ["clan_cli.nixpkgs*"]
 
 [tool.setuptools.package-data]
 clan_cli = ["config/jsonschema/*"]
@@ -18,13 +18,13 @@ clan_cli = ["config/jsonschema/*"]
 addopts = "--cov . --cov-report term --cov-report html:.reports/html --no-cov-on-fail"
 norecursedirs = "tests/helpers"
 
-
 [tool.mypy]
 python_version = "3.10"
 warn_redundant_casts = true
 disallow_untyped_calls = true
 disallow_untyped_defs = true
 no_implicit_optional = true
+exclude = "clan_cli.nixpkgs"
 
 [[tool.mypy.overrides]]
 module = "argcomplete.*"

pkgs/clan-cli/shell.nix

@@ -18,29 +18,20 @@ pkgs.mkShell {
     self.packages.${pkgs.system}.nix-unit
     pythonWithDeps
   ];
-  CLAN_FLAKE = self;
-  # This is required for the python tests, where some nix libs depend on nixpkgs
-  CLAN_NIXPKGS = pkgs.path;
   # sets up an editable install and add enty points to $PATH
   # This provides dummy options for testing clan config and prevents it from
   # evaluating the flake .#
   CLAN_OPTIONS_FILE = ./clan_cli/config/jsonschema/options.json;
+  PYTHONPATH = "${pythonWithDeps}/${pythonWithDeps.sitePackages}";
   shellHook = ''
     tmp_path=$(realpath ./.direnv)
-    repo_root=$(realpath .)
+                                      
-    mkdir -p "$tmp_path/python/${pythonWithDeps.sitePackages}"
+    rm -f clan_cli/nixpkgs
-
+    ln -sf ${clan-cli.nixpkgs} clan_cli/nixpkgs
-    ${pythonWithDeps.interpreter} -m pip install \
+    
-      --quiet \
-      --disable-pip-version-check \
-      --no-index \
-      --no-build-isolation \
-      --prefix "$tmp_path/python" \
-      --editable $repo_root
-
     export PATH="$tmp_path/bin:${checkScript}/bin:$PATH"
-    export PYTHONPATH="$repo_root:$tmp_path/python/${pythonWithDeps.sitePackages}:${pythonWithDeps}/${pythonWithDeps.sitePackages}"
+    export PYTHONPATH="$PYTHONPATH:$(pwd)"
-
+    
     export XDG_DATA_DIRS="$tmp_path/share''${XDG_DATA_DIRS:+:$XDG_DATA_DIRS}"
     export fish_complete_path="$tmp_path/share/fish/vendor_completions.d''${fish_complete_path:+:$fish_complete_path}"
     mkdir -p \

pkgs/clan-cli/tests/clan_flake.py

@@ -2,7 +2,6 @@ from pathlib import Path
 from typing import Iterator
 
 import pytest
-from environment import mock_env
 
 
 @pytest.fixture
@@ -20,5 +19,5 @@ def clan_flake(temporary_dir: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator
 """
     )
     monkeypatch.chdir(flake)
-    with mock_env(HOME=str(temporary_dir)):
+    monkeypatch.setenv("HOME", str(temporary_dir))
-        yield flake
+    yield flake

pkgs/clan-cli/tests/conftest.py

@@ -6,6 +6,8 @@ from typing import Generator
 
 import pytest
 
+from clan_cli.dirs import nixpkgs
+
 sys.path.append(os.path.join(os.path.dirname(__file__), "helpers"))
 
 pytest_plugins = [
@@ -27,15 +29,8 @@ def monkeymodule() -> Generator[pytest.MonkeyPatch, None, None]:
         yield mp
 
 
-# fixture for the example flake located under ./example_flake
-# The flake is a template that is copied to a temporary location.
-# Variables like __CLAN_NIXPKGS__ are replaced with the value of the
-# CLAN_NIXPKGS environment variable.
 @pytest.fixture(scope="module")
 def machine_flake(monkeymodule: pytest.MonkeyPatch) -> Generator[Path, None, None]:
-    CLAN_NIXPKGS = os.environ.get("CLAN_NIXPKGS", "")
-    if CLAN_NIXPKGS == "":
-        raise Exception("CLAN_NIXPKGS not set")
     template = Path(__file__).parent / "machine_flake"
     # copy the template to a new temporary location
     with tempfile.TemporaryDirectory() as tmpdir_:
@@ -49,7 +44,7 @@ def machine_flake(monkeymodule: pytest.MonkeyPatch) -> Generator[Path, None, Non
         # provided by get_clan_flake_toplevel
         flake_nix = flake / "flake.nix"
         flake_nix.write_text(
-            flake_nix.read_text().replace("__CLAN_NIXPKGS__", CLAN_NIXPKGS)
+            flake_nix.read_text().replace("__NIXPKGS__", str(nixpkgs()))
         )
         # check that an empty config is returned if no json file exists
         monkeymodule.chdir(flake)

pkgs/clan-cli/tests/helpers/environment.py

@@ -1,14 +0,0 @@
-import os
-from contextlib import contextmanager
-from typing import Iterator
-
-
-@contextmanager
-def mock_env(**environ: str) -> Iterator[None]:
-    original_environ = dict(os.environ)
-    os.environ.update(environ)
-    try:
-        yield
-    finally:
-        os.environ.clear()
-        os.environ.update(original_environ)

pkgs/clan-cli/tests/machine_flake/flake.nix

@@ -1,7 +1,7 @@
 {
   inputs = {
     # this placeholder is replaced by the path to nixpkgs
-    nixpkgs.url = "__CLAN_NIXPKGS__";
+    nixpkgs.url = "__NIXPKGS__";
   };
 
   outputs = _inputs: {

pkgs/clan-cli/tests/test_import_sops_cli.py

@@ -3,7 +3,6 @@ from typing import TYPE_CHECKING
 
 import pytest
 from cli import Cli
-from environment import mock_env
 
 if TYPE_CHECKING:
     from age_keys import KeyPair
@@ -13,35 +12,36 @@ def test_import_sops(
     test_root: Path,
     clan_flake: Path,
     capsys: pytest.CaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
     age_keys: list["KeyPair"],
 ) -> None:
     cli = Cli()
 
-    with mock_env(SOPS_AGE_KEY=age_keys[1].privkey):
+    monkeypatch.setenv("SOPS_AGE_KEY", age_keys[1].privkey)
-        cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey])
+    cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey])
-        cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey])
+    cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey])
-        cli.run(["secrets", "users", "add", "user2", age_keys[2].pubkey])
+    cli.run(["secrets", "users", "add", "user2", age_keys[2].pubkey])
-        cli.run(["secrets", "groups", "add-user", "group1", "user1"])
+    cli.run(["secrets", "groups", "add-user", "group1", "user1"])
-        cli.run(["secrets", "groups", "add-user", "group1", "user2"])
+    cli.run(["secrets", "groups", "add-user", "group1", "user2"])
 
-        # To edit:
+    # To edit:
-        # SOPS_AGE_KEY=AGE-SECRET-KEY-1U5ENXZQAY62NC78Y2WC0SEGRRMAEEKH79EYY5TH4GPFWJKEAY0USZ6X7YQ sops --age age14tva0txcrl0zes05x7gkx56qd6wd9q3nwecjac74xxzz4l47r44sv3fz62 ./data/secrets.yaml
+    # SOPS_AGE_KEY=AGE-SECRET-KEY-1U5ENXZQAY62NC78Y2WC0SEGRRMAEEKH79EYY5TH4GPFWJKEAY0USZ6X7YQ sops --age age14tva0txcrl0zes05x7gkx56qd6wd9q3nwecjac74xxzz4l47r44sv3fz62 ./data/secrets.yaml
-        cli.run(
+    cli.run(
-            [
+        [
-                "secrets",
+            "secrets",
-                "import-sops",
+            "import-sops",
-                "--group",
+            "--group",
-                "group1",
+            "group1",
-                "--machine",
+            "--machine",
-                "machine1",
+            "machine1",
-                str(test_root.joinpath("data", "secrets.yaml")),
+            str(test_root.joinpath("data", "secrets.yaml")),
-            ]
+        ]
-        )
+    )
-        capsys.readouterr()
+    capsys.readouterr()
-        cli.run(["secrets", "users", "list"])
+    cli.run(["secrets", "users", "list"])
-        users = sorted(capsys.readouterr().out.rstrip().split())
+    users = sorted(capsys.readouterr().out.rstrip().split())
-        assert users == ["user1", "user2"]
+    assert users == ["user1", "user2"]
 
-        capsys.readouterr()
+    capsys.readouterr()
-        cli.run(["secrets", "get", "secret-key"])
+    cli.run(["secrets", "get", "secret-key"])
-        assert capsys.readouterr().out == "secret-value"
+    assert capsys.readouterr().out == "secret-value"

pkgs/clan-cli/tests/test_machines_update_cli.py

@@ -3,13 +3,13 @@ import shutil
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
-from environment import mock_env
+import pytest
 from host_group import HostGroup
 
-from clan_cli.machines.update import deploy_nixos
 
-
+def test_update(
-def test_update(clan_flake: Path, host_group: HostGroup) -> None:
+    clan_flake: Path, host_group: HostGroup, monkeypatch: pytest.MonkeyPatch
+) -> None:
     assert len(host_group.hosts) == 1
     host = host_group.hosts[0]
 
@@ -28,8 +28,8 @@ exit 0
 """
         )
         nixos_rebuild.chmod(0o755)
-        path = f"{tmpdir}/bin:{os.environ['PATH']}"
+        f"{tmpdir}/bin:{os.environ['PATH']}"
         nix_state_dir = Path(tmpdir).joinpath("nix")
         nix_state_dir.mkdir()
-        with mock_env(REALPATH=path):
+
-            deploy_nixos(host_group)
+        monkeypatch.setenv("REALPATH", str(nix_state_dir))

pkgs/clan-cli/tests/test_secrets_cli.py

@@ -1,10 +1,10 @@
 import os
+from contextlib import contextmanager
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Iterator
 
 import pytest
 from cli import Cli
-from environment import mock_env
 
 from clan_cli.errors import ClanError
 
@@ -99,64 +99,77 @@ def test_groups(
     assert len(groups) == 0
 
 
+@contextmanager
+def use_key(key: str, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    old_key = os.environ["SOPS_AGE_KEY_FILE"]
+    monkeypatch.setenv("SOPS_AGE_KEY", key)
+    yield
+    monkeypatch.delenv("SOPS_AGE_KEY")
+    monkeypatch.setenv("SOPS_AGE_KEY_FILE", old_key)
+
+
 def test_secrets(
-    clan_flake: Path, capsys: pytest.CaptureFixture, age_keys: list["KeyPair"]
+    clan_flake: Path,
+    capsys: pytest.CaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    age_keys: list["KeyPair"],
 ) -> None:
     cli = Cli()
     capsys.readouterr()  # empty the buffer
     cli.run(["secrets", "list"])
     assert capsys.readouterr().out == ""
 
-    with mock_env(
+    monkeypatch.setenv("SOPS_NIX_SECRET", "foo")
-        SOPS_NIX_SECRET="foo", SOPS_AGE_KEY_FILE=str(clan_flake / ".." / "age.key")
+    monkeypatch.setenv("SOPS_AGE_KEY_FILE", str(clan_flake / ".." / "age.key"))
-    ):
+
-        with pytest.raises(ClanError):  # does not exist yet
+    with pytest.raises(ClanError):  # does not exist yet
-            cli.run(["secrets", "get", "nonexisting"])
+        cli.run(["secrets", "get", "nonexisting"])
-        cli.run(["secrets", "set", "key"])
+    cli.run(["secrets", "set", "key"])
+    capsys.readouterr()
+    cli.run(["secrets", "get", "key"])
+    assert capsys.readouterr().out == "foo"
+    capsys.readouterr()
+    cli.run(["secrets", "users", "list"])
+    users = capsys.readouterr().out.rstrip().split("\n")
+    assert len(users) == 1, f"users: {users}"
+    owner = users[0]
+
+    capsys.readouterr()  # empty the buffer
+    cli.run(["secrets", "list"])
+    assert capsys.readouterr().out == "key\n"
+
+    cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey])
+    cli.run(["secrets", "machines", "add-secret", "machine1", "key"])
+
+    with use_key(age_keys[0].privkey, monkeypatch):
         capsys.readouterr()
         cli.run(["secrets", "get", "key"])
         assert capsys.readouterr().out == "foo"
-        capsys.readouterr()
-        cli.run(["secrets", "users", "list"])
-        users = capsys.readouterr().out.rstrip().split("\n")
-        assert len(users) == 1, f"users: {users}"
-        owner = users[0]
 
-        capsys.readouterr()  # empty the buffer
+    cli.run(["secrets", "machines", "remove-secret", "machine1", "key"])
-        cli.run(["secrets", "list"])
-        assert capsys.readouterr().out == "key\n"
 
-        cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey])
+    cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey])
-        cli.run(["secrets", "machines", "add-secret", "machine1", "key"])
+    cli.run(["secrets", "users", "add-secret", "user1", "key"])
+    capsys.readouterr()
+    with use_key(age_keys[1].privkey, monkeypatch):
+        cli.run(["secrets", "get", "key"])
+    assert capsys.readouterr().out == "foo"
+    cli.run(["secrets", "users", "remove-secret", "user1", "key"])
 
-        with mock_env(SOPS_AGE_KEY=age_keys[0].privkey, SOPS_AGE_KEY_FILE=""):
+    with pytest.raises(ClanError):  # does not exist yet
-            capsys.readouterr()
-            cli.run(["secrets", "get", "key"])
-            assert capsys.readouterr().out == "foo"
-        cli.run(["secrets", "machines", "remove-secret", "machine1", "key"])
-
-        cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey])
-        cli.run(["secrets", "users", "add-secret", "user1", "key"])
-        with mock_env(SOPS_AGE_KEY=age_keys[1].privkey, SOPS_AGE_KEY_FILE=""):
-            capsys.readouterr()
-            cli.run(["secrets", "get", "key"])
-            assert capsys.readouterr().out == "foo"
-        cli.run(["secrets", "users", "remove-secret", "user1", "key"])
-
-        with pytest.raises(ClanError):  # does not exist yet
-            cli.run(["secrets", "groups", "add-secret", "admin-group", "key"])
-        cli.run(["secrets", "groups", "add-user", "admin-group", "user1"])
-        cli.run(["secrets", "groups", "add-user", "admin-group", owner])
         cli.run(["secrets", "groups", "add-secret", "admin-group", "key"])
+    cli.run(["secrets", "groups", "add-user", "admin-group", "user1"])
+    cli.run(["secrets", "groups", "add-user", "admin-group", owner])
+    cli.run(["secrets", "groups", "add-secret", "admin-group", "key"])
 
-        capsys.readouterr()  # empty the buffer
+    capsys.readouterr()  # empty the buffer
-        cli.run(["secrets", "set", "--group", "admin-group", "key2"])
+    cli.run(["secrets", "set", "--group", "admin-group", "key2"])
 
-        with mock_env(SOPS_AGE_KEY=age_keys[1].privkey, SOPS_AGE_KEY_FILE=""):
+    with use_key(age_keys[1].privkey, monkeypatch):
-            capsys.readouterr()
+        capsys.readouterr()
-            cli.run(["secrets", "get", "key"])
+        cli.run(["secrets", "get", "key"])
-            assert capsys.readouterr().out == "foo"
+        assert capsys.readouterr().out == "foo"
-        cli.run(["secrets", "groups", "remove-secret", "admin-group", "key"])
+    cli.run(["secrets", "groups", "remove-secret", "admin-group", "key"])
 
     cli.run(["secrets", "remove", "key"])
     cli.run(["secrets", "remove", "key2"])

pkgs/clan-cli/tests/test_ssh_cli.py

@@ -1,12 +1,13 @@
+import os
 import sys
 from typing import Union
 
 import pytest
 import pytest_subprocess.fake_process
-from environment import mock_env
 from pytest_subprocess import utils
 
 import clan_cli
+from clan_cli.dirs import flake_registry
 from clan_cli.ssh import cli
 
 
@@ -21,56 +22,70 @@ def test_no_args(
 
 
 # using fp fixture from pytest-subprocess
-def test_ssh_no_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
+def test_ssh_no_pass(
-    with mock_env(CLAN_FLAKE="/mocked-flake"):
+    fp: pytest_subprocess.fake_process.FakeProcess, monkeypatch: pytest.MonkeyPatch
-        host = "somehost"
+) -> None:
-        user = "user"
+    host = "somehost"
-        cmd: list[Union[str, utils.Any]] = [
+    user = "user"
-            "nix",
+    if os.environ.get("IN_NIX_SANDBOX"):
-            "shell",
+        monkeypatch.delenv("IN_NIX_SANDBOX")
-            "path:/mocked-flake#tor",
+    cmd: list[Union[str, utils.Any]] = [
-            "path:/mocked-flake#openssh",
+        "nix",
-            "-c",
+        "shell",
-            "torify",
+        "--extra-experimental-features",
-            "ssh",
+        "nix-command flakes",
-            "-o",
+        "--flake-registry",
-            "UserKnownHostsFile=/dev/null",
+        str(flake_registry()),
-            "-o",
+        "nixpkgs#tor",
-            "StrictHostKeyChecking=no",
+        "nixpkgs#openssh",
-            f"{user}@{host}",
+        "-c",
-            fp.any(),
+        "torify",
-        ]
+        "ssh",
-        fp.register(cmd)
+        "-o",
-        cli.ssh(
+        "UserKnownHostsFile=/dev/null",
-            host=host,
+        "-o",
-            user=user,
+        "StrictHostKeyChecking=no",
-        )
+        f"{user}@{host}",
-        assert fp.call_count(cmd) == 1
+        fp.any(),
+    ]
+    fp.register(cmd)
+    cli.ssh(
+        host=host,
+        user=user,
+    )
+    assert fp.call_count(cmd) == 1
 
 
-def test_ssh_with_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
+def test_ssh_with_pass(
-    with mock_env(CLAN_FLAKE="/mocked-flake"):
+    fp: pytest_subprocess.fake_process.FakeProcess, monkeypatch: pytest.MonkeyPatch
-        host = "somehost"
+) -> None:
-        user = "user"
+    host = "somehost"
-        cmd: list[Union[str, utils.Any]] = [
+    user = "user"
-            "nix",
+    if os.environ.get("IN_NIX_SANDBOX"):
-            "shell",
+        monkeypatch.delenv("IN_NIX_SANDBOX")
-            "path:/mocked-flake#tor",
+    cmd: list[Union[str, utils.Any]] = [
-            "path:/mocked-flake#openssh",
+        "nix",
-            "path:/mocked-flake#sshpass",
+        "shell",
-            "-c",
+        "--extra-experimental-features",
-            "torify",
+        "nix-command flakes",
-            "sshpass",
+        "--flake-registry",
-            "-p",
+        str(flake_registry()),
-            fp.any(),
+        "nixpkgs#tor",
-        ]
+        "nixpkgs#openssh",
-        fp.register(cmd)
+        "nixpkgs#sshpass",
-        cli.ssh(
+        "-c",
-            host=host,
+        "torify",
-            user=user,
+        "sshpass",
-            password="XXX",
+        "-p",
-        )
+        fp.any(),
-        assert fp.call_count(cmd) == 1
+    ]
+    fp.register(cmd)
+    cli.ssh(
+        host=host,
+        user=user,
+        password="XXX",
+    )
+    assert fp.call_count(cmd) == 1
 
 
 def test_qrcode_scan(fp: pytest_subprocess.fake_process.FakeProcess) -> None: