nixos-rebuild-ng: move more code to services.py (#420826)

Author: thiagokokada

pkgs/by-name/ni/nixos-rebuild-ng/package.nix

@@ -133,7 +133,8 @@ python3Packages.buildPythonApplication rec {
         };
 
         inherit (nixosTests)
-          nixos-rebuild-install-bootloader-ng
+          # FIXME: this test is disabled since it times out in @ofborg
+          # nixos-rebuild-install-bootloader-ng
           nixos-rebuild-specialisations-ng
           nixos-rebuild-target-host-ng
           ;

pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py

@@ -1,17 +1,15 @@
 import argparse
-import json
 import logging
 import os
 import sys
 from subprocess import CalledProcessError, run
 from typing import Final, assert_never
 
-from . import nix
+from . import nix, services
 from .constants import EXECUTABLE, WITH_NIX_2_18, WITH_REEXEC, WITH_SHELL_FILES
 from .models import Action, BuildAttr, Flake, Profile
 from .process import Remote
-from .services import build_and_activate_system, reexec
-from .utils import LogFormatter, tabulate
+from .utils import LogFormatter
 
 logger: Final = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
@@ -301,7 +299,7 @@ def execute(argv: list[str]) -> None:
         and not args.no_reexec
         and not os.environ.get("_NIXOS_REBUILD_REEXEC")
     ):
-        reexec(argv, args, build_flags, flake_build_flags)
+        services.reexec(argv, args, build_flags, flake_build_flags)
 
     profile = Profile.from_arg(args.profile_name)
     target_host = Remote.from_arg(args.target_host, args.ask_sudo_password)
@@ -310,10 +308,7 @@ def execute(argv: list[str]) -> None:
     flake = Flake.from_arg(args.flake, target_host)
 
     if can_run and not flake:
-        nixpkgs_path = nix.find_file("nixpkgs", build_flags)
-        rev = nix.get_nixpkgs_rev(nixpkgs_path)
-        if nixpkgs_path and rev:
-            (nixpkgs_path / ".version-suffix").write_text(rev)
+        services.write_version_suffix(build_flags)
 
     match action:
         case (
@@ -327,7 +322,7 @@ def execute(argv: list[str]) -> None:
             | Action.BUILD_VM
             | Action.BUILD_VM_WITH_BOOTLOADER
         ):
