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

nixos-rebuild-ng: change copy closure logic when copying from_host -> to_host (#364698)

Browse Source
This commit is contained in:
Thiago Kenji Okada
2024-12-15 22:03:59 +00:00
committed by GitHub
parent 9f1c5f18b8 0212452e27
commit acb1fb7b03
9 changed files with 405 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkgs/by-name/ni/nixos-rebuild-ng/README.md
+36
View File
@@ -101,6 +101,42 @@ ruff check --fix .
ruff format .
```
## Breaking changes
While `nixos-rebuild-ng` tries to be as much of a clone of the original as
possible, there are still some breaking changes that were done in other to
improve the user experience. If they break your workflow in some way that is
not possible to fix, please open an issue and we can discuss a solution.
- For `--build-host` and `--target-host`, `nixos-rebuild-ng` does not allocate
a pseudo-TTY via SSH (e.g.: `ssh -t`) anymore. The reason for this is because
pseudo-TTY breaks some expectations from SSH, like it mangles stdout and
stderr, and can
[break terminal output](https://github.com/NixOS/nixpkgs/issues/336967) in
some situations.
The issue is that `sudo` needs a TTY to ask for password, otherwise it will
fail. The solution for this is a new flag, `--ask-sudo-password`, that when
used with `--target-host` (`--build-host` doesn't need `sudo`), will ask for
the `sudo` password for the target host using Python's
[getpass](https://docs.python.org/3/library/getpass.html) and forward it to
every `sudo` request. Keep in mind that there is no check, so if you type
your password wrong, it will fail during activation (this can be improved
though)
- When `--build-host` and `--target-host` is used together, we will use `nix
copy` (or 2 `nix-copy-closure` if you're using Nix <2.18) instead of SSH'ing
to build host and using `nix-copy-closure --to target-host`. The reason for
this is documented in PR
[#364698](https://github.com/NixOS/nixpkgs/pull/364698). If you do need the
previous behavior, you can simulate it using `ssh build-host --
nixos-rebuild-ng switch --target-host target-host`. If that is not the case,
please open an issue
- We do some additional validation of flags, like exiting with an error when
`--build-host` or `--target-host` is used with `repl`, since the user could
assume that the `repl` would be run remotely while it always run the local
machine. `nixos-rebuild` silently ignored those flags, so this
[may cause some issues](https://github.com/NixOS/nixpkgs/pull/363922) for
wrappers
## Caveats
- Bugs in the profile manipulation can cause corruption of your profile that
pkgs/by-name/ni/nixos-rebuild-ng/package.nix
+7 -8
View File
@@ -19,6 +19,11 @@
}:
let
executable = if withNgSuffix then "nixos-rebuild-ng" else "nixos-rebuild";
# This version is kind of arbitrary, we use some features that were
# implemented in newer versions of Nix, but not necessary 2.18.
# However, Lix is a fork of Nix 2.18, so this looks like a good version
# to cut specific functionality.
withNix218 = lib.versionAtLeast nix.version "2.18";
in
python3Packages.buildPythonApplication rec {
pname = "nixos-rebuild-ng";
@@ -30,10 +35,6 @@ python3Packages.buildPythonApplication rec {
setuptools
];
dependencies = with python3Packages; [
tabulate
];
nativeBuildInputs = lib.optionals withShellFiles [
installShellFiles
python3Packages.shtab
@@ -53,8 +54,9 @@ python3Packages.buildPythonApplication rec {
];
postPatch = ''
substituteInPlace nixos_rebuild/__init__.py \
substituteInPlace nixos_rebuild/constants.py \
--subst-var-by executable ${executable} \
--subst-var-by withNix218 ${lib.boolToString withNix218} \
--subst-var-by withReexec ${lib.boolToString withReexec} \
--subst-var-by withShellFiles ${lib.boolToString withShellFiles}
@@ -88,9 +90,6 @@ python3Packages.buildPythonApplication rec {
mypy
pytest
ruff
types-tabulate
# dependencies
tabulate
]
);
in
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py
+44 -45
View File
@@ -9,23 +9,18 @@ from subprocess import CalledProcessError, run
from typing import assert_never
from . import nix
from .constants import EXECUTABLE, WITH_NIX_2_18, WITH_REEXEC, WITH_SHELL_FILES
from .models import Action, BuildAttr, Flake, NRError, Profile
from .process import Remote, cleanup_ssh
from .utils import Args, LogFormatter
from .utils import Args, LogFormatter, tabulate
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# Build-time flags
# Strings to avoid breaking standalone (e.g.: `python -m nixos_rebuild`) usage
EXECUTABLE = "@executable@"
WITH_REEXEC = "@withReexec@"
WITH_SHELL_FILES = "@withShellFiles@"
def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentParser]]:
common_flags = argparse.ArgumentParser(add_help=False)
common_flags.add_argument("--verbose", "-v", action="count", default=0)
common_flags.add_argument("--verbose", "-v", action="count", dest="v", default=0)
common_flags.add_argument("--max-jobs", "-j")
common_flags.add_argument("--cores")
common_flags.add_argument("--log-format")
@@ -61,7 +56,13 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
copy_flags = argparse.ArgumentParser(add_help=False)
copy_flags.add_argument(
"--use-substitutes", "--substitute-on-destination", "-s", action="store_true"
"--use-substitutes",
"--substitute-on-destination",
"-s",
action="store_true",
# `-s` is the destination since it has the same meaning in
# `nix-copy-closure` and `nix copy`
dest="s",
)
sub_parsers = {
@@ -80,6 +81,9 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
allow_abbrev=False,
)
main_parser.add_argument("--help", "-h", action="store_true", help="Show manpage")
main_parser.add_argument(
"--debug", action="store_true", help="Enable debug logging"
)
main_parser.add_argument(
"--file", "-f", help="Enable and build the NixOS system from the specified file"
)
@@ -187,7 +191,7 @@ def parse_args(
}
if args.help or args.action is None:
if WITH_SHELL_FILES == "true":
if WITH_SHELL_FILES:
r = run(["man", "8", EXECUTABLE], check=False)
parser.exit(r.returncode)
else:
@@ -197,8 +201,8 @@ def parse_args(
def parser_warn(msg: str) -> None:
print(f"{parser.prog}: warning: {msg}", file=sys.stderr)
# This flag affects both nix and this script
if args.verbose:
# verbose affects both nix commands and this script, debug only this script
if args.v or args.debug:
logger.setLevel(logging.DEBUG)
# https://github.com/NixOS/nixpkgs/blob/master/pkgs/os-specific/linux/nixos-rebuild/nixos-rebuild.sh#L56
@@ -236,6 +240,8 @@ def parse_args(
Action.BUILD.value,
Action.DRY_BUILD.value,
Action.DRY_ACTIVATE.value,
Action.BUILD_VM.value,
Action.BUILD_VM_WITH_BOOTLOADER.value,
):
parser.error(
f"--target-host/--build-host is not supported with '{args.action}'"
@@ -281,6 +287,9 @@ def reexec(
def execute(argv: list[str]) -> None:
args, args_groups = parse_args(argv)
if not WITH_NIX_2_18:
logger.warning("you're using Nix <2.18, some features will not work correctly")
atexit.register(cleanup_ssh)
common_flags = vars(args_groups["common_flags"])
@@ -303,7 +312,7 @@ def execute(argv: list[str]) -> None:
# Re-exec to a newer version of the script before building to ensure we get
# the latest fixes
if (
WITH_REEXEC == "true"
WITH_REEXEC
and can_run
and not args.fast
and not os.environ.get("_NIXOS_REBUILD_REEXEC")
@@ -330,13 +339,21 @@ def execute(argv: list[str]) -> None:
| Action.BUILD
| Action.DRY_BUILD
| Action.DRY_ACTIVATE
| Action.BUILD_VM
| Action.BUILD_VM_WITH_BOOTLOADER
):
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:
case Action.BUILD_VM:
attr = "config.system.build.vm"
case Action.BUILD_VM_WITH_BOOTLOADER:
attr = "config.system.build.vmWithBootLoader"
case _:
attr = "config.system.build.toplevel"
match (action, rollback, build_host, flake):
case (Action.SWITCH | Action.BOOT, True, _, _):
@@ -387,10 +404,14 @@ def execute(argv: list[str]) -> None:
dry_run=dry_run,
**build_flags,
)
case m:
case never:
# should never happen, but mypy is not smart enough to
# handle this with assert_never
raise NRError(f"invalid match for build: {m}")
# https://github.com/python/mypy/issues/16650
# https://github.com/python/mypy/issues/16722
raise AssertionError(
f"expected code to be unreachable, but got: {never}"
)
if not rollback:
nix.copy_closure(
@@ -406,6 +427,7 @@ def execute(argv: list[str]) -> None:
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,
@@ -415,23 +437,12 @@ def execute(argv: list[str]) -> None:
specialisation=args.specialisation,
install_bootloader=args.install_bootloader,
)
case Action.BUILD_VM | Action.BUILD_VM_WITH_BOOTLOADER:
logger.info("building the system configuration...")
attr = "vm" if action == Action.BUILD_VM else "vmWithBootLoader"
if flake:
path_to_config = nix.build_flake(
f"config.system.build.{attr}",
flake,
**flake_build_flags,
elif action in (Action.BUILD_VM, Action.BUILD_VM_WITH_BOOTLOADER):
# If you get `not-found`, please open an issue
vm_path = next(path_to_config.glob("bin/run-*-vm"), "not-found")
print(
f"Done. The virtual machine can be started by running '{vm_path}'"
)
else:
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")
print(f"Done. The virtual machine can be started by running '{vm_path}'")
case Action.EDIT:
nix.edit(flake, **flake_build_flags)
case Action.DRY_RUN:
@@ -441,8 +452,6 @@ def execute(argv: list[str]) -> None:
if args.json:
print(json.dumps(generations, indent=2))
else:
from tabulate import tabulate
headers = {
"generation": "Generation",
"date": "Build-date",
@@ -452,17 +461,7 @@ def execute(argv: list[str]) -> None:
"specialisations": "Specialisation",
"current": "Current",
}
# Not exactly the same format as legacy nixos-rebuild but close
# enough
table = tabulate(
generations,
headers=headers,
tablefmt="plain",
numalign="left",
stralign="left",
disable_numparse=True,
)
print(table)
print(tabulate(generations, headers=headers))
case Action.REPL:
if flake:
nix.repl_flake("toplevel", flake, **flake_build_flags)
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/constants.py
+9
View File
@@ -0,0 +1,9 @@
# Build-time flags
# Use strings to avoid breaking standalone (e.g.: `python -m nixos_rebuild`)
# usage
EXECUTABLE = "@executable@"
# Use either `== "true"` if the default (e.g.: `python -m nixos_rebuld`) is
# `False` or `!= "false"` if the default is `True`
WITH_NIX_2_18 = "@withNix218@" != "false" # type: ignore
WITH_REEXEC = "@withReexec@" == "true" # type: ignore
WITH_SHELL_FILES = "@withShellFiles@" == "true" # type: ignore
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/nix.py
+129 -95
View File
@@ -1,5 +1,6 @@
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from importlib.resources import files
from pathlib import Path
@@ -7,6 +8,7 @@ from string import Template
from subprocess import PIPE, CalledProcessError
from typing import Final
from .constants import WITH_NIX_2_18
from .models import (
Action,
BuildAttr,
@@ -139,30 +141,56 @@ def copy_closure(
"""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" if to_host else "--from",
host.host,
closure,
],
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,
)
extra_env = {
"NIX_SSHOPTS": " ".join(filter(lambda x: x, [*SSH_DEFAULT_OPTS, sshopts]))
}
def nix_copy_closure(host: Remote, to: bool) -> None:
run_wrapper(
[
"nix-copy-closure",
*dict_to_flags(copy_flags),
"--to" if to else "--from",
host.host,
closure,
],
extra_env=extra_env,
)
def nix_copy(to_host: Remote, from_host: Remote) -> None:
run_wrapper(
[
"nix",
"copy",
"--from",
f"ssh://{from_host.host}",
"--to",
f"ssh://{to_host.host}",
closure,
],
extra_env=extra_env,
)
match (to_host, from_host):
case (None, None):
return
case (Remote(_) as host, None) | (None, Remote(_) as host):
nix_copy_closure(host, to=bool(to_host))
case (Remote(_), Remote(_)):
if WITH_NIX_2_18:
# With newer Nix, use `nix copy` instead of `nix-copy-closure`
# since it supports `--to` and `--from` at the same time
# TODO: once we drop Nix 2.3 from nixpkgs, remove support for
# `nix-copy-closure`
nix_copy(to_host, from_host)
else:
# With older Nix, we need to copy from to local and local to
# host. This means it is slower and need additional disk space
# in local
nix_copy_closure(from_host, to=False)
nix_copy_closure(to_host, to=True)
def edit(flake: Flake | None, **flake_flags: Args) -> None:
@@ -218,7 +246,8 @@ def get_nixpkgs_rev(nixpkgs_path: Path | None) -> str | None:
r = run_wrapper(
["git", "-C", nixpkgs_path, "rev-parse", "--short", "HEAD"],
check=False,
stdout=PIPE,
# https://github.com/NixOS/nixpkgs/issues/365222
capture_output=True,
)
except FileNotFoundError:
# Git is not included in the closure so we need to check
@@ -237,68 +266,72 @@ def get_nixpkgs_rev(nixpkgs_path: Path | None) -> str | None:
return None
def _parse_generation_from_nix_store(path: Path, profile: Profile) -> Generation:
entry_id = path.name.split("-")[1]
current = path.name == profile.path.readlink().name
timestamp = datetime.fromtimestamp(path.stat().st_ctime).strftime(
"%Y-%m-%d %H:%M:%S"
)
return Generation(
id=int(entry_id),
timestamp=timestamp,
current=current,
)
def _parse_generation_from_nix_env(line: str) -> Generation:
parts = line.split()
entry_id = parts[0]
timestamp = f"{parts[1]} {parts[2]}"
current = "(current)" in parts
return Generation(
id=int(entry_id),
timestamp=timestamp,
current=current,
)
def get_generations(
profile: Profile,
target_host: Remote | None = None,
using_nix_env: bool = False,
sudo: bool = False,
) -> list[Generation]:
def get_generations(profile: Profile) -> list[Generation]:
"""Get all NixOS generations from profile.
Includes generation ID (e.g.: 1, 2), timestamp (e.g.: when it was created)
and if this is the current active profile or not.
If `lock_profile = True` this command will need root to run successfully.
"""
if not profile.path.exists():
raise NRError(f"no profile '{profile.name}' found")
result = []
if using_nix_env:
# Using `nix-env --list-generations` needs root to lock the profile
# TODO: do we actually need to lock profile for e.g.: rollback?
# https://github.com/NixOS/nix/issues/5144
r = run_wrapper(
["nix-env", "-p", profile.path, "--list-generations"],
stdout=PIPE,
remote=target_host,
sudo=sudo,
def parse_path(path: Path, profile: Profile) -> Generation:
entry_id = path.name.split("-")[1]
current = path.name == profile.path.readlink().name
timestamp = datetime.fromtimestamp(path.stat().st_ctime).strftime(
"%Y-%m-%d %H:%M:%S"
)
for line in r.stdout.splitlines():
result.append(_parse_generation_from_nix_env(line))
else:
assert not target_host, "target_host is not supported when using_nix_env=False"
for p in profile.path.parent.glob("system-*-link"):
result.append(_parse_generation_from_nix_store(p, profile))
return sorted(result, key=lambda d: d.id)
return Generation(
id=int(entry_id),
timestamp=timestamp,
current=current,
)
return sorted(
[parse_path(p, profile) for p in profile.path.parent.glob("system-*-link")],
key=lambda d: d.id,
)
def get_generations_from_nix_env(
profile: Profile,
target_host: Remote | None = None,
sudo: bool = False,
) -> list[Generation]:
"""Get all NixOS generations from profile with nix-env. Needs root.
Includes generation ID (e.g.: 1, 2), timestamp (e.g.: when it was created)
and if this is the current active profile or not.
"""
if not profile.path.exists():
raise NRError(f"no profile '{profile.name}' found")
# Using `nix-env --list-generations` needs root to lock the profile
r = run_wrapper(
["nix-env", "-p", profile.path, "--list-generations"],
stdout=PIPE,
remote=target_host,
sudo=sudo,
)
def parse_line(line: str) -> Generation:
parts = line.split()
entry_id = parts[0]
timestamp = f"{parts[1]} {parts[2]}"
current = "(current)" in parts
return Generation(
id=int(entry_id),
timestamp=timestamp,
current=current,
)
return sorted(
[parse_line(line) for line in r.stdout.splitlines()],
key=lambda d: d.id,
)
def list_generations(profile: Profile) -> list[GenerationJson]:
@@ -310,9 +343,8 @@ def list_generations(profile: Profile) -> list[GenerationJson]:
Will be formatted in a way that is expected by the output of
`nixos-rebuild list-generations --json`.
"""
generations = get_generations(profile)
result = []
for generation in reversed(generations):
def get_generation_info(generation: Generation) -> GenerationJson:
generation_path = (
profile.path.parent / f"{profile.path.name}-{generation.id}-link"
)
@@ -340,19 +372,24 @@ def list_generations(profile: Profile) -> list[GenerationJson]:
logger.debug("could not get configuration revision: %s", ex)
configuration_revision = "Unknown"
result.append(
GenerationJson(
generation=generation.id,
date=generation.timestamp,
nixosVersion=nixos_version,
kernelVersion=kernel_version,
configurationRevision=configuration_revision,
specialisations=specialisations,
current=generation.current,
)
return GenerationJson(
generation=generation.id,
date=generation.timestamp,
nixosVersion=nixos_version,
kernelVersion=kernel_version,
configurationRevision=configuration_revision,
specialisations=specialisations,
current=generation.current,
)
return result
# This can be surprisingly slow, especially with lots of generations,
# but it is basically IO work so we can run in parallel
with ThreadPoolExecutor() as executor:
return sorted(
executor.map(get_generation_info, get_generations(profile)),
key=lambda x: x["generation"],
reverse=True,
)
def repl(attr: str, build_attr: BuildAttr, **nix_flags: Args) -> None:
@@ -404,11 +441,8 @@ def rollback_temporary_profile(
sudo: bool,
) -> Path | None:
"Rollback a temporary Nix profile, like one created by `nixos-rebuild test`."
generations = get_generations(
profile,
target_host=target_host,
using_nix_env=True,
sudo=sudo,
generations = get_generations_from_nix_env(
profile, target_host=target_host, sudo=sudo
)
previous_gen_id = None
for generation in generations:
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/utils.py
+57 -3
View File
@@ -1,5 +1,6 @@
import logging
from typing import TypeAlias, override
from collections.abc import Mapping, Sequence
from typing import Any, TypeAlias, assert_never, override
Args: TypeAlias = bool | str | list[str] | int | None
@@ -18,17 +19,22 @@ class LogFormatter(logging.Formatter):
return formatter.format(record)
def dict_to_flags(d: dict[str, Args]) -> list[str]:
def dict_to_flags(d: Mapping[str, Args]) -> list[str]:
flags = []
for key, value in d.items():
flag = f"--{'-'.join(key.split('_'))}"
match value:
case None | False | 0 | []:
continue
case True if len(key) == 1:
flags.append(f"-{key}")
case True:
flags.append(flag)
case int() if len(key) == 1:
flags.append(f"-{key * value}")
case int():
flags.append(f"-{key[0] * value}")
for i in range(value):
flags.append(flag)
case str():
flags.append(flag)
flags.append(value)
@@ -36,4 +42,52 @@ def dict_to_flags(d: dict[str, Args]) -> list[str]:
flags.append(flag)
for v in value:
flags.append(v)
case _:
assert_never(value)
return flags
def remap_dicts(
dicts: Sequence[Mapping[str, Any]],
mappings: Mapping[str, str],
) -> list[dict[str, Any]]:
return [{mappings.get(k, k): v for k, v in d.items()} for d in dicts]
def tabulate(
data: Sequence[Mapping[str, Any]],
headers: Mapping[str, str] | None = None,
) -> str:
"""Convert a sequence of mappings in a tabular-style format for terminal.
It expects that all mappings (dicts) have the same keys as the first one,
otherwise it will misbehave.
"""
if not data:
return ""
if headers:
data = remap_dicts(data, headers)
data_headers = list(data[0].keys())
column_widths = [
max(
len(str(header)),
*(len(str(row.get(header, ""))) for row in data),
)
for header in data_headers
]
def format_row(row: Mapping[str, Any]) -> str:
s = (2 * " ").join(
f"{str(row[header]).ljust(width)}"
for header, width in zip(data_headers, column_widths)
)
return s.strip()
result = [format_row(dict(zip(data_headers, data_headers)))]
for row in data:
result.append(format_row(row))
return "\n".join(result)
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_main.py
+3 -3
View File
@@ -67,12 +67,12 @@ def test_parse_args() -> None:
]
)
assert nr.logger.level == logging.DEBUG
assert r2.verbose == 3
assert r2.v == 3
assert r2.flake is False
assert r2.action == "dry-build"
assert r2.file == "foo"
assert r2.attr == "bar"
assert g2["common_flags"].verbose == 3
assert g2["common_flags"].v == 3
@patch.dict(nr.process.os.environ, {}, clear=True)
@@ -109,7 +109,7 @@ def test_execute_nix_boot(mock_run: Any, tmp_path: Path) -> None:
call(
["git", "-C", nixpkgs_path, "rev-parse", "--short", "HEAD"],
check=False,
stdout=PIPE,
capture_output=True,
**DEFAULT_RUN_KWARGS,
),
call(
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py
+93 -46
View File
@@ -112,7 +112,6 @@ def test_remote_build(mock_run: Any, monkeypatch: Any) -> None:
extra_env={
"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS + ["--ssh opts"])
},
remote=None,
),
call(
["nix-store", "--realise", Path("/path/to/file"), "--build"],
@@ -166,7 +165,6 @@ def test_remote_build_flake(mock_run: Any, monkeypatch: Any) -> None:
extra_env={
"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS + ["--ssh opts"])
},
remote=None,
),
call(
[
@@ -185,37 +183,66 @@ def test_remote_build_flake(mock_run: Any, monkeypatch: Any) -> None:
)
@patch(get_qualified_name(n.run_wrapper, n), autospec=True)
def test_copy_closure(mock_run: Any, monkeypatch: Any) -> None:
def test_copy_closure(monkeypatch: Any) -> None:
closure = Path("/path/to/closure")
n.copy_closure(closure, None)
mock_run.assert_not_called()
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", [], 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@target.host", closure],
extra_env={"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS)},
remote=None,
)
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],
extra_env={"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS)},
)
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,
)
with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run:
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"])
},
)
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"},
)
monkeypatch.setattr(n, "WITH_NIX_2_18", True)
extra_env = {
"NIX_SSHOPTS": " ".join(p.SSH_DEFAULT_OPTS + ["--ssh build-target-opt"])
}
with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run:
n.copy_closure(closure, target_host, build_host)
mock_run.assert_called_with(
[
"nix",
"copy",
"--from",
"ssh://user@build.host",
"--to",
"ssh://user@target.host",
closure,
],
extra_env=extra_env,
)
monkeypatch.setattr(n, "WITH_NIX_2_18", False)
with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run:
n.copy_closure(closure, target_host, build_host)
mock_run.assert_has_calls(
[
call(
["nix-copy-closure", "--from", "user@build.host", closure],
extra_env=extra_env,
),
call(
["nix-copy-closure", "--to", "user@target.host", closure],
extra_env=extra_env,
),
]
)
@patch(get_qualified_name(n.run_wrapper, n), autospec=True)
@@ -262,14 +289,14 @@ def test_get_nixpkgs_rev() -> None:
mock_run.assert_called_with(
["git", "-C", path, "rev-parse", "--short", "HEAD"],
check=False,
stdout=PIPE,
capture_output=True,
)
expected_calls = [
call(
["git", "-C", path, "rev-parse", "--short", "HEAD"],
check=False,
stdout=PIPE,
capture_output=True,
),
call(
["git", "-C", path, "diff", "--quiet"],
@@ -300,7 +327,7 @@ def test_get_nixpkgs_rev() -> None:
mock_run.assert_has_calls(expected_calls)
def test_get_generations_from_nix_store(tmp_path: Path) -> None:
def test_get_generations(tmp_path: Path) -> None:
nixos_path = tmp_path / "nixos-system"
nixos_path.mkdir()
@@ -310,20 +337,17 @@ def test_get_generations_from_nix_store(tmp_path: Path) -> None:
(tmp_path / "system-3-link").symlink_to(nixos_path)
(tmp_path / "system-2-link").symlink_to(nixos_path)
assert n.get_generations(
m.Profile("system", tmp_path / "system"),
using_nix_env=False,
) == [
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),
]
@patch(
get_qualified_name(n.run_wrapper, n),
autospec=True,
return_value=CompletedProcess(
def test_get_generations_from_nix_env(tmp_path: Path) -> None:
path = tmp_path / "test"
path.touch()
return_value = CompletedProcess(
[],
0,
stdout=textwrap.dedent("""\
@@ -331,17 +355,40 @@ def test_get_generations_from_nix_store(tmp_path: Path) -> None:
2083 2024-11-07 22:59:41
2084 2024-11-07 23:54:17 (current)
"""),
),
)
def test_get_generations_from_nix_env(mock_run: Any, tmp_path: Path) -> None:
path = tmp_path / "test"
path.touch()
)
assert n.get_generations(m.Profile("system", path), using_nix_env=True) == [
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"),
]
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,
sudo=False,
)
remote = m.Remote("user@host", [], "password")
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, True
) == [
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,
sudo=True,
)
@patch(
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_utils.py
+27 -1
View File
@@ -1,3 +1,5 @@
import textwrap
import nixos_rebuild.utils as u
@@ -9,7 +11,9 @@ def test_dict_to_flags() -> None:
"test_flag_3": "value",
"test_flag_4": ["v1", "v2"],
"test_flag_5": None,
"verbose": 5,
"t": True,
"v": 5,
"verbose": 2,
}
)
assert r1 == [
@@ -19,7 +23,29 @@ def test_dict_to_flags() -> None:
"--test-flag-4",
"v1",
"v2",
"-t",
"-vvvvv",
"--verbose",
"--verbose",
]
r2 = u.dict_to_flags({"verbose": 0, "empty_list": []})
assert r2 == []
def test_remap_dicts() -> None:
assert u.remap_dicts(
[{"foo": 1, "bar": True}, {"qux": "keep"}],
{"foo": "Foo", "bar": "Bar"},
) == [{"Foo": 1, "Bar": True}, {"qux": "keep"}]
def test_tabulate() -> None:
assert u.tabulate([]) == ""
assert u.tabulate([{}]) == "\n"
assert u.tabulate(
[{"foo": 12345, "bar": ["abc", "cde"]}, {"foo": 345, "bar": 456}],
{"foo": "Foo", "bar": "Bar"},
) == textwrap.dedent("""\
Foo Bar
12345 ['abc', 'cde']
345 456""")