nixos-rebuild-ng: get remote hostname

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

@@ -177,8 +177,8 @@ def execute(argv: list[str]) -> None:
     tmpdir_path = Path(tmpdir.name)
 
     profile = Profile.from_name(args.profile_name)
-    flake = Flake.from_arg(args.flake)
     target_host = Ssh.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:
         upgrade_channels(bool(args.upgrade_all))

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

@@ -1,10 +1,13 @@
+from __future__ import annotations
+
 import os
 import platform
 import re
+import subprocess
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
-from typing import Any, ClassVar, Self, TypedDict, override
+from typing import Any, Callable, ClassVar, Self, TypedDict, override
 
 
 class NRError(Exception):
@@ -52,21 +55,41 @@ class Flake:
         return f"{self.path}#{self.attr}"
 
     @classmethod
-    def parse(cls, flake_str: str, hostname: str | None = None) -> Self:
+    def parse(
+        cls,
+        flake_str: str,
+        hostname_fn: Callable[[], str | None] = lambda: None,
+    ) -> Self:
         m = cls._re.match(flake_str)
         assert m is not None, f"got no matches for {flake_str}"
         attr = m.group("attr")
-        nixos_attr = f"nixosConfigurations.{attr or hostname or "default"}"
+        nixos_attr = f"nixosConfigurations.{attr or hostname_fn() or "default"}"
         return cls(Path(m.group("path")), nixos_attr)
 
     @classmethod
-    def from_arg(cls, flake_arg: Any) -> Self | None:
-        hostname = platform.node()
+    def from_arg(cls, flake_arg: Any, target_host: Ssh | 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"],
+                        check=True,
+                        capture_output=True,
+                        allow_tty=False,
+                        remote=target_host,
+                    ).stdout.strip()
+                except (AttributeError, subprocess.CalledProcessError):
+                    return None
+            else:
+                return platform.node()
+
         match flake_arg:
             case str(s):
-                return cls.parse(s, hostname)
+                return cls.parse(s, get_hostname)
             case True:
-                return cls.parse(".", hostname)
+                return cls.parse(".", get_hostname)
             case False:
                 return None
             case _:
@@ -76,7 +99,7 @@ class Flake:
                     # It can be a symlink to the actual flake.
                     if default_path.is_symlink():
                         default_path = default_path.readlink()
-                    return cls.parse(str(default_path.parent), hostname)
+                    return cls.parse(str(default_path.parent), get_hostname)
                 else:
                     return None
 

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

@@ -1,4 +1,5 @@
 import platform
+import subprocess
 from pathlib import Path
 from typing import Any
 from unittest.mock import patch
@@ -12,16 +13,15 @@ def test_flake_parse() -> None:
     assert m.Flake.parse("/path/to/flake#attr") == m.Flake(
         Path("/path/to/flake"), "nixosConfigurations.attr"
     )
-    assert m.Flake.parse("/path/ to /flake", "hostname") == m.Flake(
+    assert m.Flake.parse("/path/ to /flake", lambda: "hostname") == m.Flake(
         Path("/path/ to /flake"), "nixosConfigurations.hostname"
     )
-    assert m.Flake.parse("/path/to/flake", "hostname") == m.Flake(
+    assert m.Flake.parse("/path/to/flake", lambda: "hostname") == m.Flake(
         Path("/path/to/flake"), "nixosConfigurations.hostname"
     )
     assert m.Flake.parse(".#attr") == m.Flake(Path("."), "nixosConfigurations.attr")
     assert m.Flake.parse("#attr") == m.Flake(Path("."), "nixosConfigurations.attr")
-    assert m.Flake.parse(".", None) == m.Flake(Path("."), "nixosConfigurations.default")
-    assert m.Flake.parse("", "") == m.Flake(Path("."), "nixosConfigurations.default")
+    assert m.Flake.parse(".") == m.Flake(Path("."), "nixosConfigurations.default")
 
 
 @patch(get_qualified_name(platform.node), autospec=True)
@@ -29,15 +29,17 @@ def test_flake_from_arg(mock_node: Any) -> None:
     mock_node.return_value = "hostname"
 
     # Flake string
-    assert m.Flake.from_arg("/path/to/flake#attr") == m.Flake(
+    assert m.Flake.from_arg("/path/to/flake#attr", None) == m.Flake(
         Path("/path/to/flake"), "nixosConfigurations.attr"
     )
 
     # False
-    assert m.Flake.from_arg(False) is None
+    assert m.Flake.from_arg(False, None) is None
 
     # True
-    assert m.Flake.from_arg(True) == m.Flake(Path("."), "nixosConfigurations.hostname")
+    assert m.Flake.from_arg(True, None) == m.Flake(
+        Path("."), "nixosConfigurations.hostname"
+    )
 
     # None when we do not have /etc/nixos/flake.nix
     with patch(
@@ -45,7 +47,7 @@ def test_flake_from_arg(mock_node: Any) -> None:
         autospec=True,
         return_value=False,
     ):
-        assert m.Flake.from_arg(None) is None
+        assert m.Flake.from_arg(None, None) is None
 
     # None when we have a file in /etc/nixos/flake.nix
     with (
@@ -60,7 +62,7 @@ def test_flake_from_arg(mock_node: Any) -> None:
             return_value=False,
         ),
     ):
-        assert m.Flake.from_arg(None) == m.Flake(
+        assert m.Flake.from_arg(None, None) == m.Flake(
             Path("/etc/nixos"), "nixosConfigurations.hostname"
         )
 
@@ -81,10 +83,21 @@ def test_flake_from_arg(mock_node: Any) -> None:
             return_value=Path("/path/to/flake.nix"),
         ),
     ):
-        assert m.Flake.from_arg(None) == m.Flake(
+        assert m.Flake.from_arg(None, None) == m.Flake(
             Path("/path/to"), "nixosConfigurations.hostname"
         )
 
+    with (
+        patch(
+            get_qualified_name(m.subprocess.run),
+            autospec=True,
+            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"
+        )
+
 
 @patch(get_qualified_name(m.Path.mkdir, m), autospec=True)
 def test_profile_from_name(mock_mkdir: Any) -> None: