gador/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

nixos-rebuild-ng: implement the remaining missing features (#360215)

Browse Source
This commit is contained in:
Thiago Kenji Okada
2024-12-04 09:48:21 +00:00
committed by GitHub
parent b6bd36b868 98e9372c1c
commit 83ef7bfcbb
14 changed files with 941 additions and 358 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkgs/by-name/ni/nixos-rebuild-ng/README.md
+15 -24
View File
@@ -91,24 +91,6 @@ ruff format .
conflicting with the current `nixos-rebuild`. This means you can keep both in
your system at the same time, but it also means that a few things like bash
completion are broken right now (since it looks at `nixos-rebuild` binary)
- `_NIXOS_REBUILD_EXEC` is **not** implemented yet, so different from
`nixos-rebuild`, this will use the current version of `nixos-rebuild-ng` in
your `PATH` to build/set profile/switch, while `nixos-rebuild` builds the new
version (the one that will be switched) and re-exec to it instead. This means
that in case of bugs in `nixos-rebuild-ng`, the only way that you will get
them fixed is **after** you switch to a new version
- `nix` bootstrap is also **not** implemented yet, so this means that you will
eval with an old version of Nix instead of a newer one. This is unlikely to
cause issues, because the build will happen in the daemon anyway (that is
only changed after the switch), and unless you are using bleeding edge `nix`
features you will probably have zero problems here. You can basically think
that using `nixos-rebuild-ng` is similar to running `nixos-rebuild --fast`
right now
- Ignore any performance advantages of the rewrite right now, because of the 2
caveats above
- `--target-host` and `--build-host` are not implemented yet and this is
probably the thing that will be most difficult to implement. Help here is
welcome
- Bugs in the profile manipulation can cause corruption of your profile that
may be difficult to fix, so right now I only recommend using
`nixos-rebuild-ng` if you are testing in a VM or in a filesystem with
@@ -118,20 +100,29 @@ ruff format .
## TODO
- [ ] Remote host/builders (via SSH)
- [x] Remote host/builders (via SSH)
- [x] Improve nix arguments handling (e.g.: `nixFlags` vs `copyFlags` in the
old `nixos-rebuild`)
- [ ] `_NIXOS_REBUILD_EXEC`
- [x] `_NIXOS_REBUILD_REEXEC`
- [ ] Port `nixos-rebuild.passthru.tests`
- [ ] Change module system to allow easier opt-in, like
`system.switch.enableNg` for `switch-to-configuration-ng`
- [ ] Improve documentation
- [ ] `nixos-rebuild repl` (calling old `nixos-rebuild` for now)
- [ ] `nix` build/bootstrap
- [x] `nixos-rebuild repl`
- [ ] Generate tab completion via [`shtab`](https://docs.iterative.ai/shtab/)
- [x] Reduce build closure
## TODON'T
- Reimplement `systemd-run` logic (will be moved to the new
[`apply`](https://github.com/NixOS/nixpkgs/pull/344407) script)
- Reimplement `systemd-run` logic: will be moved to the new
[`apply`](https://github.com/NixOS/nixpkgs/pull/344407) script
- Nix bootstrap: it is only used for non-Flake paths and it is basically
useless nowadays. It was created at a time when Nix was changing frequently
and there was a need to bootstrap a new version of Nix before evaluating the
configuration (otherwise the new Nixpkgs version may have code that is only
compatible with a newer version of Nix). Nixpkgs now has a policy to be
compatible with Nix 2.3, and even if this is bumped as long we don't do
drastic minimum version changes this should not be an issue. Also, the daemon
itself always run with the previous version since even we can replace Nix in
`PATH` (so Nix client), but we can't replace the daemon without switching to
a new version.
pkgs/by-name/ni/nixos-rebuild-ng/package.nix
-5
View File
@@ -39,11 +39,6 @@ python3Packages.buildPythonApplication rec {
nix
];
preBuild = ''
substituteInPlace nixos_rebuild/__init__.py \
--subst-var-by nixos_rebuild ${lib.getExe nixos-rebuild}
'';
postInstall =
''
installManPage ${nixos-rebuild}/share/man/man8/nixos-rebuild.8
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py
+182 -96
View File
@@ -1,19 +1,20 @@
import argparse
import atexit
import json
import logging
import os
import sys
from pathlib import Path
from subprocess import run
from tempfile import TemporaryDirectory
from subprocess import CalledProcessError, run
from typing import assert_never
from . import nix
from .models import Action, Flake, NRError, Profile
from .models import Action, BuildAttr, Flake, NRError, Profile
from .process import Remote, cleanup_ssh
from .utils import info
from .utils import Args, LogFormatter
VERBOSE = 0
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentParser]]:
@@ -52,7 +53,9 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
classic_build_flags.add_argument("--no-build-output", "-Q", action="store_true")
copy_flags = argparse.ArgumentParser(add_help=False)
copy_flags.add_argument("--use-substitutes", "-s", action="store_true")
copy_flags.add_argument(
"--use-substitutes", "--substitute-on-destination", "-s", action="store_true"
)
sub_parsers = {
"common_flags": common_flags,
@@ -86,8 +89,10 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
main_parser.add_argument("--ask-sudo-password", action="store_true")
main_parser.add_argument("--use-remote-sudo", action="store_true") # deprecated
main_parser.add_argument("--no-ssh-tty", action="store_true") # deprecated
# parser.add_argument("--build-host") # TODO
main_parser.add_argument("--fast", action="store_true")
main_parser.add_argument("--build-host")
main_parser.add_argument("--target-host")
main_parser.add_argument("--no-build-nix", action="store_true") # deprecated
main_parser.add_argument("action", choices=Action.values(), nargs="?")
return main_parser, sub_parsers
@@ -104,11 +109,11 @@ def parse_args(
}
def parser_warn(msg: str) -> None:
info(f"{parser.prog}: warning: {msg}")
print(f"{parser.prog}: warning: {msg}", file=sys.stderr)
global VERBOSE
# This flag affects both nix and this script
VERBOSE = args.verbose
if args.verbose:
logger.setLevel(logging.DEBUG)
# https://github.com/NixOS/nixpkgs/blob/master/pkgs/os-specific/linux/nixos-rebuild/nixos-rebuild.sh#L56
if args.action == Action.DRY_RUN.value:
@@ -117,6 +122,10 @@ def parse_args(
if args.ask_sudo_password:
args.sudo = True
if args.help or args.action is None:
r = run(["man", "8", "nixos-rebuild"], check=False)
parser.exit(r.returncode)
# TODO: use deprecated=True in Python >=3.13
if args.install_grub:
parser_warn("--install-grub deprecated, use --install-bootloader instead")
@@ -131,10 +140,14 @@ def parse_args(
if args.no_ssh_tty:
parser_warn("--no-ssh-tty deprecated, SSH's TTY is never used anymore")
# TODO: use deprecated=True in Python >=3.13
if args.no_build_nix:
parser_warn("--no-build-nix deprecated, we do not build nix anymore")
if args.action == Action.EDIT.value and (args.file or args.attr):
parser.error("--file and --attr are not supported with 'edit'")
if args.target_host and args.action not in (
if (args.target_host or args.build_host) and args.action not in (
Action.SWITCH.value,
Action.BOOT.value,
Action.TEST.value,
@@ -142,36 +155,67 @@ def parse_args(
Action.DRY_BUILD.value,
Action.DRY_ACTIVATE.value,
):
parser.error(f"--target-host is not supported with '{args.action}'")
parser.error(
f"--target-host/--build-host is not supported with '{args.action}'"
)
if args.flake and (args.file or args.attr):
parser.error("--flake cannot be used with --file or --attr")
if args.help or args.action is None:
r = run(["man", "8", "nixos-rebuild"], check=False)
parser.exit(r.returncode)
return args, args_groups
def reexec(
argv: list[str],
args: argparse.Namespace,
build_flags: dict[str, Args],
flake_build_flags: dict[str, Args],
) -> None:
drv = None
try:
# Need to set target_host=None, to avoid connecting to remote
if flake := Flake.from_arg(args.flake, None):
drv = nix.build_flake(
"pkgs.nixos-rebuild-ng",
flake,
**flake_build_flags,
no_link=True,
)
else:
drv = nix.build(
"pkgs.nixos-rebuild-ng",
BuildAttr.from_arg(args.attr, args.file),
**build_flags,
no_out_link=True,
)
except CalledProcessError:
logger.warning("could not find a newer version of nixos-rebuild")
if drv:
new = drv / "bin/nixos-rebuild-ng"
current = Path(argv[0])
# Disable re-exec during development
if current.name != "__main__.py" and new != current:
logging.debug(
"detected newer version of script, re-exec'ing, current=%s, new=%s",
argv[0],
new,
)
cleanup_ssh()
os.execve(new, argv, os.environ | {"_NIXOS_REBUILD_REEXEC": "1"})
def execute(argv: list[str]) -> None:
args, args_groups = parse_args(argv)
atexit.register(cleanup_ssh)
common_flags = vars(args_groups["common_flags"])
common_build_flags = common_flags | vars(args_groups["common_build_flags"])
build_flags = common_build_flags | vars(args_groups["classic_build_flags"])
flake_build_flags = common_build_flags | vars(args_groups["flake_build_flags"])
copy_flags = common_flags | vars(args_groups["copy_flags"])
# Will be cleaned up on exit automatically.
tmpdir = TemporaryDirectory(prefix="nixos-rebuild.")
tmpdir_path = Path(tmpdir.name)
atexit.register(cleanup_ssh, tmpdir_path)
profile = Profile.from_name(args.profile_name)
target_host = Remote.from_arg(args.target_host, args.ask_sudo_password, tmpdir_path)
flake = Flake.from_arg(args.flake, target_host)
if args.upgrade or args.upgrade_all:
nix.upgrade_channels(bool(args.upgrade_all))
@@ -182,6 +226,23 @@ def execute(argv: list[str]) -> None:
# executed, so it's safe to run nixos-rebuild against a potentially
# untrusted tree.
can_run = action in (Action.SWITCH, Action.BOOT, Action.TEST)
# Re-exec to a newer version of the script before building to ensure we get
# the latest fixes
if (
False # disabled until we introduce `config.system.build.nixos-rebuild-ng`
and can_run
and not args.fast
and not os.environ.get("_NIXOS_REBUILD_REEXEC")
):
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)
build_host = Remote.from_arg(args.build_host, False, validate_opts=False)
build_attr = BuildAttr.from_arg(args.attr, args.file)
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)
@@ -189,88 +250,111 @@ def execute(argv: list[str]) -> None:
(nixpkgs_path / ".version-suffix").write_text(rev)
match action:
case Action.SWITCH | Action.BOOT:
info("building the system configuration...")
if args.rollback:
path_to_config = nix.rollback(profile, target_host, sudo=args.sudo)
else:
if flake:
path_to_config = nix.nixos_build_flake(
"toplevel",
case (
Action.SWITCH
| Action.BOOT
| Action.TEST
| Action.BUILD
| Action.DRY_BUILD
| Action.DRY_ACTIVATE
):
logger.info("building the system configuration...")
attr = "config.system.build.toplevel"
dry_run = action == Action.DRY_BUILD
no_link = action in (Action.SWITCH, Action.BOOT)
rollback = bool(args.rollback)
match (action, rollback, build_host, flake):
case (Action.SWITCH | Action.BOOT, True, _, _):
path_to_config = nix.rollback(profile, target_host, sudo=args.sudo)
case (Action.TEST | Action.BUILD, True, _, _):
maybe_path_to_config = nix.rollback_temporary_profile(
profile,
target_host,
sudo=args.sudo,
)
if maybe_path_to_config: # kinda silly but this makes mypy happy
path_to_config = maybe_path_to_config
else:
raise NRError("could not find previous generation")
case (_, True, _, _):
raise NRError(f"--rollback is incompatible with '{action}'")
case (_, False, Remote(_), Flake(_)):
path_to_config = nix.remote_build_flake(
attr,
flake,
no_link=True,
build_host,
flake_build_flags=flake_build_flags,
copy_flags=copy_flags,
build_flags=build_flags,
)
case (_, False, None, Flake(_)):
path_to_config = nix.build_flake(
attr,
flake,
no_link=no_link,
dry_run=dry_run,
**flake_build_flags,
)
else:
path_to_config = nix.nixos_build(
"system",
args.attr,
args.file,
no_out_link=True,
case (_, False, Remote(_), None):
path_to_config = nix.remote_build(
attr,
build_attr,
build_host,
instantiate_flags=common_flags,
copy_flags=copy_flags,
build_flags=build_flags,
)
case (_, False, None, None):
path_to_config = nix.build(
attr,
build_attr,
no_out_link=no_link,
dry_run=dry_run,
**build_flags,
)
nix.copy_closure(path_to_config, target_host, **copy_flags)
nix.set_profile(profile, path_to_config, target_host, sudo=args.sudo)
nix.switch_to_configuration(
path_to_config,
action,
target_host,
sudo=args.sudo,
specialisation=args.specialisation,
install_bootloader=args.install_bootloader,
)
case Action.TEST | Action.BUILD | Action.DRY_BUILD | Action.DRY_ACTIVATE:
info("building the system configuration...")
dry_run = action == Action.DRY_BUILD
if args.rollback:
if action not in (Action.TEST, Action.BUILD):
raise NRError(f"--rollback is incompatible with '{action}'")
maybe_path_to_config = nix.rollback_temporary_profile(
profile,
target_host,
sudo=args.sudo,
case m:
# should never happen, but mypy is not smart enough to
# handle this with assert_never
raise NRError(f"invalid match for build: {m}")
if not rollback:
nix.copy_closure(
path_to_config,
to_host=target_host,
from_host=build_host,
**copy_flags,
)
if maybe_path_to_config: # kinda silly but this makes mypy happy
path_to_config = maybe_path_to_config
else:
raise NRError("could not find previous generation")
elif flake:
path_to_config = nix.nixos_build_flake(
"toplevel",
flake,
dry_run=dry_run,
**flake_build_flags,
)
else:
path_to_config = nix.nixos_build(
"system",
args.attr,
args.file,
dry_run=dry_run,
**build_flags,
)
if action in (Action.TEST, Action.DRY_ACTIVATE):
if action in (Action.SWITCH, Action.BOOT):
nix.set_profile(
profile,
path_to_config,
target_host=target_host,
sudo=args.sudo,
)
if action in (Action.SWITCH, Action.BOOT, Action.TEST, Action.DRY_ACTIVATE):
nix.switch_to_configuration(
path_to_config,
action,
target_host,
target_host=target_host,
sudo=args.sudo,
specialisation=args.specialisation,
install_bootloader=args.install_bootloader,
)
case Action.BUILD_VM | Action.BUILD_VM_WITH_BOOTLOADER:
info("building the system configuration...")
logger.info("building the system configuration...")
attr = "vm" if action == Action.BUILD_VM else "vmWithBootLoader"
if flake:
path_to_config = nix.nixos_build_flake(
attr,
path_to_config = nix.build_flake(
f"config.system.build.{attr}",
flake,
**flake_build_flags,
)
else:
path_to_config = nix.nixos_build(
attr,
args.attr,
args.file,
path_to_config = nix.build(
f"config.system.build.{attr}",
build_attr,
**build_flags,
)
vm_path = next(path_to_config.glob("bin/run-*-vm"), "./result/bin/run-*-vm")
@@ -307,21 +391,23 @@ def execute(argv: list[str]) -> None:
)
print(table)
case Action.REPL:
# For now just redirect it to `nixos-rebuild` instead of
# duplicating the code
os.execv(
"@nixos_rebuild@",
argv,
)
if flake:
nix.repl_flake("toplevel", flake, **flake_build_flags)
else:
nix.repl("system", build_attr, **build_flags)
case _:
assert_never(action)
def main() -> None:
ch = logging.StreamHandler()
ch.setFormatter(LogFormatter())
logger.addHandler(ch)
try:
execute(sys.argv)
except (Exception, KeyboardInterrupt) as ex:
if VERBOSE:
raise ex
if logger.level == logging.DEBUG:
raise
else:
sys.exit(str(ex))
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/models.py
+19 -3
View File
@@ -1,5 +1,3 @@
from __future__ import annotations
import platform
import re
import subprocess
@@ -45,12 +43,30 @@ class Action(Enum):
return [a.value for a in Action]
@dataclass(frozen=True)
class BuildAttr:
path: str | Path
attr: str | None
def to_attr(self, *attrs: str) -> str:
return f"{self.attr + '.' if self.attr else ''}{".".join(attrs)}"
@classmethod
def from_arg(cls, attr: str | None, file: str | None) -> Self:
if not (attr or file):
return cls("<nixpkgs/nixos>", None)
return cls(Path(file or "default.nix"), attr)
@dataclass(frozen=True)
class Flake:
path: Path
attr: str
_re: ClassVar = re.compile(r"^(?P<path>[^\#]*)\#?(?P<attr>[^\#\"]*)$")
def to_attr(self, *attrs: str) -> str:
return f"{self}.{".".join(attrs)}"
@override
def __str__(self) -> str:
return f"{self.path}#{self.attr}"
@@ -125,7 +141,7 @@ class Profile:
path: Path
@classmethod
def from_name(cls, name: str = "system") -> Self:
def from_arg(cls, name: str) -> Self:
match name:
case "system":
return cls(name, Path("/nix/var/nix/profiles/system"))
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/nix.py
+168 -63
View File
@@ -1,11 +1,15 @@
import logging
import os
from datetime import datetime
from importlib.resources import files
from pathlib import Path
from string import Template
from subprocess import PIPE, CalledProcessError
from typing import Final
from .models import (
Action,
BuildAttr,
Flake,
Generation,
GenerationJson,
@@ -13,30 +17,152 @@ from .models import (
Profile,
Remote,
)
from .process import run_wrapper
from .utils import Args, dict_to_flags, info
from .process import SSH_DEFAULT_OPTS, run_wrapper
from .utils import Args, dict_to_flags
FLAKE_FLAGS: Final = ["--extra-experimental-features", "nix-command flakes"]
FLAKE_REPL_TEMPLATE: Final = "repl.nix.template"
logger = logging.getLogger(__name__)
def build(
attr: str,
build_attr: BuildAttr,
**build_flags: Args,
) -> Path:
"""Build NixOS attribute using classic Nix.
Returns the built attribute as path.
"""
run_args = [
"nix-build",
build_attr.path,
"--attr",
build_attr.to_attr(attr),
*dict_to_flags(build_flags),
]
r = run_wrapper(run_args, stdout=PIPE)
return Path(r.stdout.strip())
def build_flake(
attr: str,
flake: Flake,
**flake_build_flags: Args,
) -> Path:
"""Build NixOS attribute using Flakes.
Returns the built attribute as path.
"""
run_args = [
"nix",
*FLAKE_FLAGS,
"build",
"--print-out-paths",
flake.to_attr(attr),
*dict_to_flags(flake_build_flags),
]
r = run_wrapper(run_args, stdout=PIPE)
return Path(r.stdout.strip())
def remote_build(
attr: str,
build_attr: BuildAttr,
build_host: Remote | None,
build_flags: dict[str, Args] | None = None,
instantiate_flags: dict[str, Args] | None = None,
copy_flags: dict[str, Args] | None = None,
) -> Path:
r = run_wrapper(
[
"nix-instantiate",
"--raw",
build_attr.path,
"--attr",
build_attr.to_attr(attr),
*dict_to_flags(instantiate_flags or {}),
],
stdout=PIPE,
)
drv = Path(r.stdout.strip())
copy_closure(drv, to_host=build_host, from_host=None, **(copy_flags or {}))
r = run_wrapper(
["nix-store", "--realise", drv, *dict_to_flags(build_flags or {})],
remote=build_host,
stdout=PIPE,
)
return Path(r.stdout.strip())
def remote_build_flake(
attr: str,
flake: Flake,
build_host: Remote,
flake_build_flags: dict[str, Args] | None = None,
copy_flags: dict[str, Args] | None = None,
build_flags: dict[str, Args] | None = None,
) -> Path:
r = run_wrapper(
[
"nix",
*FLAKE_FLAGS,
"eval",
"--raw",
flake.to_attr(attr, "drvPath"),
*dict_to_flags(flake_build_flags or {}),
],
stdout=PIPE,
)
drv = Path(r.stdout.strip())
copy_closure(drv, to_host=build_host, from_host=None, **(copy_flags or {}))
r = run_wrapper(
[
"nix",
*FLAKE_FLAGS,
"build",
f"{drv}^*",
"--print-out-paths",
*dict_to_flags(build_flags or {}),
],
remote=build_host,
stdout=PIPE,
)
return Path(r.stdout.strip())
def copy_closure(
closure: Path,
target_host: Remote | None,
to_host: Remote | None,
from_host: Remote | None = None,
**copy_flags: Args,
) -> None:
host = target_host
"""Copy a nix closure to or from host to localhost.
Also supports copying a closure from a remote to another remote."""
host = to_host or from_host
if not host:
return
sshopts = os.getenv("NIX_SSHOPTS", "")
run_wrapper(
[
"nix-copy-closure",
*dict_to_flags(copy_flags),
"--to",
"--to" if to_host else "--from",
host.host,
closure,
],
extra_env={"NIX_SSHOPTS": " ".join(host.opts)},
extra_env={
# Using raw NIX_SSHOPTS here to avoid messing up with the passed
# parameters, and we do not add the SSH_DEFAULT_OPTS in the remote
# to remote case, otherwise it will fail because of ControlPath
# will not exist in remote
"NIX_SSHOPTS": sshopts
if from_host and to_host
else " ".join(filter(lambda x: x, [*SSH_DEFAULT_OPTS, sshopts]))
},
remote=from_host if to_host else None,
)
@@ -58,13 +184,7 @@ def edit(flake: Flake | None, **flake_flags: Args) -> None:
if flake_flags:
raise NRError("'edit' does not support extra Nix flags")
nixos_config = Path(
os.getenv("NIXOS_CONFIG")
or run_wrapper(
["nix-instantiate", "--find-file", "nixos-config"],
stdout=PIPE,
check=False,
).stdout.strip()
or "/etc/nixos/default.nix"
os.getenv("NIXOS_CONFIG") or find_file("nixos-config") or "/etc/nixos"
)
if nixos_config.is_dir():
nixos_config /= "default.nix"
@@ -76,7 +196,7 @@ def edit(flake: Flake | None, **flake_flags: Args) -> None:
def find_file(file: str, **nix_flags: Args) -> Path | None:
"Find classic Nixpkgs location."
"Find classic Nix file location."
r = run_wrapper(
["nix-instantiate", "--find-file", file, *dict_to_flags(nix_flags)],
stdout=PIPE,
@@ -103,7 +223,7 @@ def get_nixpkgs_rev(nixpkgs_path: Path | None) -> str | None:
)
except FileNotFoundError:
# Git is not included in the closure so we need to check
info(f"warning: Git not found; cannot figure out revision of '{nixpkgs_path}'")
logger.warning(f"Git not found; cannot figure out revision of '{nixpkgs_path}'")
return None
if rev := r.stdout.strip():
@@ -199,13 +319,15 @@ def list_generations(profile: Profile) -> list[GenerationJson]:
)
try:
nixos_version = (generation_path / "nixos-version").read_text().strip()
except IOError:
except IOError as ex:
logger.debug("could not get nixos-version: %s", ex)
nixos_version = "Unknown"
try:
kernel_version = next(
(generation_path / "kernel-modules/lib/modules").iterdir()
).name
except IOError:
except IOError as ex:
logger.debug("could not get kernel version: %s", ex)
kernel_version = "Unknown"
specialisations = [
s.name for s in (generation_path / "specialisation").glob("*") if s.is_dir()
@@ -215,7 +337,8 @@ def list_generations(profile: Profile) -> list[GenerationJson]:
[generation_path / "sw/bin/nixos-version", "--configuration-revision"],
capture_output=True,
).stdout.strip()
except (CalledProcessError, IOError):
except (CalledProcessError, IOError) as ex:
logger.debug("could not get configuration revision: %s", ex)
configuration_revision = "Unknown"
result.append(
@@ -233,53 +356,35 @@ def list_generations(profile: Profile) -> list[GenerationJson]:
return result
def nixos_build(
attr: str,
pre_attr: str | None,
file: str | None,
**nix_flags: Args,
) -> Path:
"""Build NixOS attribute using classic Nix.
def repl(attr: str, build_attr: BuildAttr, **nix_flags: Args) -> None:
run_args = ["nix", "repl", "--file", build_attr.path]
if build_attr.attr:
run_args.append(build_attr.attr)
run_wrapper([*run_args, *dict_to_flags(nix_flags)])
It will by default build `<nixpkgs/nixos>` with `attr`, however it
optionally supports building from an external file and custom attributes
paths.
Returns the built attribute as path.
"""
if pre_attr or file:
run_args = [
"nix-build",
file or "default.nix",
"--attr",
f"{'.'.join(x for x in [pre_attr, attr] if x)}",
def repl_flake(attr: str, flake: Flake, **flake_flags: Args) -> None:
expr = Template(
files(__package__).joinpath(FLAKE_REPL_TEMPLATE).read_text()
).substitute(
flake_path=flake.path,
flake_attr=flake.attr,
bold="\033[1m",
blue="\033[34;1m",
attention="\033[35;1m",
reset="\033[0m",
)
run_wrapper(
[
"nix",
*FLAKE_FLAGS,
"repl",
"--impure",
"--expr",
expr,
*dict_to_flags(flake_flags),
]
else:
run_args = ["nix-build", "<nixpkgs/nixos>", "--attr", attr]
run_args += dict_to_flags(nix_flags)
r = run_wrapper(run_args, stdout=PIPE)
return Path(r.stdout.strip())
def nixos_build_flake(
attr: str,
flake: Flake,
**flake_flags: Args,
) -> Path:
"""Build NixOS attribute using Flakes.
Returns the built attribute as path.
"""
run_args = [
"nix",
*FLAKE_FLAGS,
"build",
"--print-out-paths",
f"{flake}.config.system.build.{attr}",
*dict_to_flags(flake_flags),
]
r = run_wrapper(run_args, stdout=PIPE)
return Path(r.stdout.strip())
)
def rollback(profile: Profile, target_host: Remote | None, sudo: bool) -> Path:
@@ -368,7 +473,7 @@ def upgrade_channels(all: bool = False) -> None:
that has a `.update-on-nixos-rebuild` file) or all.
"""
for channel_path in Path("/nix/var/nix/profiles/per-user/root/channels/").glob("*"):
if (
if channel_path.is_dir() and (
all
or channel_path.name == "nixos"
or (channel_path / ".update-on-nixos-rebuild").exists()
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py
+73 -33
View File
@@ -1,13 +1,25 @@
from __future__ import annotations
import logging
import os
import shlex
import subprocess
from dataclasses import dataclass
from getpass import getpass
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Self, Sequence, TypedDict, Unpack
from .utils import info
logger = logging.getLogger(__name__)
TMPDIR = TemporaryDirectory(prefix="nixos-rebuild.")
TMPDIR_PATH = Path(TMPDIR.name)
SSH_DEFAULT_OPTS = [
"-o",
"ControlMaster=auto",
"-o",
f"ControlPath={TMPDIR_PATH / "ssh-%n"}",
"-o",
"ControlPersist=60",
]
@dataclass(frozen=True)
@@ -21,22 +33,14 @@ class Remote:
cls,
host: str | None,
ask_sudo_password: bool | None,
tmp_dir: Path,
validate_opts: bool = True,
) -> Self | None:
if not host:
return None
opts = os.getenv("NIX_SSHOPTS", "").split()
cls._validate_opts(opts, ask_sudo_password)
opts += [
# SSH ControlMaster flags, allow for faster re-connection
"-o",
"ControlMaster=auto",
"-o",
f"ControlPath={tmp_dir / "ssh-%n"}",
"-o",
"ControlPersist=60",
]
opts = shlex.split(os.getenv("NIX_SSHOPTS", ""))
if validate_opts:
cls._validate_opts(opts, ask_sudo_password)
sudo_password = None
if ask_sudo_password:
sudo_password = getpass(f"[sudo] password for {host}: ")
@@ -46,13 +50,13 @@ class Remote:
def _validate_opts(opts: list[str], ask_sudo_password: bool | None) -> None:
for o in opts:
if o in ["-t", "-tt", "RequestTTY=yes", "RequestTTY=force"]:
info(
f"warning: detected option '{o}' in NIX_SSHOPTS. SSH's TTY "
+ "may cause issues, it is recommended to remove this option"
logger.warning(
f"detected option '{o}' in NIX_SSHOPTS. SSH's TTY may "
+ "cause issues, it is recommended to remove this option"
)
if not ask_sudo_password:
info(
"If you want to prompt for sudo password use "
logger.warning(
"if you want to prompt for sudo password use "
+ "'--ask-sudo-password' option instead"
)
@@ -64,10 +68,15 @@ class RunKwargs(TypedDict, total=False):
stdout: int | None
def cleanup_ssh(tmp_dir: Path) -> None:
def cleanup_ssh() -> None:
"Close SSH ControlMaster connection."
for ctrl in tmp_dir.glob("ssh-*"):
subprocess.run(["ssh", "-o", f"ControlPath={ctrl}", "exit"], check=False)
for ctrl in TMPDIR_PATH.glob("ssh-*"):
run_wrapper(
["ssh", "-o", f"ControlPath={ctrl}", "-O", "exit", "dummyhost"],
check=False,
capture_output=True,
)
TMPDIR.cleanup()
def run_wrapper(
@@ -92,21 +101,52 @@ def run_wrapper(
input = remote.sudo_password + "\n"
else:
args = ["sudo", *args]
args = ["ssh", *remote.opts, remote.host, "--", *args]
args = [
"ssh",
*remote.opts,
*SSH_DEFAULT_OPTS,
remote.host,
"--",
# SSH will join the parameters here and pass it to the shell, so we
# need to quote it to avoid issues.
# We can't use `shlex.join`, otherwise we will hit MAX_ARG_STRLEN
# limits when the command becomes too big.
*[shlex.quote(str(a)) for a in args],
]
else:
if extra_env:
env = os.environ | extra_env
if sudo:
args = ["sudo", *args]
return subprocess.run(
logger.debug(
"calling run with args=%r, kwargs=%r, extra_env=%r",
args,
check=check,
env=env,
input=input,
# Hope nobody is using NixOS with non-UTF8 encodings, but "surrogateescape"
# should still work in those systems.
text=True,
errors="surrogateescape",
**kwargs,
kwargs,
extra_env,
)
try:
r = subprocess.run(
args,
check=check,
env=env,
input=input,
# Hope nobody is using NixOS with non-UTF8 encodings, but "surrogateescape"
# should still work in those systems.
text=True,
errors="surrogateescape",
**kwargs,
)
if kwargs.get("capture_output") or kwargs.get("stderr") or kwargs.get("stdout"):
logger.debug("captured output stdout=%r, stderr=%r", r.stdout, r.stderr)
return r
except subprocess.CalledProcessError:
if sudo and remote:
logger.error(
"while running command with remote sudo, did you forget to use "
+ "--ask-sudo-password?"
)
raise
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/repl.nix.template
+41
View File
@@ -0,0 +1,41 @@
# vim: set syntax=nix:
let
flake = builtins.getFlake ''${flake_path}'';
configuration = flake.${flake_attr};
motd = ''
$${"\n"}
Hello and welcome to the NixOS configuration
${flake_attr}
in ${flake_path}
The following is loaded into nix repl's scope:
- ${blue}config${reset} All option values
- ${blue}options${reset} Option data and metadata
- ${blue}pkgs${reset} Nixpkgs package set
- ${blue}lib${reset} Nixpkgs library functions
- other module arguments
- ${blue}flake${reset} Flake outputs, inputs and source info of ${flake_path}
Use tab completion to browse around ${blue}config${reset}.
Use ${bold}:r${reset} to ${bold}reload${reset} everything after making a change in the flake.
(assuming ${flake_path} is a mutable flake ref)
See ${bold}:?${reset} for more repl commands.
${attention}warning:${reset} nixos-rebuild repl does not currently enforce pure evaluation.
'';
scope =
assert configuration._type or null == ''configuration'';
assert configuration.class or ''nixos'' == ''nixos'';
configuration._module.args
// configuration._module.specialArgs
// {
inherit (configuration) config options;
lib = configuration.lib or configuration.pkgs.lib;
inherit flake;
};
in
builtins.seq scope builtins.trace motd scope
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/utils.py
+16 -4
View File
@@ -1,11 +1,23 @@
import sys
from functools import partial
from typing import TypeAlias
import logging
from typing import TypeAlias, override
info = partial(print, file=sys.stderr)
Args: TypeAlias = bool | str | list[str] | int | None
class LogFormatter(logging.Formatter):
formatters = {
logging.INFO: logging.Formatter("%(message)s"),
logging.DEBUG: logging.Formatter("%(levelname)s: %(name)s: %(message)s"),
"DEFAULT": logging.Formatter("%(levelname)s: %(message)s"),
}
@override
def format(self, record: logging.LogRecord) -> str:
record.levelname = record.levelname.lower()
formatter = self.formatters.get(record.levelno, self.formatters["DEFAULT"])
return formatter.format(record)
def dict_to_flags(d: dict[str, Args]) -> list[str]:
flags = []
for key, value in d.items():
pkgs/by-name/ni/nixos-rebuild-ng/src/pyproject.toml
+5
View File
@@ -9,6 +9,9 @@ version = "0.0.0"
[project.scripts]
nixos-rebuild = "nixos_rebuild:main"
[tool.setuptools.package-data]
nixos_rebuild = ["*.template.nix"]
[tool.mypy]
# `--strict` config, but explicit options to avoid breaking build when mypy is
# updated
@@ -40,6 +43,8 @@ extend-select = [
"I",
# require `check` argument for `subprocess.run`
"PLW1510",
# check for needless exception names in raise statements
"TRY201",
]
[tool.ruff.lint.per-file-ignores]
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/helpers.py
-2
View File
@@ -1,5 +1,3 @@
from __future__ import annotations
from types import ModuleType
from typing import Any, Callable
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_main.py
+158 -38
View File
@@ -1,3 +1,4 @@
import logging
import textwrap
from pathlib import Path
from subprocess import PIPE, CompletedProcess
@@ -43,7 +44,7 @@ def test_parse_args() -> None:
"bar",
]
)
assert nr.VERBOSE == 0
assert nr.logger.level == logging.INFO
assert r1.flake == "/etc/nixos"
assert r1.install_bootloader is True
assert r1.install_grub is True
@@ -65,7 +66,7 @@ def test_parse_args() -> None:
"-vvv",
]
)
assert nr.VERBOSE == 3
assert nr.logger.level == logging.DEBUG
assert r2.verbose == 3
assert r2.flake is False
assert r2.action == "dry-build"
@@ -94,7 +95,7 @@ def test_execute_nix_boot(mock_run: Any, tmp_path: Path) -> None:
CompletedProcess([], 0),
]
nr.execute(["nixos-rebuild", "boot", "--no-flake", "-vvv"])
nr.execute(["nixos-rebuild", "boot", "--no-flake", "-vvv", "--fast"])
assert mock_run.call_count == 6
mock_run.assert_has_calls(
@@ -121,7 +122,7 @@ def test_execute_nix_boot(mock_run: Any, tmp_path: Path) -> None:
"nix-build",
"<nixpkgs/nixos>",
"--attr",
"system",
"config.system.build.toplevel",
"--no-out-link",
"-vvv",
],
@@ -172,6 +173,7 @@ def test_execute_nix_switch_flake(mock_run: Any, tmp_path: Path) -> None:
"--install-bootloader",
"--sudo",
"--verbose",
"--fast",
]
)
@@ -216,9 +218,9 @@ def test_execute_nix_switch_flake(mock_run: Any, tmp_path: Path) -> None:
@patch.dict(nr.process.os.environ, {}, clear=True)
@patch(get_qualified_name(nr.process.subprocess.run), autospec=True)
@patch(get_qualified_name(nr.TemporaryDirectory, nr)) # can't autospec
def test_execute_nix_switch_flake_remote(
mock_tmpdir: Any,
@patch(get_qualified_name(nr.cleanup_ssh, nr), autospec=True)
def test_execute_nix_switch_flake_target_host(
mock_cleanup_ssh: Any,
mock_run: Any,
tmp_path: Path,
) -> None:
@@ -234,7 +236,6 @@ def test_execute_nix_switch_flake_remote(
# switch_to_configuration
CompletedProcess([], 0),
]
mock_tmpdir.return_value.name = "/tmp/test"
nr.execute(
[
@@ -245,6 +246,7 @@ def test_execute_nix_switch_flake_remote(
"--use-remote-sudo",
"--target-host",
"user@localhost",
"--fast",
]
)
@@ -273,15 +275,123 @@ def test_execute_nix_switch_flake_remote(
call(
[
"ssh",
"-o",
"ControlMaster=auto",
"-o",
"ControlPath=/tmp/test/ssh-%n",
"-o",
"ControlPersist=60",
*nr.process.SSH_DEFAULT_OPTS,
"user@localhost",
"--",
"sudo",
"nix-env",
"-p",
"/nix/var/nix/profiles/system",
"--set",
str(config_path),
],
check=True,
**DEFAULT_RUN_KWARGS,
),
call(
[
"ssh",
*nr.process.SSH_DEFAULT_OPTS,
"user@localhost",
"--",
"sudo",
"env",
"NIXOS_INSTALL_BOOTLOADER=0",
f"{config_path / 'bin/switch-to-configuration'}",
"switch",
],
check=True,
**DEFAULT_RUN_KWARGS,
),
]
)
@patch.dict(nr.process.os.environ, {}, clear=True)
@patch(get_qualified_name(nr.process.subprocess.run), autospec=True)
@patch(get_qualified_name(nr.cleanup_ssh, nr), autospec=True)
def test_execute_nix_switch_flake_build_host(
mock_cleanup_ssh: Any,
mock_run: Any,
tmp_path: Path,
) -> None:
config_path = tmp_path / "test"
config_path.touch()
mock_run.side_effect = [
# nixos_build_flake
CompletedProcess([], 0, str(config_path)),
CompletedProcess([], 0),
CompletedProcess([], 0, str(config_path)),
# set_profile
CompletedProcess([], 0),
# copy_closure
CompletedProcess([], 0),
# switch_to_configuration
CompletedProcess([], 0),
]
nr.execute(
[
"nixos-rebuild",
"switch",
"--flake",
"/path/to/config#hostname",
"--build-host",
"user@localhost",
"--fast",
]
)
assert mock_run.call_count == 6
mock_run.assert_has_calls(
[
call(
[
"nix",
"--extra-experimental-features",
"nix-command flakes",
"eval",
"--raw",
"/path/to/config#nixosConfigurations.hostname.config.system.build.toplevel.drvPath",
],
check=True,
stdout=PIPE,
**DEFAULT_RUN_KWARGS,
),
call(
["nix-copy-closure", "--to", "user@localhost", config_path],
check=True,
**DEFAULT_RUN_KWARGS,
),
call(
[
"ssh",
*nr.process.SSH_DEFAULT_OPTS,
"user@localhost",
"--",
"nix",
"--extra-experimental-features",
"'nix-command flakes'",
"build",
f"'{config_path}^*'",
"--print-out-paths",
],
check=True,
stdout=PIPE,
**DEFAULT_RUN_KWARGS,
),
call(
[
"nix-copy-closure",
"--from",
"user@localhost",
config_path,
],
check=True,
**DEFAULT_RUN_KWARGS,
),
call(
[
"nix-env",
"-p",
Path("/nix/var/nix/profiles/system"),
@@ -292,22 +402,7 @@ def test_execute_nix_switch_flake_remote(
**DEFAULT_RUN_KWARGS,
),
call(
[
"ssh",
"-o",
"ControlMaster=auto",
"-o",
"ControlPath=/tmp/test/ssh-%n",
"-o",
"ControlPersist=60",
"user@localhost",
"--",
"sudo",
"env",
"NIXOS_INSTALL_BOOTLOADER=0",
config_path / "bin/switch-to-configuration",
"switch",
],
[config_path / "bin/switch-to-configuration", "switch"],
check=True,
**DEFAULT_RUN_KWARGS,
),
@@ -320,7 +415,9 @@ def test_execute_switch_rollback(mock_run: Any, tmp_path: Path) -> None:
nixpkgs_path = tmp_path / "nixpkgs"
nixpkgs_path.touch()
nr.execute(["nixos-rebuild", "switch", "--rollback", "--install-bootloader"])
nr.execute(
["nixos-rebuild", "switch", "--rollback", "--install-bootloader", "--fast"]
)
assert mock_run.call_count >= 2
# ignoring update_nixpkgs_rev calls
@@ -348,6 +445,35 @@ def test_execute_switch_rollback(mock_run: Any, tmp_path: Path) -> None:
)
@patch(get_qualified_name(nr.process.subprocess.run), autospec=True)
def test_execute_build(mock_run: Any, tmp_path: Path) -> None:
config_path = tmp_path / "test"
config_path.touch()
mock_run.side_effect = [
# nixos_build_flake
CompletedProcess([], 0, str(config_path)),
]
nr.execute(["nixos-rebuild", "build", "--no-flake", "--fast"])
assert mock_run.call_count == 1
mock_run.assert_has_calls(
[
call(
[
"nix-build",
"<nixpkgs/nixos>",
"--attr",
"config.system.build.toplevel",
],
check=True,
stdout=PIPE,
**DEFAULT_RUN_KWARGS,
)
]
)
@patch(get_qualified_name(nr.process.subprocess.run), autospec=True)
@patch(get_qualified_name(nr.nix.Path.exists, nr.nix), autospec=True, return_value=True)
@patch(get_qualified_name(nr.nix.Path.mkdir, nr.nix), autospec=True)
@@ -372,13 +498,7 @@ def test_execute_test_rollback(
]
nr.execute(
[
"nixos-rebuild",
"test",
"--rollback",
"--profile-name",
"foo",
]
["nixos-rebuild", "test", "--rollback", "--profile-name", "foo", "--fast"]
)
assert mock_run.call_count == 2
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_models.py
+33 -3
View File
@@ -9,6 +9,27 @@ import nixos_rebuild.models as m
from .helpers import get_qualified_name
def test_build_attr_from_arg() -> None:
assert m.BuildAttr.from_arg(None, None) == m.BuildAttr("<nixpkgs/nixos>", None)
assert m.BuildAttr.from_arg("attr", None) == m.BuildAttr(
Path("default.nix"), "attr"
)
assert m.BuildAttr.from_arg("attr", "file.nix") == m.BuildAttr(
Path("file.nix"), "attr"
)
assert m.BuildAttr.from_arg(None, "file.nix") == m.BuildAttr(Path("file.nix"), None)
def test_build_attr_to_attr() -> None:
assert (
m.BuildAttr("<nixpkgs/nixos>", None).to_attr("attr1", "attr2") == "attr1.attr2"
)
assert (
m.BuildAttr("<nixpkgs/nixos>", "preAttr").to_attr("attr1", "attr2")
== "preAttr.attr1.attr2"
)
def test_flake_parse() -> None:
assert m.Flake.parse("/path/to/flake#attr") == m.Flake(
Path("/path/to/flake"), "nixosConfigurations.attr"
@@ -24,6 +45,15 @@ def test_flake_parse() -> None:
assert m.Flake.parse(".") == m.Flake(Path("."), "nixosConfigurations.default")
def test_flake_to_attr() -> None:
assert (
m.Flake(Path("/path/to/flake"), "nixosConfigurations.preAttr").to_attr(
"attr1", "attr2"
)
== "/path/to/flake#nixosConfigurations.preAttr.attr1.attr2"
)
@patch(get_qualified_name(platform.node), autospec=True)
def test_flake_from_arg(mock_node: Any) -> None:
mock_node.return_value = "hostname"
@@ -100,13 +130,13 @@ def test_flake_from_arg(mock_node: Any) -> None:
@patch(get_qualified_name(m.Path.mkdir, m), autospec=True)
def test_profile_from_name(mock_mkdir: Any) -> None:
assert m.Profile.from_name("system") == m.Profile(
def test_profile_from_arg(mock_mkdir: Any) -> None:
assert m.Profile.from_arg("system") == m.Profile(
"system",
Path("/nix/var/nix/profiles/system"),
)
assert m.Profile.from_name("something") == m.Profile(
assert m.Profile.from_arg("something") == m.Profile(
"something",
Path("/nix/var/nix/profiles/system-profiles/something"),
)
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py
+211 -59
View File
@@ -8,21 +8,214 @@ import pytest
import nixos_rebuild.models as m
import nixos_rebuild.nix as n
import nixos_rebuild.process as p
from .helpers import get_qualified_name
@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: Any, monkeypatch: Any) -> None:
assert n.build(
"config.system.build.attr", m.BuildAttr("<nixpkgs/nixos>", None), nix_flag="foo"
) == Path("/path/to/file")
mock_run.assert_called_with(
[
"nix-build",
"<nixpkgs/nixos>",
"--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: Any) -> None:
flake = m.Flake.parse(".#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",
".#nixosConfigurations.hostname.config.system.build.toplevel",
"--no-link",
"--nix-flag",
"foo",
],
stdout=PIPE,
)
@patch(
get_qualified_name(n.run_wrapper, n),
autospec=True,
return_value=CompletedProcess([], 0, stdout=" \n/path/to/file\n "),
)
def test_remote_build(mock_run: Any, monkeypatch: Any) -> None:
build_host = m.Remote("user@host", [], None)
monkeypatch.setenv("NIX_SSHOPTS", "--ssh opts")
assert n.remote_build(
"config.system.build.toplevel",
m.BuildAttr("<nixpkgs/nixos>", "preAttr"),
build_host,
build_flags={"build": True},
instantiate_flags={"inst": True},
copy_flags={"copy": True},
) == Path("/path/to/file")
mock_run.assert_has_calls(
[
call(
[
"nix-instantiate",
"--raw",
"<nixpkgs/nixos>",
"--attr",
"preAttr.config.system.build.toplevel",
"--inst",
],
stdout=PIPE,
),
call(
[
"nix-copy-closure",
"--copy",
"--to",
"user@host",
Path("/path/to/file"),
],
extra_env={
"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS + ["--ssh opts"])
},
remote=None,
),
call(
["nix-store", "--realise", Path("/path/to/file"), "--build"],
remote=build_host,
stdout=PIPE,
),
]
)
@patch(
get_qualified_name(n.run_wrapper, n),
autospec=True,
return_value=CompletedProcess([], 0, stdout=" \n/path/to/file\n "),
)
def test_remote_build_flake(mock_run: Any, monkeypatch: Any) -> None:
flake = m.Flake.parse(".#hostname")
build_host = m.Remote("user@host", [], None)
monkeypatch.setenv("NIX_SSHOPTS", "--ssh opts")
assert n.remote_build_flake(
"config.system.build.toplevel",
flake,
build_host,
flake_build_flags={"flake": True},
copy_flags={"copy": True},
build_flags={"build": True},
) == Path("/path/to/file")
mock_run.assert_has_calls(
[
call(
[
"nix",
"--extra-experimental-features",
"nix-command flakes",
"eval",
"--raw",
".#nixosConfigurations.hostname.config.system.build.toplevel.drvPath",
"--flake",
],
stdout=PIPE,
),
call(
[
"nix-copy-closure",
"--copy",
"--to",
"user@host",
Path("/path/to/file"),
],
extra_env={
"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS + ["--ssh opts"])
},
remote=None,
),
call(
[
"nix",
"--extra-experimental-features",
"nix-command flakes",
"build",
"/path/to/file^*",
"--print-out-paths",
"--build",
],
remote=build_host,
stdout=PIPE,
),
]
)
@patch(get_qualified_name(n.run_wrapper, n), autospec=True)
def test_copy_closure(mock_run: Any) -> None:
def test_copy_closure(mock_run: Any, monkeypatch: Any) -> None:
closure = Path("/path/to/closure")
n.copy_closure(closure, None)
mock_run.assert_not_called()
target_host = m.Remote("user@host", ["--ssh", "opt"], None)
target_host = m.Remote("user@target.host", [], None)
build_host = m.Remote("user@build.host", [], None)
n.copy_closure(closure, target_host)
mock_run.assert_called_with(
["nix-copy-closure", "--to", "user@host", closure],
extra_env={"NIX_SSHOPTS": "--ssh opt"},
["nix-copy-closure", "--to", "user@target.host", closure],
extra_env={"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS)},
remote=None,
)
monkeypatch.setenv("NIX_SSHOPTS", "--ssh build-opt")
n.copy_closure(closure, None, build_host)
mock_run.assert_called_with(
["nix-copy-closure", "--from", "user@build.host", closure],
extra_env={"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS + ["--ssh build-opt"])},
remote=None,
)
monkeypatch.setenv("NIX_SSHOPTS", "--ssh build-target-opt")
n.copy_closure(closure, target_host, build_host)
mock_run.assert_called_with(
["nix-copy-closure", "--to", "user@target.host", closure],
remote=build_host,
extra_env={"NIX_SSHOPTS": "--ssh build-target-opt"},
)
@@ -193,65 +386,23 @@ def test_list_generations(mock_get_generations: Any, tmp_path: Path) -> None:
]
@patch(
get_qualified_name(n.run_wrapper, n),
autospec=True,
return_value=CompletedProcess([], 0, stdout=" \n/path/to/file\n "),
)
def test_nixos_build_flake(mock_run: Any) -> None:
flake = m.Flake.parse(".#hostname")
assert n.nixos_build_flake(
"toplevel",
flake,
no_link=True,
nix_flag="foo",
) == Path("/path/to/file")
@patch(get_qualified_name(n.run_wrapper, n), autospec=True)
def test_repl(mock_run: Any) -> None:
n.repl("attr", m.BuildAttr("<nixpkgs/nixos>", None), nix_flag=True)
mock_run.assert_called_with(
[
"nix",
"--extra-experimental-features",
"nix-command flakes",
"build",
"--print-out-paths",
".#nixosConfigurations.hostname.config.system.build.toplevel",
"--no-link",
"--nix-flag",
"foo",
],
stdout=PIPE,
["nix", "repl", "--file", "<nixpkgs/nixos>", "--nix-flag"]
)
n.repl("attr", 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=" \n/path/to/file\n "),
)
def test_nixos_build(mock_run: Any, monkeypatch: Any) -> None:
assert n.nixos_build("attr", None, None, nix_flag="foo") == Path("/path/to/file")
mock_run.assert_called_with(
["nix-build", "<nixpkgs/nixos>", "--attr", "attr", "--nix-flag", "foo"],
stdout=PIPE,
)
n.nixos_build("attr", "preAttr", "file")
mock_run.assert_called_with(
["nix-build", "file", "--attr", "preAttr.attr"],
stdout=PIPE,
)
n.nixos_build("attr", None, "file", no_out_link=True)
mock_run.assert_called_with(
["nix-build", "file", "--attr", "attr", "--no-out-link"],
stdout=PIPE,
)
n.nixos_build("attr", "preAttr", None, no_out_link=False, keep_going=True)
mock_run.assert_called_with(
["nix-build", "default.nix", "--attr", "preAttr.attr", "--keep-going"],
stdout=PIPE,
)
@patch(get_qualified_name(n.run_wrapper, n), autospec=True)
def test_repl_flake(mock_run: Any) -> None:
n.repl_flake("attr", m.Flake(Path("flake.nix"), "myAttr"), nix_flag=True)
# This method would be really annoying to test, and it is not that important
# So just check that we are at least calling it
assert mock_run.called
@patch(get_qualified_name(n.run_wrapper, n), autospec=True)
@@ -418,7 +569,8 @@ def test_switch_to_configuration(mock_run: Any, monkeypatch: Any) -> None:
Path("/nix/var/nix/profiles/per-user/root/channels/home-manager"),
],
)
def test_upgrade_channels(mock_glob: Any) -> None:
@patch(get_qualified_name(n.Path.is_dir, n), autospec=True, return_value=True)
def test_upgrade_channels(mock_is_dir: Any, mock_glob: Any) -> None:
with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run:
n.upgrade_channels(False)
mock_run.assert_called_with(["nix-channel", "--update", "nixos"], check=False)
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py
+20 -28
View File
@@ -1,4 +1,3 @@
from pathlib import Path
from typing import Any
from unittest.mock import patch
@@ -8,7 +7,7 @@ import nixos_rebuild.process as p
from .helpers import get_qualified_name
@patch(get_qualified_name(p.subprocess.run))
@patch(get_qualified_name(p.subprocess.run), autospec=True)
def test_run(mock_run: Any) -> None:
p.run_wrapper(["test", "--with", "flags"], check=True)
mock_run.assert_called_with(
@@ -40,12 +39,22 @@ def test_run(mock_run: Any) -> None:
)
p.run_wrapper(
["test", "--with", "flags"],
["test", "--with", "some flags"],
check=True,
remote=m.Remote("user@localhost", ["--ssh", "opt"], "password"),
)
mock_run.assert_called_with(
["ssh", "--ssh", "opt", "user@localhost", "--", "test", "--with", "flags"],
[
"ssh",
"--ssh",
"opt",
*p.SSH_DEFAULT_OPTS,
"user@localhost",
"--",
"test",
"--with",
"'some flags'",
],
check=True,
text=True,
errors="surrogateescape",
@@ -65,6 +74,7 @@ def test_run(mock_run: Any) -> None:
"ssh",
"--ssh",
"opt",
*p.SSH_DEFAULT_OPTS,
"user@localhost",
"--",
"sudo",
@@ -84,38 +94,20 @@ def test_run(mock_run: Any) -> None:
)
def test_remote_from_name(monkeypatch: Any, tmpdir: Path) -> None:
def test_remote_from_name(monkeypatch: Any) -> None:
monkeypatch.setenv("NIX_SSHOPTS", "")
assert m.Remote.from_arg("user@localhost", None, tmpdir) == m.Remote(
assert m.Remote.from_arg("user@localhost", None, False) == m.Remote(
"user@localhost",
opts=[
"-o",
"ControlMaster=auto",
"-o",
f"ControlPath={tmpdir / "ssh-%n"}",
"-o",
"ControlPersist=60",
],
opts=[],
sudo_password=None,
)
# get_qualified_name doesn't work because getpass is aliased to another
# function
with patch(f"{p.__name__}.getpass", return_value="password"):
monkeypatch.setenv("NIX_SSHOPTS", "-f foo -b bar")
assert m.Remote.from_arg("user@localhost", True, tmpdir) == m.Remote(
monkeypatch.setenv("NIX_SSHOPTS", "-f foo -b bar -t")
assert m.Remote.from_arg("user@localhost", True, True) == m.Remote(
"user@localhost",
opts=[
"-f",
"foo",
"-b",
"bar",
"-o",
"ControlMaster=auto",
"-o",
f"ControlPath={tmpdir / "ssh-%n"}",
"-o",
"ControlPersist=60",
],
opts=["-f", "foo", "-b", "bar", "-t"],
sudo_password="password",
)