-            build_and_activate_system(
+            services.build_and_activate_system(
                 action=action,
                 args=args,
                 build_host=build_host,
@@ -343,32 +338,21 @@ def execute(argv: list[str]) -> None:
             )
 
         case Action.EDIT:
-            nix.edit(flake, flake_build_flags)
+            services.edit(flake=flake, flake_build_flags=flake_build_flags)
 
         case Action.DRY_RUN:
             raise AssertionError("DRY_RUN should be a DRY_BUILD alias")
 
         case Action.LIST_GENERATIONS:
-            generations = nix.list_generations(profile)
-            if args.json:
-                print(json.dumps(generations, indent=2))
-            else:
-                headers = {
-                    "generation": "Generation",
-                    "date": "Build-date",
-                    "nixosVersion": "NixOS version",
-                    "kernelVersion": "Kernel",
-                    "configurationRevision": "Configuration Revision",
-                    "specialisations": "Specialisation",
-                    "current": "Current",
-                }
-                print(tabulate(generations, headers=headers))
+            services.list_generations(args=args, profile=profile)
 
         case Action.REPL:
-            if flake:
-                nix.repl_flake(flake, flake_build_flags)
-            else:
-                nix.repl(build_attr, build_flags)
+            services.repl(
+                flake=flake,
+                build_attr=build_attr,
+                flake_build_flags=flake_build_flags,
+                build_flags=build_flags,
+            )
 
         case _:
             assert_never(action)

pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/nix.py

@@ -242,33 +242,33 @@ def nix_copy(to_host: Remote, from_host: Remote) -> None:
                 nix_copy_closure(to_host, to=True)
 
 
-def edit(flake: Flake | None, flake_flags: Args | None = None) -> None:
+def edit() -> None:
     "Try to find and open NixOS configuration file in editor."
-    if flake:
-        run_wrapper(
-            [
-                "nix",
-                *FLAKE_FLAGS,
-                "edit",
-                *dict_to_flags(flake_flags),
-                "--",
-                str(flake),
-            ],
-            check=False,
-        )
+    nixos_config = Path(
+        os.getenv("NIXOS_CONFIG") or find_file("nixos-config") or "/etc/nixos"
+    )
+    if nixos_config.is_dir():
+        nixos_config /= "default.nix"
+
+    if nixos_config.exists():
+        run_wrapper([os.getenv("EDITOR", "nano"), nixos_config], check=False)
     else:
-        if flake_flags:
-            raise NixOSRebuildError("'edit' does not support extra Nix flags")
-        nixos_config = Path(
-            os.getenv("NIXOS_CONFIG") or find_file("nixos-config") or "/etc/nixos"
-        )
-        if nixos_config.is_dir():
-            nixos_config /= "default.nix"
+        raise NixOSRebuildError("cannot find NixOS config file")
 
-        if nixos_config.exists():
-            run_wrapper([os.getenv("EDITOR", "nano"), nixos_config], check=False)
-        else:
-            raise NixOSRebuildError("cannot find NixOS config file")
+
+def edit_flake(flake: Flake | None, flake_flags: Args | None = None) -> None:
+    "Try to find and open NixOS configuration file in editor for Flake config."
+    run_wrapper(
+        [
+            "nix",
+            *FLAKE_FLAGS,
+            "edit",
+            *dict_to_flags(flake_flags),
+            "--",
+            str(flake),
+        ],
+        check=False,
+    )
 
 
 def find_file(file: str, nix_flags: Args | None = None) -> Path | None:

pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/services.py

@@ -1,4 +1,5 @@
 import argparse
+import json
 import logging
 import os
 import sys
@@ -10,7 +11,7 @@
 from .constants import EXECUTABLE
 from .models import Action, BuildAttr, Flake, ImageVariants, NixOSRebuildError, Profile
 from .process import Remote, cleanup_ssh
-from .utils import Args
+from .utils import Args, tabulate
 
 NIXOS_REBUILD_ATTR: Final = "config.system.build.nixos-rebuild"
 
@@ -315,3 +316,49 @@ def build_and_activate_system(
         common_flags=common_flags,
         flake_common_flags=flake_common_flags,
     )
+
+
+def edit(flake: Flake | None, flake_build_flags: Args | None = None) -> None:
+    if flake:
+        nix.edit_flake(flake, flake_build_flags)
+    else:
+        nix.edit()
+
+
+def list_generations(
+    args: argparse.Namespace,
+    profile: Profile,
+) -> None:
+    generations = nix.list_generations(profile)
+    if args.json:
+        print(json.dumps(generations, indent=2))
+    else:
+        headers = {
+            "generation": "Generation",
+            "date": "Build-date",
+            "nixosVersion": "NixOS version",
+            "kernelVersion": "Kernel",
+            "configurationRevision": "Configuration Revision",
+            "specialisations": "Specialisation",
+            "current": "Current",
+        }
+        print(tabulate(generations, headers=headers))
+
+
+def repl(
+    flake: Flake | None,
+    build_attr: BuildAttr,
+    flake_build_flags: Args,
+    build_flags: Args,
+) -> None:
+    if flake:
+        nix.repl_flake(flake, flake_build_flags)
+    else:
+        nix.repl(build_attr, build_flags)
+
+
+def write_version_suffix(build_flags: Args) -> None:
+    nixpkgs_path = nix.find_file("nixpkgs", build_flags)
+    rev = nix.get_nixpkgs_rev(nixpkgs_path)
+    if nixpkgs_path and rev:
+        (nixpkgs_path / ".version-suffix").write_text(rev)

pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_main.py

@@ -8,7 +8,6 @@
 from unittest.mock import ANY, Mock, call, patch
 
 import pytest
-from pytest import MonkeyPatch
 
 import nixos_rebuild as nr
 from nixos_rebuild.constants import WITH_NIX_2_18
@@ -128,92 +127,6 @@ def test_parse_args() -> None:
     ]
 
 
