dev/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

nixos-rebuild-ng: move models.Ssh to process.Remote

Browse Source
This commit is contained in:
Thiago Kenji Okada 2024-11-24 20:51:21 +00:00
parent 37d6a2688f
commit 866e1786e3
7 changed files with 48 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -7,7 +7,7 @@ from subprocess import run
from tempfile import TemporaryDirectory
from typing import assert_never
from .models import Action, Flake, NRError, Profile, Ssh
from .models import Action, Flake, NRError, Profile
from .nix import (
copy_closure,
edit,
@ -22,6 +22,7 @@ from .nix import (
switch_to_configuration,
upgrade_channels,
)
from .process import Remote
from .utils import flags_to_dict, info
VERBOSE = 0
@ -177,7 +178,7 @@ def execute(argv: list[str]) -> None:
tmpdir_path = Path(tmpdir.name)
profile = Profile.from_name(args.profile_name)
target_host = Ssh.from_arg(args.target_host, not args.no_ssh_tty, tmpdir_path)
target_host = Remote.from_arg(args.target_host, not args.no_ssh_tty, tmpdir_path)
flake = Flake.from_arg(args.flake, target_host)
if args.upgrade or args.upgrade_all:

29
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/models.py
View File

@ -1,6 +1,5 @@
from __future__ import annotations
import os
import platform
import re
import subprocess
@ -9,6 +8,8 @@ from enum import Enum
from pathlib import Path
from typing import Any, Callable, ClassVar, Self, TypedDict, override
from .process import Remote, run_wrapper
class NRError(Exception):
"nixos-rebuild general error."
@ -67,11 +68,9 @@ class Flake:
return cls(Path(m.group("path")), nixos_attr)
@classmethod
def from_arg(cls, flake_arg: Any, target_host: Ssh | None) -> Self | None:
def from_arg(cls, flake_arg: Any, target_host: Remote | None) -> Self | None:
def get_hostname() -> str | None:
if target_host:
from .process import run_wrapper # circumvent circular import
try:
return run_wrapper(
["uname", "-n"],
@ -136,25 +135,3 @@ class Profile:
path = Path("/nix/var/nix/profiles/system-profiles") / name
path.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
return cls(name, path)
@dataclass(frozen=True)
class Ssh:
host: str
opts: list[str]
tty: bool
@classmethod
def from_arg(cls, host: str | None, tty: bool | None, tmp_dir: Path) -> Self | None:
if host:
opts = os.getenv("NIX_SSHOPTS", "").split() + [
"-o",
"ControlMaster=auto",
"-o",
f"ControlPath={tmp_dir / "ssh-%n"}",
"-o",
"ControlPersist=60",
]
return cls(host, opts, bool(tty))
else:
return None

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

@ -12,7 +12,7 @@ from .models import (
GenerationJson,
NRError,
Profile,
Ssh,
Remote,
)
from .process import run_wrapper
from .utils import Args, dict_to_flags, info
@ -22,7 +22,7 @@ FLAKE_FLAGS: Final = ["--extra-experimental-features", "nix-command flakes"]
def copy_closure(
closure: Path,
target_host: Ssh | None,
target_host: Remote | None,
**copy_flags: Args,
) -> None:
host = target_host
@ -303,7 +303,7 @@ def rollback_temporary_profile(profile: Profile) -> Path | None:
def set_profile(
profile: Profile,
path_to_config: Path,
target_host: Ssh | None,
target_host: Remote | None,
sudo: bool,
) -> None:
"Set a path as the current active Nix profile."
@ -318,7 +318,7 @@ def set_profile(
def switch_to_configuration(
path_to_config: Path,
action: Action,
target_host: Ssh | None,
target_host: Remote | None,
sudo: bool,
install_bootloader: bool = False,
specialisation: str | None = None,

30
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py
View File

@ -2,9 +2,9 @@ from __future__ import annotations
import os
import subprocess
from typing import Sequence, TypedDict, Unpack
from .models import Ssh
from dataclasses import dataclass
from pathlib import Path
from typing import Self, Sequence, TypedDict, Unpack
# Not exhaustive, but we can always extend it later.
@ -16,6 +16,28 @@ class RunKwargs(TypedDict, total=False):
stdout: int | None
@dataclass(frozen=True)
class Remote:
host: str
opts: list[str]
tty: bool
@classmethod
def from_arg(cls, host: str | None, tty: bool | None, tmp_dir: Path) -> Self | None:
if host:
opts = os.getenv("NIX_SSHOPTS", "").split() + [
"-o",
"ControlMaster=auto",
"-o",
f"ControlPath={tmp_dir / "ssh-%n"}",
"-o",
"ControlPersist=60",
]
return cls(host, opts, bool(tty))
else:
return None
def run_wrapper(
args: Sequence[str | bytes | os.PathLike[str] | os.PathLike[bytes]],
*,
@ -23,7 +45,7 @@ def run_wrapper(
env: dict[str, str] | None = None, # replaces the current environment
extra_env: dict[str, str] | None = None, # appends to the current environment
allow_tty: bool = True,
remote: Ssh | None = None,
remote: Remote | None = None,
sudo: bool = False,
**kwargs: Unpack[RunKwargs],
) -> subprocess.CompletedProcess[str]:

12
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_models.py
View File

@ -94,9 +94,9 @@ def test_flake_from_arg(mock_node: Any) -> None:
return_value=subprocess.CompletedProcess([], 0, "remote-hostname\n"),
),
):
assert m.Flake.from_arg("/path/to", m.Ssh("user@host", [], False)) == m.Flake(
Path("/path/to"), "nixosConfigurations.remote-hostname"
)
assert m.Flake.from_arg(
"/path/to", m.Remote("user@host", [], False)
) == m.Flake(Path("/path/to"), "nixosConfigurations.remote-hostname")
@patch(get_qualified_name(m.Path.mkdir, m), autospec=True)
@ -113,9 +113,9 @@ def test_profile_from_name(mock_mkdir: Any) -> None:
mock_mkdir.assert_called_once()
def test_ssh_from_name(monkeypatch: Any, tmpdir: Path) -> None:
def test_remote_from_name(monkeypatch: Any, tmpdir: Path) -> None:
monkeypatch.setenv("NIX_SSHOPTS", "")
assert m.Ssh.from_arg("user@localhost", None, tmpdir) == m.Ssh(
assert m.Remote.from_arg("user@localhost", None, tmpdir) == m.Remote(
"user@localhost",
[
"-o",
@ -129,7 +129,7 @@ def test_ssh_from_name(monkeypatch: Any, tmpdir: Path) -> None:
)
monkeypatch.setenv("NIX_SSHOPTS", "-f foo -b bar")
assert m.Ssh.from_arg("user@localhost", True, tmpdir) == m.Ssh(
assert m.Remote.from_arg("user@localhost", True, tmpdir) == m.Remote(
"user@localhost",
[
"-f",

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

@ -18,7 +18,7 @@ def test_copy_closure(mock_run: Any) -> None:
n.copy_closure(closure, None)
mock_run.assert_not_called()
target_host = m.Ssh("user@host", ["--ssh", "opt"], False)
target_host = m.Remote("user@host", ["--ssh", "opt"], False)
n.copy_closure(closure, target_host)
mock_run.assert_called_with(
["nix-copy-closure", "--to", "user@host", closure],
@ -366,7 +366,7 @@ def test_switch_to_configuration(mock_run: Any, monkeypatch: Any) -> None:
Path("/path/to/config"),
m.Action.TEST,
sudo=True,
target_host=m.Ssh("user@localhost", [], False),
target_host=m.Remote("user@localhost", [], False),
install_bootloader=True,
specialisation="special",
)
@ -378,7 +378,7 @@ def test_switch_to_configuration(mock_run: Any, monkeypatch: Any) -> None:
extra_env={"NIXOS_INSTALL_BOOTLOADER": "1"},
check=True,
sudo=True,
remote=m.Ssh("user@localhost", [], False),
remote=m.Remote("user@localhost", [], False),
)

6
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py
View File

@ -41,7 +41,7 @@ def test_run(mock_run: Any) -> None:
p.run_wrapper(
["test", "--with", "flags"],
check=True,
remote=m.Ssh("user@localhost", ["--ssh", "opt"], False),
remote=m.Remote("user@localhost", ["--ssh", "opt"], False),
)
mock_run.assert_called_with(
["ssh", "--ssh", "opt", "user@localhost", "--", "test", "--with", "flags"],
@ -69,7 +69,7 @@ def test_run(mock_run: Any) -> None:
check=True,
sudo=True,
extra_env={"FOO": "bar"},
remote=m.Ssh("user@localhost", ["--ssh", "opt"], True),
remote=m.Remote("user@localhost", ["--ssh", "opt"], True),
)
mock_run.assert_called_with(
[
@ -97,5 +97,5 @@ def test_run(mock_run: Any) -> None:
["test", "--with", "flags"],
check=False,
env={"foo": "bar"},
remote=m.Ssh("user@localhost", [], False),
remote=m.Remote("user@localhost", [], False),
)