import json import sys import textwrap import uuid from pathlib import Path from subprocess import PIPE, CompletedProcess from typing import Any from unittest.mock import ANY, Mock, call, patch import pytest from pytest import MonkeyPatch import nixos_rebuild.elevate as e import nixos_rebuild.models as m import nixos_rebuild.nix as n import nixos_rebuild.process as p from .helpers import get_qualified_name SUDO = e.SudoElevator() @patch( get_qualified_name(n.run_wrapper, n), autospec=True, return_value=CompletedProcess([], 0, stdout=" \n/path/to/file\n "), ) def test_build(mock_run: Mock) -> None: assert n.build( "config.system.build.attr", m.BuildAttr("", None), {"nix_flag": "foo"}, ) == Path("/path/to/file") mock_run.assert_called_with( [ "nix-build", "", "--attr", "config.system.build.attr", "--nix-flag", "foo", ], stdout=PIPE, ) assert n.build( "config.system.build.attr", m.BuildAttr(Path("file"), "preAttr") ) == Path("/path/to/file") mock_run.assert_called_with( ["nix-build", Path("file"), "--attr", "preAttr.config.system.build.attr"], stdout=PIPE, ) @patch( get_qualified_name(n.run_wrapper, n), autospec=True, return_value=CompletedProcess([], 0, stdout=" \n/path/to/file\n "), ) def test_build_flake(mock_run: Mock, monkeypatch: MonkeyPatch, tmpdir: Path) -> None: monkeypatch.chdir(tmpdir) flake = m.Flake.parse("/flake.nix#hostname") assert n.build_flake( "config.system.build.toplevel", flake, {"no_link": True, "nix_flag": "foo"}, ) == Path("/path/to/file") mock_run.assert_called_with( [ "nix", "--extra-experimental-features", "nix-command flakes", "build", "--print-out-paths", '/flake.nix#nixosConfigurations."hostname".config.system.build.toplevel', "--no-link", "--nix-flag", "foo", ], stdout=PIPE, ) @patch(get_qualified_name(n.run_wrapper, n), autospec=True) @patch("uuid.uuid4", autospec=True) def test_build_remote( mock_uuid4: Mock, mock_run: Mock, monkeypatch: MonkeyPatch ) -> None: build_host = m.Remote("user@host", [], "ssh") monkeypatch.setenv("NIX_SSHOPTS", "--ssh opts") def run_wrapper_side_effect( args: list[str], **kwargs: Any ) -> CompletedProcess[str]: if args[0] == "nix-instantiate": return CompletedProcess([], 0, stdout=" \n/path/to/file\n ") elif args[0] == "mktemp": return CompletedProcess([], 0, stdout=" \n/tmp/tmpdir\n ") elif args[0] == "nix-store": return CompletedProcess([], 0, stdout=" \n/tmp/tmpdir/config\n ") elif args[0] == "readlink": return CompletedProcess([], 0, stdout=" \n/path/to/config\n ") else: return CompletedProcess([], 0) mock_run.side_effect = run_wrapper_side_effect mock_uuid4.side_effect = [uuid.UUID(int=1), uuid.UUID(int=2)] assert n.build_remote( "config.system.build.toplevel", m.BuildAttr("", "preAttr"), build_host, realise_flags={"realise": True}, instantiate_flags={"inst": True}, copy_flags={"copy": True}, ) == Path("/path/to/config") mock_run.assert_has_calls( [ call( [ "nix-instantiate", "", "--attr", "preAttr.config.system.build.toplevel", "--add-root", n.tmpdir.TMPDIR_PATH / "00000000000000000000000000000001", "--inst", ], stdout=PIPE, ), call( [ "nix-copy-closure", "--copy", "--to", "user@host", Path("/path/to/file"), ], append_local_env={ "NIX_SSHOPTS": " ".join(["--ssh opts", *p.SSH_DEFAULT_OPTS]), }, ), call( ["mktemp", "-d", "-t", "nixos-rebuild.XXXXX"], remote=build_host, stdout=PIPE, ), call( [ "nix-store", "--realise", Path("/path/to/file"), "--add-root", Path("/tmp/tmpdir/00000000000000000000000000000002"), "--realise", ], remote=build_host, stdout=PIPE, ), call( ["readlink", "-f", "/tmp/tmpdir/config"], remote=build_host, stdout=PIPE, ), call(["rm", "-rf", Path("/tmp/tmpdir")], remote=build_host, check=False), ] ) @patch( get_qualified_name(n.run_wrapper, n), autospec=True, return_value=CompletedProcess([], 0, stdout=" \n/path/to/file\n "), ) def test_build_remote_flake( mock_run: Mock, monkeypatch: MonkeyPatch, tmpdir: Path, ) -> None: monkeypatch.chdir(tmpdir) flake = m.Flake.parse("/flake.nix#hostname") build_host = m.Remote("user@host", [], "ssh") monkeypatch.setenv("NIX_SSHOPTS", "--ssh opts") assert n.build_remote_flake( "config.system.build.toplevel", flake, build_host, eval_flags={"flake": True}, copy_flags={"copy": True}, flake_build_flags={"build": True}, ) == Path("/path/to/file") mock_run.assert_has_calls( [ call( [ "nix", "--extra-experimental-features", "nix-command flakes", "eval", "--raw", '/flake.nix#nixosConfigurations."hostname".config.system.build.toplevel.drvPath', "--flake", ], stdout=PIPE, ), call( [ "nix-copy-closure", "--copy", "--to", "user@host", Path("/path/to/file"), ], append_local_env={ "NIX_SSHOPTS": " ".join(["--ssh opts", *p.SSH_DEFAULT_OPTS]), }, ), call( [ "nix", "--extra-experimental-features", "nix-command flakes", "build", "/path/to/file^*", "--print-out-paths", "--build", ], remote=build_host, stdout=PIPE, ), ] ) def test_copy_closure(monkeypatch: MonkeyPatch) -> None: closure = Path("/path/to/closure") with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run: n.copy_closure(closure, None) mock_run.assert_not_called() target_host = m.Remote("user@target.host", [], "ssh") build_host = m.Remote("user@build.host", [], "ssh") target_host_ng = m.Remote("user@target.host", [], "ssh-ng") with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run: n.copy_closure(closure, target_host) mock_run.assert_called_with( ["nix-copy-closure", "--to", "user@target.host", closure], append_local_env={"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS)}, ) with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run: n.copy_closure(closure, target_host_ng) mock_run.assert_called_with( [ "nix", "--extra-experimental-features", "nix-command flakes", "copy", "--to", "ssh-ng://user@target.host", closure, ], append_local_env={"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS)}, ) monkeypatch.setenv("NIX_SSHOPTS", "--ssh build-opt") with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run: n.copy_closure(closure, None, build_host, {"copy_flag": True}) mock_run.assert_called_with( ["nix-copy-closure", "--copy-flag", "--from", "user@build.host", closure], append_local_env={ "NIX_SSHOPTS": " ".join(["--ssh build-opt", *p.SSH_DEFAULT_OPTS]) }, ) # NIXOS_REBUILD_SSH_DEFAULT_OPTS replaces the ControlMaster defaults monkeypatch.setenv("NIX_SSHOPTS", "-oControlPath=/run/user/1000/%C") monkeypatch.setenv("NIXOS_REBUILD_SSH_DEFAULT_OPTS", "") with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run: n.copy_closure(closure, target_host) mock_run.assert_called_with( ["nix-copy-closure", "--to", "user@target.host", closure], append_local_env={"NIX_SSHOPTS": "-oControlPath=/run/user/1000/%C"}, ) monkeypatch.delenv("NIXOS_REBUILD_SSH_DEFAULT_OPTS") monkeypatch.setenv("NIX_SSHOPTS", "--ssh build-target-opt") env = {"NIX_SSHOPTS": " ".join(["--ssh build-target-opt", *p.SSH_DEFAULT_OPTS])} with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run: n.copy_closure(closure, target_host, build_host, {"copy_flag": True}) mock_run.assert_called_with( [ "nix", "--extra-experimental-features", "nix-command flakes", "copy", "--copy-flag", "--from", "ssh://user@build.host", "--to", "ssh://user@target.host", closure, ], append_local_env=env, ) @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_edit(mock_run: Mock, monkeypatch: MonkeyPatch, tmpdir: Path) -> None: with monkeypatch.context() as mp: default_nix = tmpdir / "default.nix" default_nix.write_text("{}", encoding="utf-8") mp.setenv("NIXOS_CONFIG", str(tmpdir)) mp.setenv("EDITOR", "editor") n.edit() mock_run.assert_called_with(["editor", default_nix], check=False) @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_edit_flake(mock_run: Mock) -> None: flake = m.Flake.parse("/flake.nix#attr") n.edit_flake(flake, {"commit_lock_file": True}) mock_run.assert_called_with( [ "nix", "--extra-experimental-features", "nix-command flakes", "edit", "--commit-lock-file", "--", '/flake.nix#nixosConfigurations."attr"', ], check=False, ) @patch( get_qualified_name(n.run_wrapper, n), autospec=True, return_value=CompletedProcess( [], 0, """ { "azure": "nixos-image-azure-25.05.20250102.6df2492-x86_64-linux.vhd", "vmware": "nixos-image-vmware-25.05.20250102.6df2492-x86_64-linux.vmdk" } """, ), ) def test_get_build_image_variants(mock_run: Mock, tmp_path: Path) -> None: build_attr = m.BuildAttr("", None) assert n.get_build_image_variants(build_attr) == { "azure": "nixos-image-azure-25.05.20250102.6df2492-x86_64-linux.vhd", "vmware": "nixos-image-vmware-25.05.20250102.6df2492-x86_64-linux.vmdk", } mock_run.assert_called_with( [ "nix-instantiate", "--eval", "--strict", "--json", "--expr", textwrap.dedent(""" let value = import ; set = if builtins.isFunction value then value {} else value; in builtins.attrNames set.config.system.build.images """), ], stdout=PIPE, ) build_attr = m.BuildAttr(tmp_path, "preAttr") assert n.get_build_image_variants(build_attr, {"inst_flag": True}) == { "azure": "nixos-image-azure-25.05.20250102.6df2492-x86_64-linux.vhd", "vmware": "nixos-image-vmware-25.05.20250102.6df2492-x86_64-linux.vmdk", } mock_run.assert_called_with( [ "nix-instantiate", "--eval", "--strict", "--json", "--expr", textwrap.dedent(f""" let value = import "{tmp_path}"; set = if builtins.isFunction value then value {{}} else value; in builtins.attrNames set.preAttr.config.system.build.images """), "--inst-flag", ], stdout=PIPE, ) @patch( get_qualified_name(n.run_wrapper, n), autospec=True, return_value=CompletedProcess( [], 0, """ { "azure": "nixos-image-azure-25.05.20250102.6df2492-x86_64-linux.vhd", "vmware": "nixos-image-vmware-25.05.20250102.6df2492-x86_64-linux.vmdk" } """, ), ) def test_get_build_image_variants_flake(mock_run: Mock) -> None: flake = m.Flake("/flake.nix", "myAttr") assert n.get_build_image_variants_flake(flake, {"eval_flag": True}) == { "azure": "nixos-image-azure-25.05.20250102.6df2492-x86_64-linux.vhd", "vmware": "nixos-image-vmware-25.05.20250102.6df2492-x86_64-linux.vmdk", } mock_run.assert_called_with( [ "nix", "--extra-experimental-features", "nix-command flakes", "eval", "--json", "/flake.nix#myAttr.config.system.build.images", "--apply", "builtins.attrNames", "--eval-flag", ], stdout=PIPE, ) def test_get_nixpkgs_rev(tmpdir: Path) -> None: assert n.get_nixpkgs_rev(None) is None assert n.get_nixpkgs_rev(tmpdir) is None (tmpdir / ".git").mkdir() expected_calls = [ call( ["git", "-C", tmpdir, "rev-parse", "--short", "HEAD"], check=False, capture_output=True, ), call( ["git", "-C", tmpdir, "diff", "--quiet"], check=False, ), ] with patch( get_qualified_name(n.run_wrapper, n), autospec=True, side_effect=[ CompletedProcess([], 0, "0f7c82403fd6"), CompletedProcess([], returncode=0), ], ) as mock_run: assert n.get_nixpkgs_rev(tmpdir) == ".git.0f7c82403fd6" mock_run.assert_has_calls(expected_calls) with patch( get_qualified_name(n.run_wrapper, n), autospec=True, side_effect=[ CompletedProcess([], 0, "0f7c82403fd6"), CompletedProcess([], returncode=1), ], ) as mock_run: assert n.get_nixpkgs_rev(tmpdir) == ".git.0f7c82403fd6M" mock_run.assert_has_calls(expected_calls) def test_get_generations(tmp_path: Path) -> None: nixos_path = tmp_path / "nixos-system" nixos_path.mkdir() (tmp_path / "system").symlink_to(tmp_path / "system-2-link") # In the "wrong" order on purpose to make sure we are sorting the results (tmp_path / "system-1-link").symlink_to(nixos_path) (tmp_path / "system-3-link").symlink_to(nixos_path) (tmp_path / "system-2-link").symlink_to(nixos_path) # An alternate profile; this shouldn't appear. (tmp_path / "custom").symlink_to(tmp_path / "custom-1-link") (tmp_path / "custom-1-link").symlink_to(nixos_path) assert n.get_generations(m.Profile("system", tmp_path / "system")) == [ m.Generation(id=1, current=False, timestamp=ANY), m.Generation(id=2, current=True, timestamp=ANY), m.Generation(id=3, current=False, timestamp=ANY), ] def test_get_generations_with_profile(tmp_path: Path) -> None: nixos_path = tmp_path / "nixos-system" nixos_path.mkdir() (tmp_path / "custom").symlink_to(tmp_path / "custom-2-link") # In the "wrong" order on purpose to make sure we are sorting the results (tmp_path / "custom-1-link").symlink_to(nixos_path) (tmp_path / "custom-3-link").symlink_to(nixos_path) (tmp_path / "custom-2-link").symlink_to(nixos_path) # An alternate profile; none of these should appear. (tmp_path / "system").symlink_to(tmp_path / "system-1-link") (tmp_path / "system-1-link").symlink_to(nixos_path) assert n.get_generations(m.Profile("custom", tmp_path / "custom")) == [ m.Generation(id=1, current=False, timestamp=ANY), m.Generation(id=2, current=True, timestamp=ANY), m.Generation(id=3, current=False, timestamp=ANY), ] def test_get_generations_from_nix_env(tmp_path: Path) -> None: path = tmp_path / "test" path.touch() return_value = CompletedProcess( [], 0, stdout=textwrap.dedent("""\ 2082 2024-11-07 22:58:56 2083 2024-11-07 22:59:41 2084 2024-11-07 23:54:17 (current) """), ) with patch( get_qualified_name(n.run_wrapper, n), autospec=True, return_value=return_value ) as mock_run: assert n.get_generations_from_nix_env(m.Profile("system", path)) == [ m.Generation(id=2082, current=False, timestamp="2024-11-07 22:58:56"), m.Generation(id=2083, current=False, timestamp="2024-11-07 22:59:41"), m.Generation(id=2084, current=True, timestamp="2024-11-07 23:54:17"), ] mock_run.assert_called_with( ["nix-env", "-p", path, "--list-generations"], stdout=PIPE, remote=None, elevate=e.NO_ELEVATOR, ) remote = m.Remote("user@host", [], "ssh") with patch( get_qualified_name(n.run_wrapper, n), autospec=True, return_value=return_value ) as mock_run: assert n.get_generations_from_nix_env( m.Profile("system", path), remote, SUDO ) == [ m.Generation(id=2082, current=False, timestamp="2024-11-07 22:58:56"), m.Generation(id=2083, current=False, timestamp="2024-11-07 22:59:41"), m.Generation(id=2084, current=True, timestamp="2024-11-07 23:54:17"), ] mock_run.assert_called_with( ["nix-env", "-p", path, "--list-generations"], stdout=PIPE, remote=remote, elevate=SUDO, ) @patch( get_qualified_name(n.get_generations), autospec=True, return_value=[ m.Generation( id=1, timestamp="2024-11-07 23:54:17", current=False, ), m.Generation( id=2, timestamp="2024-11-07 23:54:17", current=True, ), ], ) def test_list_generations(mock_get_generations: Mock, tmp_path: Path) -> None: # Probably better to test this function in a real system, this test is # mostly to make sure it doesn't break horribly assert n.list_generations(m.Profile("system", tmp_path)) == [ { "configurationRevision": "Unknown", "current": True, "date": "2024-11-07 23:54:17", "generation": 2, "kernelVersion": "Unknown", "nixosVersion": "Unknown", "specialisations": [], }, { "configurationRevision": "Unknown", "current": False, "date": "2024-11-07 23:54:17", "generation": 1, "kernelVersion": "Unknown", "nixosVersion": "Unknown", "specialisations": [], }, ] @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_diff_closures(mock_run: Mock) -> None: n.diff_closures( Path("/run/current-system"), Path("/nix/var/nix/profiles/system"), None ) mock_run.assert_called_with( [ "nix", "--extra-experimental-features", "nix-command flakes", "store", "diff-closures", Path("/run/current-system"), Path("/nix/var/nix/profiles/system"), ], remote=None, stdout=sys.stderr, ) @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_repl(mock_run: Mock) -> None: n.repl(m.BuildAttr("", None), {"nix_flag": True}) mock_run.assert_called_with( ["nix", "repl", "--file", "", "--nix-flag"] ) n.repl(m.BuildAttr(Path("file.nix"), "myAttr")) mock_run.assert_called_with(["nix", "repl", "--file", Path("file.nix"), "myAttr"]) @patch( get_qualified_name(n.run_wrapper, n), autospec=True, return_value=CompletedProcess( [], 0, stdout=json.dumps({"resolvedUrl": "path:/flake.nix"}) ), ) def test_repl_flake(mock_run: Mock) -> None: n.repl_flake(m.Flake("flake.nix", "myAttr"), {"nix_flag": True}) # See nixos-rebuild-ng.tests.repl for a better test, # this is mostly for sanity check assert mock_run.call_count == 2 @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_rollback(mock_run: Mock, tmp_path: Path) -> None: path = tmp_path / "test" path.touch() profile = m.Profile("system", path) assert n.rollback(profile, None, e.NO_ELEVATOR) == profile.path mock_run.assert_called_with( ["nix-env", "--rollback", "-p", path], remote=None, elevate=e.NO_ELEVATOR, ) target_host = m.Remote("user@localhost", [], "ssh") assert n.rollback(profile, target_host, SUDO) == profile.path mock_run.assert_called_with( ["nix-env", "--rollback", "-p", path], remote=target_host, elevate=SUDO, ) def test_rollback_temporary_profile(tmp_path: Path) -> None: path = tmp_path / "test" path.touch() profile = m.Profile("system", path) with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run: mock_run.return_value = CompletedProcess( [], 0, stdout=textwrap.dedent("""\ 2082 2024-11-07 22:58:56 2083 2024-11-07 22:59:41 2084 2024-11-07 23:54:17 (current) """), ) assert ( n.rollback_temporary_profile(m.Profile("system", path), None, e.NO_ELEVATOR) == path.parent / "system-2083-link" ) mock_run.assert_called_with( [ "nix-env", "-p", path, "--list-generations", ], stdout=PIPE, remote=None, elevate=e.NO_ELEVATOR, ) target_host = m.Remote("user@localhost", [], "ssh") assert ( n.rollback_temporary_profile(m.Profile("foo", path), target_host, SUDO) == path.parent / "foo-2083-link" ) mock_run.assert_called_with( [ "nix-env", "-p", path, "--list-generations", ], stdout=PIPE, remote=target_host, elevate=SUDO, ) with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run: mock_run.return_value = CompletedProcess([], 0, stdout="") assert n.rollback_temporary_profile(profile, None, e.NO_ELEVATOR) is None @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_set_profile(mock_run: Mock) -> None: profile_path = Path("/path/to/profile") config_path = Path("/path/to/config") mock_run.return_value = CompletedProcess([], 0) n.set_profile( m.Profile("system", profile_path), config_path, target_host=None, elevate=e.NO_ELEVATOR, ) mock_run.assert_called_with( ["nix-env", "-p", profile_path, "--set", config_path], remote=None, elevate=e.NO_ELEVATOR, ) mock_run.return_value = CompletedProcess([], 1) with pytest.raises(m.NixOSRebuildError) as exc: n.set_profile( m.Profile("system", profile_path), config_path, target_host=None, elevate=e.NO_ELEVATOR, ) assert str(exc.value).startswith( "error: your NixOS configuration path seems to be missing essential files." ) @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_switch_to_configuration_without_systemd_run( mock_run: Any, monkeypatch: MonkeyPatch ) -> None: profile_path = Path("/path/to/profile") config_path = Path("/path/to/config") mock_run.return_value = CompletedProcess([], 1) with monkeypatch.context() as mp: mp.setenv("LOCALE_ARCHIVE", "") n.switch_to_configuration( profile_path, m.Action.SWITCH, elevate=e.NO_ELEVATOR, target_host=None, specialisation=None, install_bootloader=False, ) mock_run.assert_called_with( [profile_path / "bin/switch-to-configuration", "switch"], env={ "LOCALE_ARCHIVE": e.PRESERVE_ENV, "NIXOS_NO_CHECK": e.PRESERVE_ENV, "NIXOS_INSTALL_BOOTLOADER": "0", }, elevate=e.NO_ELEVATOR, remote=None, stdout=sys.stderr, ) with pytest.raises(m.NixOSRebuildError) as exc: n.switch_to_configuration( config_path, m.Action.BOOT, elevate=e.NO_ELEVATOR, target_host=None, specialisation="special", ) assert ( str(exc.value) == "error: '--specialisation' can only be used with 'switch' and 'test'" ) target_host = m.Remote("user@localhost", [], "ssh") with monkeypatch.context() as mp: mp.setenv("LOCALE_ARCHIVE", "/path/to/locale") mp.setenv("PATH", "/path/to/bin") mp.setattr(Path, Path.exists.__name__, lambda self: True) n.switch_to_configuration( Path("/path/to/config"), m.Action.TEST, elevate=SUDO, target_host=target_host, install_bootloader=True, specialisation="special", ) mock_run.assert_called_with( [ config_path / "specialisation/special/bin/switch-to-configuration", "test", ], env={ "LOCALE_ARCHIVE": e.PRESERVE_ENV, "NIXOS_NO_CHECK": e.PRESERVE_ENV, "NIXOS_INSTALL_BOOTLOADER": "1", }, elevate=SUDO, remote=target_host, stdout=sys.stderr, ) @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_switch_to_configuration_without_systemd_run_env_var( mock_run: Any, monkeypatch: MonkeyPatch ) -> None: profile_path = Path("/path/to/profile") mock_run.return_value = CompletedProcess([], 0) with monkeypatch.context() as mp: mp.setenv("LOCALE_ARCHIVE", "") mp.setenv("NIXOS_REBUILD_NO_SYSTEMD_RUN", "1") n.switch_to_configuration( profile_path, m.Action.SWITCH, elevate=e.NO_ELEVATOR, target_host=None, specialisation=None, install_bootloader=False, ) mock_run.assert_called_with( [profile_path / "bin/switch-to-configuration", "switch"], env={ "LOCALE_ARCHIVE": e.PRESERVE_ENV, "NIXOS_NO_CHECK": e.PRESERVE_ENV, "NIXOS_INSTALL_BOOTLOADER": "0", }, elevate=e.NO_ELEVATOR, remote=None, stdout=sys.stderr, ) @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_switch_to_configuration_with_systemd_run( mock_run: Mock, monkeypatch: MonkeyPatch ) -> None: profile_path = Path("/path/to/profile") config_path = Path("/path/to/config") mock_run.return_value = CompletedProcess([], 0) with monkeypatch.context() as mp: mp.setenv("LOCALE_ARCHIVE", "") n.switch_to_configuration( profile_path, m.Action.SWITCH, elevate=e.NO_ELEVATOR, target_host=None, specialisation=None, install_bootloader=False, ) mock_run.assert_called_with( [ *n.SWITCH_TO_CONFIGURATION_CMD_PREFIX, profile_path / "bin/switch-to-configuration", "switch", ], env={ "LOCALE_ARCHIVE": e.PRESERVE_ENV, "NIXOS_NO_CHECK": e.PRESERVE_ENV, "NIXOS_INSTALL_BOOTLOADER": "0", }, elevate=e.NO_ELEVATOR, remote=None, stdout=sys.stderr, ) target_host = m.Remote("user@localhost", [], "ssh") with monkeypatch.context() as mp: mp.setenv("LOCALE_ARCHIVE", "/path/to/locale") mp.setenv("PATH", "/path/to/bin") mp.setattr(Path, Path.exists.__name__, lambda self: True) n.switch_to_configuration( Path("/path/to/config"), m.Action.TEST, elevate=SUDO, target_host=target_host, install_bootloader=True, specialisation="special", ) mock_run.assert_called_with( [ *n.SWITCH_TO_CONFIGURATION_CMD_PREFIX, config_path / "specialisation/special/bin/switch-to-configuration", "test", ], env={ "LOCALE_ARCHIVE": e.PRESERVE_ENV, "NIXOS_NO_CHECK": e.PRESERVE_ENV, "NIXOS_INSTALL_BOOTLOADER": "1", }, elevate=SUDO, remote=target_host, stdout=sys.stderr, ) @patch("os.geteuid", autospec=True, return_value=1000) @patch(get_qualified_name(n.run_wrapper, n), autospec=True) def test_upgrade_channels(mock_run: Mock, mock_geteuid: Mock, tmpdir: Path) -> None: tmp_path = Path(tmpdir) with pytest.raises(m.NixOSRebuildError) as exc: n.upgrade_channels( all_channels=False, elevate=e.NO_ELEVATOR, channels_dir=tmp_path ) assert str(exc.value) == ( "error: if you pass the '--upgrade' or '--upgrade-all' flag, you must " "also pass '--elevate' or run the command as root" ) (tmp_path / "nixos").mkdir() (tmp_path / "nixos-hardware").mkdir() (tmp_path / "nixos-hardware" / ".update-on-nixos-rebuild").touch() (tmp_path / "home-manager").mkdir() # should work because we are passing an elevator even with os.geteuid == 1000 n.upgrade_channels(all_channels=False, elevate=SUDO, channels_dir=tmp_path) # Path.glob order is filesystem-dependent, so don't assert call order. mock_run.assert_has_calls( [ call( ["nix-channel", "--update", "nixos-hardware"], check=False, elevate=SUDO, ), call( ["nix-channel", "--update", "nixos"], check=False, elevate=SUDO, ), ], any_order=True, ) mock_run.reset_mock() # root check mock_geteuid.return_value = 0 n.upgrade_channels(all_channels=True, elevate=e.NO_ELEVATOR, channels_dir=tmp_path) mock_run.assert_has_calls( [ call( ["nix-channel", "--update", "home-manager"], check=False, elevate=e.NO_ELEVATOR, ), call( ["nix-channel", "--update", "nixos-hardware"], check=False, elevate=e.NO_ELEVATOR, ), call( ["nix-channel", "--update", "nixos"], check=False, elevate=e.NO_ELEVATOR, ), ], any_order=True, )