From 76e653f37f49e2a3ad1f3580f798786962d5755b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Thalheim?= Date: Fri, 4 Jul 2025 16:56:42 +0200 Subject: [PATCH] nixoTestLib: split setting up function for port-forwarding and setuping up flake --- checks/installation/flake-module.nix | 74 ++++++++++++++--------- pkgs/clan-cli/clan_lib/ssh/remote_test.py | 2 +- pkgs/testing/nixos_test_lib/nix_setup.py | 23 +++++++ pkgs/testing/nixos_test_lib/ssh.py | 30 ++++----- 4 files changed, 79 insertions(+), 50 deletions(-) diff --git a/checks/installation/flake-module.nix b/checks/installation/flake-module.nix index 163e6e653..473dde6c1 100644 --- a/checks/installation/flake-module.nix +++ b/checks/installation/flake-module.nix @@ -174,7 +174,8 @@ import tempfile import os import subprocess - from nixos_test_lib.ssh import setup_test_environment # type: ignore[import-untyped] + from nixos_test_lib.ssh import setup_ssh_connection # type: ignore[import-untyped] + from nixos_test_lib.nix_setup import prepare_test_flake # type: ignore[import-untyped] def create_test_machine(oldmachine, qemu_test_bin: str, **kwargs): """Create a new test machine from an installed disk image""" @@ -201,14 +202,20 @@ target.start() - # Set up test environment using common helper + # Set up test environment with tempfile.TemporaryDirectory() as temp_dir: - env = setup_test_environment( - target, - temp_dir, - "${closureInfo}", - "${../assets/ssh/privkey}", - "${self.checks.x86_64-linux.clan-core-for-checks}" + # Prepare test flake and Nix store + flake_dir = prepare_test_flake( + temp_dir, + "${self.checks.x86_64-linux.clan-core-for-checks}", + "${closureInfo}" + ) + + # Set up SSH connection + ssh_conn = setup_ssh_connection( + target, + temp_dir, + "${../assets/ssh/privkey}" ) # Run clan install from host using port forwarding @@ -218,10 +225,10 @@ "install", "--phases", "disko,install", "--debug", - "--flake", env.flake_dir, + "--flake", flake_dir, "--yes", "test-install-machine-without-system", - "--target-host", f"nonrootuser@localhost:{env.host_port}", - "-i", env.ssh_key, + "--target-host", f"nonrootuser@localhost:{ssh_conn.host_port}", + "-i", ssh_conn.ssh_key, "--option", "store", os.environ['CLAN_TEST_STORE'], "--update-hardware-config", "nixos-facter", ] @@ -254,29 +261,36 @@ import tempfile import os import subprocess - from nixos_test_lib.ssh import setup_test_environment # type: ignore[import-untyped] + from nixos_test_lib.ssh import setup_ssh_connection # type: ignore[import-untyped] + from nixos_test_lib.nix_setup import prepare_test_flake # type: ignore[import-untyped] target.start() - # Set up test environment using common helper + # Set up test environment with tempfile.TemporaryDirectory() as temp_dir: - env = setup_test_environment( - target, - temp_dir, - "${closureInfo}", - "${../assets/ssh/privkey}", - "${self.checks.x86_64-linux.clan-core-for-checks}" + # Prepare test flake and Nix store + flake_dir = prepare_test_flake( + temp_dir, + "${self.checks.x86_64-linux.clan-core-for-checks}", + "${closureInfo}" + ) + + # Set up SSH connection + ssh_conn = setup_ssh_connection( + target, + temp_dir, + "${../assets/ssh/privkey}" ) # Verify files don't exist initially - hw_config_file = os.path.join(env.flake_dir, "machines/test-install-machine/hardware-configuration.nix") - facter_file = os.path.join(env.flake_dir, "machines/test-install-machine/facter.json") + hw_config_file = os.path.join(flake_dir, "machines/test-install-machine/hardware-configuration.nix") + facter_file = os.path.join(flake_dir, "machines/test-install-machine/facter.json") assert not os.path.exists(hw_config_file), "hardware-configuration.nix should not exist initially" assert not os.path.exists(facter_file), "facter.json should not exist initially" # Set CLAN_FLAKE for the commands - os.environ["CLAN_FLAKE"] = env.flake_dir + os.environ["CLAN_FLAKE"] = flake_dir # Test facter backend clan_cmd = [ @@ -287,17 +301,17 @@ "--flake", ".", "--host-key-check", "none", "test-install-machine-without-system", - "-i", env.ssh_key, + "-i", ssh_conn.ssh_key, "--option", "store", os.environ['CLAN_TEST_STORE'], - f"nonrootuser@localhost:{env.host_port}" + f"nonrootuser@localhost:{ssh_conn.host_port}" ] - result = subprocess.run(clan_cmd, capture_output=True, cwd=env.flake_dir) + result = subprocess.run(clan_cmd, capture_output=True, cwd=flake_dir) if result.returncode != 0: print(f"Clan update-hardware-config failed: {result.stderr.decode()}") raise Exception(f"Clan update-hardware-config failed with return code {result.returncode}") - facter_without_system_file = os.path.join(env.flake_dir, "machines/test-install-machine-without-system/facter.json") + facter_without_system_file = os.path.join(flake_dir, "machines/test-install-machine-without-system/facter.json") assert os.path.exists(facter_without_system_file), "facter.json should exist after update" os.remove(facter_without_system_file) @@ -311,17 +325,17 @@ "--host-key-check", "none", "--flake", ".", "test-install-machine-without-system", - "-i", env.ssh_key, + "-i", ssh_conn.ssh_key, "--option", "store", os.environ['CLAN_TEST_STORE'], - f"nonrootuser@localhost:{env.host_port}" + f"nonrootuser@localhost:{ssh_conn.host_port}" ] - result = subprocess.run(clan_cmd, capture_output=True, cwd=env.flake_dir) + result = subprocess.run(clan_cmd, capture_output=True, cwd=flake_dir) if result.returncode != 0: print(f"Clan update-hardware-config (nixos-generate-config) failed: {result.stderr.decode()}") raise Exception(f"Clan update-hardware-config failed with return code {result.returncode}") - hw_config_without_system_file = os.path.join(env.flake_dir, "machines/test-install-machine-without-system/hardware-configuration.nix") + hw_config_without_system_file = os.path.join(flake_dir, "machines/test-install-machine-without-system/hardware-configuration.nix") assert os.path.exists(hw_config_without_system_file), "hardware-configuration.nix should exist after update" ''; } { inherit pkgs self; }; diff --git a/pkgs/clan-cli/clan_lib/ssh/remote_test.py b/pkgs/clan-cli/clan_lib/ssh/remote_test.py index 00b6ca53c..7eeb5feac 100644 --- a/pkgs/clan-cli/clan_lib/ssh/remote_test.py +++ b/pkgs/clan-cli/clan_lib/ssh/remote_test.py @@ -8,8 +8,8 @@ import pytest from clan_lib.async_run import AsyncRuntime from clan_lib.cmd import ClanCmdTimeoutError, Log, RunOpts from clan_lib.errors import ClanError, CmdOut -from clan_lib.ssh.remote import Remote from clan_lib.ssh.host_key import HostKeyCheck +from clan_lib.ssh.remote import Remote from clan_lib.ssh.sudo_askpass_proxy import SudoAskpassProxy if sys.platform == "darwin": diff --git a/pkgs/testing/nixos_test_lib/nix_setup.py b/pkgs/testing/nixos_test_lib/nix_setup.py index 8ec2b76cc..aeb4e35b7 100644 --- a/pkgs/testing/nixos_test_lib/nix_setup.py +++ b/pkgs/testing/nixos_test_lib/nix_setup.py @@ -71,3 +71,26 @@ def setup_nix_in_nix(closure_info: str | None) -> None: text=True, check=True, ) + + +def prepare_test_flake( + temp_dir: str, clan_core_for_checks: str, closure_info: str +) -> str: + """Set up Nix store and copy test flake to temporary directory + + Args: + temp_dir: Temporary directory + clan_core_for_checks: Path to clan-core-for-checks + closure_info: Path to closure info for Nix store setup + + Returns: + Path to the test flake directory + """ + # Set up Nix store + setup_nix_in_nix(closure_info) + + # Copy test flake + flake_dir = Path(temp_dir) / "test-flake" + subprocess.run(["cp", "-r", clan_core_for_checks, flake_dir], check=True) # noqa: S603, S607 + subprocess.run(["chmod", "-R", "+w", flake_dir], check=True) # noqa: S603, S607 + return str(flake_dir) diff --git a/pkgs/testing/nixos_test_lib/ssh.py b/pkgs/testing/nixos_test_lib/ssh.py index 97a52766c..dc51213ac 100644 --- a/pkgs/testing/nixos_test_lib/ssh.py +++ b/pkgs/testing/nixos_test_lib/ssh.py @@ -1,34 +1,31 @@ """SSH and test setup utilities""" -import subprocess from pathlib import Path from typing import NamedTuple -from .nix_setup import setup_nix_in_nix from .port import find_free_port, setup_port_forwarding -class TestEnvironment(NamedTuple): +class SSHConnection(NamedTuple): host_port: int ssh_key: str - flake_dir: str -def setup_test_environment( +def setup_ssh_connection( target, temp_dir: str, - closure_info: str, assets_ssh_privkey: str, - clan_core_for_checks: str, -) -> TestEnvironment: - """Set up common test environment including SSH, port forwarding, and flake setup +) -> SSHConnection: + """Set up SSH connection with port forwarding to test VM + + Args: + target: Test VM target + temp_dir: Temporary directory for SSH key + assets_ssh_privkey: Path to SSH private key asset Returns: - TestEnvironment with host_port, ssh_key, and flake_dir + SSHConnection with host_port and ssh_key path """ - # Run setup function - setup_nix_in_nix(closure_info) - host_port = find_free_port() target.wait_for_unit("sshd.service") target.wait_for_open_port(22) @@ -40,9 +37,4 @@ def setup_test_environment( f.write(src.read()) ssh_key.chmod(0o600) - # Copy test flake to temp directory - flake_dir = Path(temp_dir) / "test-flake" - subprocess.run(["cp", "-r", clan_core_for_checks, flake_dir], check=True) # noqa: S603, S607 - subprocess.run(["chmod", "-R", "+w", flake_dir], check=True) # noqa: S603, S607 - - return TestEnvironment(host_port, str(ssh_key), str(flake_dir)) + return SSHConnection(host_port, str(ssh_key))