yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

nixoTestLib: split setting up function for port-forwarding and setuping up flake

Browse Source
This commit is contained in:
Jörg Thalheim
2025-07-04 16:56:42 +02:00
parent 247151e93f
commit 76e653f37f
4 changed files with 79 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
checks/installation/flake-module.nix
View File

@@ -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; };

2
pkgs/clan-cli/clan_lib/ssh/remote_test.py
View File

@@ -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":

23
pkgs/testing/nixos_test_lib/nix_setup.py
View File

@@ -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)

30
pkgs/testing/nixos_test_lib/ssh.py
View File

@@ -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))