-@patch.dict(os.environ, {}, clear=True)
-@patch("os.execve", autospec=True)
-@patch(get_qualified_name(nr.nix.build), autospec=True)
-def test_reexec(mock_build: Mock, mock_execve: Mock, monkeypatch: MonkeyPatch) -> None:
-    monkeypatch.setattr(nr.services, "EXECUTABLE", "nixos-rebuild-ng")
-    argv = ["/path/bin/nixos-rebuild-ng", "switch", "--no-flake"]
-    args, _ = nr.parse_args(argv)
-    mock_build.return_value = Path("/path")
-
-    nr.reexec(argv, args, {"build": True}, {"flake": True})
-    mock_build.assert_has_calls(
-        [
-            call(
-                nr.services.NIXOS_REBUILD_ATTR,
-                nr.models.BuildAttr(ANY, ANY),
-                {"build": True, "no_out_link": True},
-            )
-        ]
-    )
-    # do not exec if there is no new version
-    mock_execve.assert_not_called()
-
-    mock_build.return_value = Path("/path/new")
-
-    nr.reexec(argv, args, {}, {})
-    # exec in the new version successfully
-    mock_execve.assert_called_once_with(
-        Path("/path/new/bin/nixos-rebuild-ng"),
-        ["/path/bin/nixos-rebuild-ng", "switch", "--no-flake"],
-        {"_NIXOS_REBUILD_REEXEC": "1"},
-    )
-
-    mock_execve.reset_mock()
-    mock_execve.side_effect = [OSError("BOOM"), None]
-
-    nr.reexec(argv, args, {}, {})
-    # exec in the previous version if the new version fails
-    mock_execve.assert_any_call(
-        Path("/path/bin/nixos-rebuild-ng"),
-        ["/path/bin/nixos-rebuild-ng", "switch", "--no-flake"],
-        {"_NIXOS_REBUILD_REEXEC": "1"},
-    )
-
-
-@patch.dict(os.environ, {}, clear=True)
-@patch("os.execve", autospec=True)
-@patch(get_qualified_name(nr.nix.build_flake), autospec=True)
-def test_reexec_flake(
-    mock_build: Mock, mock_execve: Mock, monkeypatch: MonkeyPatch
-) -> None:
-    monkeypatch.setattr(nr.services, "EXECUTABLE", "nixos-rebuild-ng")
-    argv = ["/path/bin/nixos-rebuild-ng", "switch", "--flake"]
-    args, _ = nr.parse_args(argv)
-    mock_build.return_value = Path("/path")
-
-    nr.reexec(argv, args, {"build": True}, {"flake": True})
-    mock_build.assert_called_once_with(
-        nr.services.NIXOS_REBUILD_ATTR,
-        nr.models.Flake(ANY, ANY),
-        {"flake": True, "no_link": True},
-    )
-    # do not exec if there is no new version
-    mock_execve.assert_not_called()
-
-    mock_build.return_value = Path("/path/new")
-
-    nr.reexec(argv, args, {}, {})
-    # exec in the new version successfully
-    mock_execve.assert_called_once_with(
-        Path("/path/new/bin/nixos-rebuild-ng"),
-        ["/path/bin/nixos-rebuild-ng", "switch", "--flake"],
-        {"_NIXOS_REBUILD_REEXEC": "1"},
-    )
-
-    mock_execve.reset_mock()
-    mock_execve.side_effect = [OSError("BOOM"), None]
-
-    nr.reexec(argv, args, {}, {})
-    # exec in the previous version if the new version fails
-    mock_execve.assert_any_call(
-        Path("/path/bin/nixos-rebuild-ng"),
-        ["/path/bin/nixos-rebuild-ng", "switch", "--flake"],
-        {"_NIXOS_REBUILD_REEXEC": "1"},
-    )
-
-
 @patch.dict(
     os.environ,
     {"NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM": "1"},

pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py

@@ -295,9 +295,21 @@ def test_copy_closure(monkeypatch: MonkeyPatch) -> None:
 
 @patch(get_qualified_name(n.run_wrapper, n), autospec=True)
 def test_edit(mock_run: Mock, monkeypatch: MonkeyPatch, tmpdir: Path) -> None:
-    # Flake
+    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_editd_flake(mock_run: Mock, monkeypatch: MonkeyPatch, tmpdir: Path) -> None:
     flake = m.Flake.parse(f"{tmpdir}#attr")
-    n.edit(flake, {"commit_lock_file": True})
+    n.edit_flake(flake, {"commit_lock_file": True})
     mock_run.assert_called_with(
         [
             "nix",
@@ -311,17 +323,6 @@ def test_edit(mock_run: Mock, monkeypatch: MonkeyPatch, tmpdir: Path) -> None:
         check=False,
     )
 
-    # Classic
-    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(None)
-        mock_run.assert_called_with(["editor", default_nix], check=False)
-
 
 @patch(
     get_qualified_name(n.run_wrapper, n),