move nixosTestLib to pkgs/testing
This commit is contained in:
Jörg Thalheim
@@ -167,7 +167,7 @@
|
||||
name = "installation";
|
||||
nodes.target = (import ./test-helpers.nix { inherit lib pkgs self; }).target;
|
||||
extraPythonPackages = _p: [
|
||||
(import ./test-helpers.nix { inherit lib pkgs self; }).nixosTestLib
|
||||
self.legacyPackages.${pkgs.system}.nixosTestLib
|
||||
];
|
||||
|
||||
testScript = ''
|
||||
@@ -225,7 +225,7 @@
|
||||
name = "update-hardware-configuration";
|
||||
nodes.target = (import ./test-helpers.nix { inherit lib pkgs self; }).target;
|
||||
extraPythonPackages = _p: [
|
||||
(import ./test-helpers.nix { inherit lib pkgs self; }).nixosTestLib
|
||||
self.legacyPackages.${pkgs.system}.nixosTestLib
|
||||
];
|
||||
|
||||
testScript = ''
|
||||
|
||||
@@ -1,3 +0,0 @@
|
||||
"""NixOS test library for clan VM testing"""
|
||||
|
||||
__version__ = "1.0.0"
|
||||
@@ -1,13 +0,0 @@
|
||||
"""Environment setup utilities for VM tests"""
|
||||
|
||||
import os
|
||||
|
||||
|
||||
def setup_nix_environment(temp_dir: str, closure_info: str) -> None:
|
||||
"""Set up nix chroot store environment"""
|
||||
if "NIX_REMOTE" in os.environ:
|
||||
del os.environ["NIX_REMOTE"] # we don't have any nix daemon running
|
||||
|
||||
os.environ["TMPDIR"] = temp_dir
|
||||
# Set NIX_CONFIG globally to disable substituters for speed
|
||||
os.environ["NIX_CONFIG"] = "substituters = \ntrusted-public-keys = "
|
||||
@@ -1,70 +0,0 @@
|
||||
"""Nix store setup utilities for VM tests"""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def setup_nix_in_nix(closure_info: str) -> None:
|
||||
"""Set up Nix store inside test environment
|
||||
|
||||
Args:
|
||||
closure_info: Path to closure info directory containing store-paths file
|
||||
"""
|
||||
tmpdir = os.environ.get("TMPDIR", "/tmp")
|
||||
|
||||
# Remove NIX_REMOTE if present (we don't have any nix daemon running)
|
||||
if "NIX_REMOTE" in os.environ:
|
||||
del os.environ["NIX_REMOTE"]
|
||||
|
||||
# Set NIX_CONFIG globally to disable substituters for speed
|
||||
os.environ["NIX_CONFIG"] = "substituters = \ntrusted-public-keys = "
|
||||
|
||||
# Set up environment variables for test environment
|
||||
os.environ["HOME"] = tmpdir
|
||||
os.environ["NIX_STATE_DIR"] = f"{tmpdir}/nix"
|
||||
os.environ["NIX_CONF_DIR"] = f"{tmpdir}/etc"
|
||||
os.environ["IN_NIX_SANDBOX"] = "1"
|
||||
os.environ["CLAN_TEST_STORE"] = f"{tmpdir}/store"
|
||||
os.environ["LOCK_NIX"] = f"{tmpdir}/nix_lock"
|
||||
|
||||
# Create necessary directories
|
||||
Path(f"{tmpdir}/nix").mkdir(parents=True, exist_ok=True)
|
||||
Path(f"{tmpdir}/etc").mkdir(parents=True, exist_ok=True)
|
||||
Path(f"{tmpdir}/store").mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Set up Nix store if closure info is provided
|
||||
if closure_info and os.path.exists(closure_info):
|
||||
store_paths_file = os.path.join(closure_info, "store-paths")
|
||||
if os.path.exists(store_paths_file):
|
||||
with open(store_paths_file) as f:
|
||||
store_paths = f.read().strip().split("\n")
|
||||
|
||||
# Copy store paths to test store
|
||||
for store_path in store_paths:
|
||||
if store_path.strip():
|
||||
dest_path = f"{tmpdir}/store{store_path}"
|
||||
if not os.path.exists(dest_path):
|
||||
# Create parent directories
|
||||
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
|
||||
# Copy the store path
|
||||
if os.path.isdir(store_path):
|
||||
shutil.copytree(store_path, dest_path, dirs_exist_ok=True)
|
||||
else:
|
||||
shutil.copy2(store_path, dest_path)
|
||||
|
||||
# Load Nix database
|
||||
registration_file = os.path.join(closure_info, "registration")
|
||||
if os.path.exists(registration_file):
|
||||
env = os.environ.copy()
|
||||
env["NIX_REMOTE"] = f"local?store={tmpdir}/store"
|
||||
|
||||
with open(registration_file) as f:
|
||||
subprocess.run(
|
||||
["nix-store", "--load-db"],
|
||||
input=f.read(),
|
||||
text=True,
|
||||
env=env,
|
||||
check=True,
|
||||
)
|
||||
@@ -1,51 +0,0 @@
|
||||
"""Port management utilities for NixOS installation tests."""
|
||||
|
||||
import socket
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
|
||||
class PortUtilsError(Exception):
|
||||
"""Port utils related errors."""
|
||||
|
||||
|
||||
def find_free_port() -> int:
|
||||
"""Find a free port on the host."""
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.bind(("", 0))
|
||||
return s.getsockname()[1]
|
||||
|
||||
|
||||
def check_host_port_open(port: int) -> bool:
|
||||
"""Verify port forwarding is working by checking if the host port is listening."""
|
||||
try:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.settimeout(1)
|
||||
result = s.connect_ex(("localhost", port))
|
||||
return result == 0
|
||||
except Exception as e:
|
||||
print(f"Port check failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def setup_port_forwarding(target: Any, host_port: int) -> None:
|
||||
"""Set up port forwarding and wait for it to be ready."""
|
||||
print(f"Setting up port forwarding from host port {host_port} to guest port 22")
|
||||
target.forward_port(host_port=host_port, guest_port=22)
|
||||
|
||||
# Give the port forwarding time to establish
|
||||
time.sleep(2)
|
||||
|
||||
# Wait up to 30 seconds for the port to become available
|
||||
port_ready = False
|
||||
for i in range(30):
|
||||
if check_host_port_open(host_port):
|
||||
port_ready = True
|
||||
print(f"Host port {host_port} is now listening")
|
||||
break
|
||||
print(f"Waiting for host port {host_port} to be ready... attempt {i + 1}/30")
|
||||
time.sleep(1)
|
||||
|
||||
if not port_ready:
|
||||
msg = f"Host port {host_port} never became available for forwarding"
|
||||
raise PortUtilsError(msg)
|
||||
@@ -1,49 +0,0 @@
|
||||
"""SSH and test setup utilities"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from typing import NamedTuple
|
||||
|
||||
from .nix_setup import setup_nix_in_nix
|
||||
from .port import find_free_port, setup_port_forwarding
|
||||
|
||||
|
||||
class TestEnvironment(NamedTuple):
|
||||
host_port: int
|
||||
ssh_key: str
|
||||
flake_dir: str
|
||||
|
||||
|
||||
def setup_test_environment(
|
||||
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
|
||||
|
||||
Returns:
|
||||
TestEnvironment with host_port, ssh_key, and flake_dir
|
||||
"""
|
||||
# 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)
|
||||
|
||||
setup_port_forwarding(target, host_port)
|
||||
|
||||
ssh_key = os.path.join(temp_dir, "id_ed25519")
|
||||
with open(ssh_key, "w") as f:
|
||||
with open(assets_ssh_privkey) as src:
|
||||
f.write(src.read())
|
||||
os.chmod(ssh_key, 0o600)
|
||||
|
||||
# Copy test flake to temp directory
|
||||
flake_dir = os.path.join(temp_dir, "test-flake")
|
||||
subprocess.run(["cp", "-r", clan_core_for_checks, flake_dir], check=True)
|
||||
subprocess.run(["chmod", "-R", "+w", flake_dir], check=True)
|
||||
|
||||
return TestEnvironment(host_port, ssh_key, flake_dir)
|
||||
@@ -23,7 +23,7 @@ nixosLib.runTest (
|
||||
clan.test.fromFlake = ./.;
|
||||
|
||||
extraPythonPackages = _p: [
|
||||
clan-core.legacyPackages.${hostPkgs.system}.setupNixInNixPythonPackage
|
||||
clan-core.legacyPackages.${hostPkgs.system}.nixosTestLib
|
||||
];
|
||||
|
||||
testScript =
|
||||
|
||||
Reference in New Issue
Block a user