fix: probe correct value

This commit is contained in:
eric
2026-03-18 03:21:10 +01:00
parent 9e0eb5b583
commit f150afec0a
7 changed files with 29 additions and 101 deletions

View File

@@ -1,30 +0,0 @@
{ lib, ... }:
{
# Generated by nodeiwest host init.
imports = [
./disko.nix
./hardware-configuration.nix
];
networking.hostName = "lab";
networking.useDHCP = lib.mkDefault true;
time.timeZone = "UTC";
boot.loader.efi.canTouchEfiVariables = true;
boot.loader.grub = {
enable = true;
efiSupport = true;
device = "nodev";
};
nodeiwest.ssh.userCAPublicKeys = [
"ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIE6c2oMkM7lLg9qWHVgbrFaFBDrrFyynFlPviiydQdFi openbao-user-ca"
];
nodeiwest.tailscale.openbao = {
enable = true;
};
system.stateVersion = "25.05";
}

View File

@@ -1,47 +0,0 @@
{
lib,
...
}:
{
# Generated by nodeiwest host init.
# Replace the disk only if the provider exposes a different primary device.
disko.devices = {
disk.main = {
type = "disk";
device = lib.mkDefault "/dev/sda 11";
content = {
type = "gpt";
partitions = {
ESP = {
priority = 1;
name = "ESP";
start = "1MiB";
end = "512MiB";
type = "EF00";
content = {
type = "filesystem";
format = "vfat";
mountpoint = "/boot";
mountOptions = [ "umask=0077" ];
};
};
swap = {
size = "4GiB";
content = {
type = "swap";
resumeDevice = true;
};
};
root = {
size = "100%";
content = {
type = "filesystem";
format = "ext4";
mountpoint = "/";
};
};
};
};
};
};
}

View File

@@ -1,5 +0,0 @@
{ ... }:
{
# Placeholder generated by nodeiwest host init.
# nixos-anywhere will replace this with the generated hardware config.
}

View File

@@ -584,7 +584,7 @@ def validate_host_name(name: str) -> None:
def probe_host(ip: str, user: str) -> ProbeFacts: def probe_host(ip: str, user: str) -> ProbeFacts:
lsblk_cmd = "lsblk -o NAME,SIZE,TYPE,MODEL,FSTYPE,PTTYPE,MOUNTPOINTS" lsblk_cmd = "lsblk -P -o NAME,SIZE,TYPE,MODEL,FSTYPE,PTTYPE,MOUNTPOINTS"
boot_cmd = "test -d /sys/firmware/efi && echo UEFI || echo BIOS" boot_cmd = "test -d /sys/firmware/efi && echo UEFI || echo BIOS"
root_cmd = "findmnt -no SOURCE /" root_cmd = "findmnt -no SOURCE /"
swap_cmd = "cat /proc/swaps" swap_cmd = "cat /proc/swaps"
@@ -635,28 +635,25 @@ def probe_host(ip: str, user: str) -> ProbeFacts:
def parse_lsblk_output(output: str) -> list[dict[str, str]]: def parse_lsblk_output(output: str) -> list[dict[str, str]]:
lines = [line.rstrip("\n") for line in output.splitlines() if line.strip()] lines = [line.strip() for line in output.splitlines() if line.strip()]
if len(lines) < 2: if not lines:
raise NodeiwestError("Unexpected lsblk output: not enough lines to parse.") raise NodeiwestError("Unexpected lsblk output: not enough lines to parse.")
header = lines[0]
columns = ["NAME", "SIZE", "TYPE", "MODEL", "FSTYPE", "PTTYPE", "MOUNTPOINTS"] columns = ["NAME", "SIZE", "TYPE", "MODEL", "FSTYPE", "PTTYPE", "MOUNTPOINTS"]
starts = []
for column in columns:
index = header.find(column)
if index == -1:
raise NodeiwestError(f"Unexpected lsblk output: missing {column} column.")
starts.append(index)
starts.append(len(header) + 20)
rows: list[dict[str, str]] = [] rows: list[dict[str, str]] = []
for line in lines[1:]: for line in lines:
row: dict[str, str] = {} tokens = shlex.split(line)
for idx, column in enumerate(columns): row = {}
start = starts[idx] for token in tokens:
end = starts[idx + 1] if idx + 1 < len(columns) else len(line) if "=" not in token:
row[column] = line[start:end].strip() continue
row["NAME"] = re.sub(r"^[^0-9A-Za-z]+", "", row["NAME"]) key, value = token.split("=", 1)
row[key] = value
missing = [column for column in columns if column not in row]
if missing:
raise NodeiwestError(
f"Unexpected lsblk output: missing columns {', '.join(missing)} in line {line!r}."
)
rows.append(row) rows.append(row)
return rows return rows

View File

@@ -46,6 +46,19 @@ class HelperCliTests(unittest.TestCase):
self.assertIn('device = lib.mkDefault "/dev/vda";', rendered) self.assertIn('device = lib.mkDefault "/dev/vda";', rendered)
self.assertIn('size = "8GiB";', rendered) self.assertIn('size = "8GiB";', rendered)
def test_parse_lsblk_output_reads_pairs_without_smearing_columns(self) -> None:
output = (
'NAME="sda" SIZE="11G" TYPE="disk" MODEL="QEMU HARDDISK" FSTYPE="" PTTYPE="gpt" MOUNTPOINTS=""\n'
'NAME="sda1" SIZE="512M" TYPE="part" MODEL="" FSTYPE="vfat" PTTYPE="" MOUNTPOINTS="/boot"\n'
)
rows = cli.parse_lsblk_output(output)
self.assertEqual(rows[0]["NAME"], "sda")
self.assertEqual(rows[0]["SIZE"], "11G")
self.assertEqual(rows[0]["MODEL"], "QEMU HARDDISK")
self.assertEqual(rows[1]["NAME"], "sda1")
self.assertEqual(rows[1]["MOUNTPOINTS"], "/boot")
def test_bao_kv_get_uses_explicit_kv_mount(self) -> None: def test_bao_kv_get_uses_explicit_kv_mount(self) -> None:
completed = mock.Mock() completed = mock.Mock()
completed.stdout = '{"data": {"data": {"CLIENT_ID": "x"}}}' completed.stdout = '{"data": {"data": {"CLIENT_ID": "x"}}}'