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

make vm test pure

Browse Source
This commit is contained in:
Jörg Thalheim
2025-08-15 15:54:25 +02:00
parent 6fe2b06f09
commit 4074a184b2
9 changed files with 282 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pkgs/clan-cli/clan_cli/tests/age_keys.py
View File

@@ -1,6 +1,5 @@
import dataclasses
import json
import os
from collections.abc import Iterable
from pathlib import Path
@@ -26,7 +25,7 @@ class SopsSetup:
def __init__(self, keys: list[KeyPair]) -> None:
self.keys = keys
self.user = os.environ.get("USER", "admin")
self.user = "admin"
def init(self, flake_path: Path) -> None:
cli.run(

90
pkgs/clan-cli/clan_cli/tests/fixtures_flakes.py
View File

@@ -422,3 +422,93 @@ def test_flake_with_core(
monkeypatch=monkeypatch,
inventory_expr=inventory_expr,
)
@pytest.fixture
def writable_clan_core(
clan_core: Path,
tmp_path: Path,
) -> Path:
"""
Creates a writable copy of clan_core in a temporary directory.
If clan_core is a git repo, copies tracked files and uncommitted changes.
Removes vars/ and sops/ directories if they exist.
"""
temp_flake = tmp_path / "clan-core"
# Check if it's a git repository
if (clan_core / ".git").exists():
# Create the target directory
temp_flake.mkdir(parents=True)
# Copy all tracked and untracked files (excluding ignored)
# Using git ls-files with -z for null-terminated output to handle filenames with spaces
sp.run(
f"(git ls-files -z; git ls-files -z --others --exclude-standard) | "
f"xargs -0 cp --parents -t {temp_flake}/",
shell=True,
cwd=clan_core,
check=True,
)
# Copy .git directory to maintain git functionality
if (clan_core / ".git").is_dir():
shutil.copytree(
clan_core / ".git", temp_flake / ".git", ignore_dangling_symlinks=True
)
else:
# It's a git file (for submodules/worktrees)
shutil.copy2(clan_core / ".git", temp_flake / ".git")
else:
# Regular copy if not a git repo
shutil.copytree(clan_core, temp_flake, ignore_dangling_symlinks=True)
# Make writable
sp.run(["chmod", "-R", "+w", str(temp_flake)], check=True)
# Remove vars and sops directories
shutil.rmtree(temp_flake / "vars", ignore_errors=True)
shutil.rmtree(temp_flake / "sops", ignore_errors=True)
return temp_flake
@pytest.fixture
def vm_test_flake(
clan_core: Path,
tmp_path: Path,
) -> Path:
"""
Creates a test flake that imports the VM test nixOS modules from clan-core.
"""
test_flake_dir = tmp_path / "test-flake"
test_flake_dir.mkdir(parents=True)
metadata = sp.run(
nix_command(["flake", "metadata", "--json"]),
cwd=CLAN_CORE,
capture_output=True,
text=True,
check=True,
).stdout.strip()
metadata_json = json.loads(metadata)
clan_core_url = f"path:{metadata_json['path']}"
# Read the template and substitute the clan-core path
template_path = Path(__file__).parent / "vm_test_flake.nix"
template_content = template_path.read_text()
# Substitute the clan-core URL
flake_content = template_content.replace("__CLAN_CORE__", clan_core_url)
# Write the flake.nix
(test_flake_dir / "flake.nix").write_text(flake_content)
# Lock the flake with --allow-dirty to handle uncommitted changes
sp.run(
nix_command(["flake", "lock", "--allow-dirty-locks"]),
cwd=test_flake_dir,
check=True,
)
return test_flake_dir

119
pkgs/clan-cli/clan_cli/tests/flake-module.nix Normal file
View File

@@ -0,0 +1,119 @@
{ self, ... }:
{
# Define machines that use the nixOS modules
clan.machines = {
test-vm-persistence = {
imports = [ self.nixosModules.test-vm-persistence ];
};
test-vm-deployment = {
imports = [ self.nixosModules.test-vm-deployment ];
};
};
flake.nixosModules = {
# NixOS module for test_vm_persistence
test-vm-persistence =
{ config, ... }:
{
nixpkgs.hostPlatform = "x86_64-linux";
system.stateVersion = config.system.nixos.release;
# Disable services that might cause issues in tests
systemd.services.logrotate-checkconf.enable = false;
services.getty.autologinUser = "root";
# Basic networking setup
networking.useDHCP = false;
networking.firewall.enable = false;
# VM-specific settings
clan.virtualisation.graphics = false;
clan.core.networking.targetHost = "client";
# State configuration for persistence test
clan.core.state.my_state.folders = [
"/var/my-state"
"/var/user-state"
];
# Initialize users for tests
users.users = {
root = {
initialPassword = "root";
};
test = {
initialPassword = "test";
isSystemUser = true;
group = "users";
};
};
};
# NixOS module for test_vm_deployment
test-vm-deployment =
{ config, lib, ... }:
{
nixpkgs.hostPlatform = "x86_64-linux";
system.stateVersion = config.system.nixos.release;
# Disable services that might cause issues in tests
systemd.services.logrotate-checkconf.enable = false;
services.getty.autologinUser = "root";
# Basic networking setup
networking.useDHCP = false;
networking.firewall.enable = false;
# VM-specific settings
clan.virtualisation.graphics = false;
# SSH for deployment tests
services.openssh.enable = true;
# Initialize users for tests
users.users = {
root = {
initialPassword = "root";
};
};
# hack to make sure
sops.validateSopsFiles = false;
sops.secrets."vars/m1_generator/my_secret" = lib.mkDefault {
sopsFile = builtins.toFile "fake" "";
};
# Vars generators configuration
clan.core.vars.generators = {
m1_generator = {
files.my_secret = {
secret = true;
path = "/run/secrets/vars/m1_generator/my_secret";
};
script = ''
echo hello > "$out"/my_secret
'';
};
my_shared_generator = {
share = true;
files = {
shared_secret = {
secret = true;
path = "/run/secrets/vars/my_shared_generator/shared_secret";
};
no_deploy_secret = {
secret = true;
deploy = false;
path = "/run/secrets/vars/my_shared_generator/no_deploy_secret";
};
};
script = ''
echo hello > "$out"/shared_secret
echo hello > "$out"/no_deploy_secret
'';
};
};
};
};
}

56
pkgs/clan-cli/clan_cli/tests/test_vars_deployment.py
View File

@@ -2,63 +2,33 @@ import json
import subprocess
import sys
from contextlib import ExitStack
from pathlib import Path
import pytest
from clan_cli.tests.age_keys import SopsSetup
from clan_cli.tests.fixtures_flakes import ClanFlake
from clan_cli.tests.helpers import cli
from clan_cli.vms.run import inspect_vm, spawn_vm
from clan_lib import cmd
from clan_lib.flake import Flake
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_config, nix_eval, run
from clan_lib.nix import nix_eval, run
@pytest.mark.impure
@pytest.mark.skipif(sys.platform == "darwin", reason="preload doesn't work on darwin")
def test_vm_deployment(
flake: ClanFlake,
vm_test_flake: Path,
sops_setup: SopsSetup,
) -> None:
# machine 1
config = nix_config()
machine1_config = flake.machines["m1_machine"]
machine1_config["nixpkgs"]["hostPlatform"] = config["system"]
machine1_config["clan"]["virtualisation"]["graphics"] = False
machine1_config["services"]["getty"]["autologinUser"] = "root"
machine1_config["services"]["openssh"]["enable"] = True
machine1_config["networking"]["firewall"]["enable"] = False
machine1_config["users"]["users"]["root"]["openssh"]["authorizedKeys"]["keys"] = [
# put your key here when debugging and pass ssh_port in run_vm_in_thread call below
]
m1_generator = machine1_config["clan"]["core"]["vars"]["generators"]["m1_generator"]
m1_generator["files"]["my_secret"]["secret"] = True
m1_generator["script"] = """
echo hello > "$out"/my_secret
"""
m1_shared_generator = machine1_config["clan"]["core"]["vars"]["generators"][
"my_shared_generator"
]
m1_shared_generator["share"] = True
m1_shared_generator["files"]["shared_secret"]["secret"] = True
m1_shared_generator["files"]["no_deploy_secret"]["secret"] = True
m1_shared_generator["files"]["no_deploy_secret"]["deploy"] = False
m1_shared_generator["script"] = """
echo hello > "$out"/shared_secret
echo hello > "$out"/no_deploy_secret
"""
flake.refresh()
sops_setup.init(flake.path)
cli.run(["vars", "generate", "--flake", str(flake.path)])
# Set up sops for the test flake machines
sops_setup.init(vm_test_flake)
cli.run(["vars", "generate", "--flake", str(vm_test_flake), "test-vm-deployment"])
# check sops secrets not empty
sops_secrets = json.loads(
run(
nix_eval(
[
f"{flake.path}#nixosConfigurations.m1_machine.config.sops.secrets",
f"{vm_test_flake}#nixosConfigurations.test-vm-deployment.config.sops.secrets",
]
)
).stdout.strip()
@@ -67,7 +37,7 @@ def test_vm_deployment(
my_secret_path = run(
nix_eval(
[
f"{flake.path}#nixosConfigurations.m1_machine.config.clan.core.vars.generators.m1_generator.files.my_secret.path",
f"{vm_test_flake}#nixosConfigurations.test-vm-deployment.config.clan.core.vars.generators.m1_generator.files.my_secret.path",
]
)
).stdout.strip()
@@ -75,15 +45,15 @@ def test_vm_deployment(
shared_secret_path = run(
nix_eval(
[
f"{flake.path}#nixosConfigurations.m1_machine.config.clan.core.vars.generators.my_shared_generator.files.shared_secret.path",
f"{vm_test_flake}#nixosConfigurations.test-vm-deployment.config.clan.core.vars.generators.my_shared_generator.files.shared_secret.path",
]
)
).stdout.strip()
assert "no-such-path" not in shared_secret_path
# run nix flake lock
cmd.run(["nix", "flake", "lock"], cmd.RunOpts(cwd=flake.path))
vm1_config = inspect_vm(machine=Machine("m1_machine", Flake(str(flake.path))))
vm1_config = inspect_vm(
machine=Machine("test-vm-deployment", Flake(str(vm_test_flake)))
)
with ExitStack() as stack:
vm1 = stack.enter_context(spawn_vm(vm1_config, stdin=subprocess.DEVNULL))
qga_m1 = stack.enter_context(vm1.qga_connect())
@@ -92,7 +62,7 @@ def test_vm_deployment(
# check my_secret is deployed
result = qga_m1.run(["cat", "/run/secrets/vars/m1_generator/my_secret"])
assert result.stdout == "hello\n"
# check shared_secret is deployed on m1
# check shared_secret is deployed
result = qga_m1.run(
["cat", "/run/secrets/vars/my_shared_generator/shared_secret"]
)

33
pkgs/clan-cli/clan_cli/tests/test_vms_cli.py
View File

@@ -2,7 +2,7 @@ from pathlib import Path
from typing import TYPE_CHECKING
import pytest
from clan_cli.tests.fixtures_flakes import ClanFlake, FlakeForTest
from clan_cli.tests.fixtures_flakes import FlakeForTest
from clan_cli.tests.helpers import cli
from clan_cli.tests.stdout import CaptureOutput
from clan_cli.vms.run import inspect_vm, spawn_vm
@@ -24,8 +24,7 @@ def test_inspect(
assert "Cores" in output.out
# @pytest.mark.skipif(no_kvm, reason="Requires KVM")
@pytest.mark.skipif(True, reason="We need to fix vars support for vms for this test")
@pytest.mark.skipif(no_kvm, reason="Requires KVM")
@pytest.mark.impure
def test_run(
monkeypatch: pytest.MonkeyPatch,
@@ -60,30 +59,12 @@ def test_run(
@pytest.mark.skipif(no_kvm, reason="Requires KVM")
@pytest.mark.impure
def test_vm_persistence(
flake: ClanFlake,
vm_test_flake: Path,
) -> None:
# set up a clan flake with some systemd services to test persistence
config = flake.machines["my_machine"]
config["nixpkgs"]["hostPlatform"] = "x86_64-linux"
# logrotate-checkconf doesn't work in VM because /nix/store is owned by nobody
config["systemd"]["services"]["logrotate-checkconf"]["enable"] = False
config["services"]["getty"]["autologinUser"] = "root"
config["clan"]["virtualisation"] = {"graphics": False}
config["clan"]["core"]["networking"] = {"targetHost": "client"}
config["clan"]["core"]["state"]["my_state"]["folders"] = [
# to be owned by root
"/var/my-state",
# to be owned by user 'test'
"/var/user-state",
]
config["users"]["users"] = {
"test": {"initialPassword": "test", "isSystemUser": True, "group": "users"},
"root": {"initialPassword": "root"},
}
flake.refresh()
vm_config = inspect_vm(machine=Machine("my_machine", Flake(str(flake.path))))
# Use the pre-built test VM from the test flake
vm_config = inspect_vm(
machine=Machine("test-vm-persistence", Flake(str(vm_test_flake)))
)
with spawn_vm(vm_config) as vm, vm.qga_connect() as qga:
# create state via qmp command instead of systemd service

26
pkgs/clan-cli/clan_cli/tests/vm_test_flake.nix Normal file
View File

@@ -0,0 +1,26 @@
{
inputs.clan-core.url = "__CLAN_CORE__";
outputs =
{ self, clan-core }:
let
clan = clan-core.lib.clan {
inherit self;
meta.name = "test-flake";
machines = {
test-vm-persistence = {
imports = [ clan-core.nixosModules.test-vm-persistence ];
};
test-vm-deployment = {
imports = [ clan-core.nixosModules.test-vm-deployment ];
};
};
};
in
{
inherit (clan.config) nixosConfigurations;
inherit (clan.config) nixosModules;
inherit (clan.config) clanInternals;
clan = clan.config;
};
}

2
pkgs/clan-cli/clan_cli/vms/run.py
View File

@@ -16,7 +16,7 @@ from clan_lib.cmd import CmdOut, Log, RunOpts, handle_io, run
from clan_lib.dirs import module_root, user_cache_dir, vm_state_dir
from clan_lib.errors import ClanCmdError, ClanError
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_shell
from clan_lib.nix import nix_shell, nix_test_store
from clan_lib.vars.generate import run_generators
from clan_cli.completions import add_dynamic_completer, complete_machines

21
pkgs/clan-cli/default.nix
View File

@@ -19,6 +19,7 @@
templateDerivation,
zerotierone,
minifakeroot,
nixosConfigurations,
}:
let
pyDeps = ps: [
@@ -225,6 +226,26 @@ pythonRuntime.pkgs.buildPythonApplication {
# needed by flash list tests
nixpkgs.legacyPackages.x86_64-linux.kbd
nixpkgs.legacyPackages.x86_64-linux.glibcLocales
# Pre-built VMs for impure tests
pkgs.stdenv.drvPath
pkgs.bash.drvPath
pkgs.buildPackages.xorg.lndir
(pkgs.perl.withPackages (
p: with p; [
ConfigIniFiles
FileSlurp
]
))
(pkgs.closureInfo { rootPaths = [ ]; }).drvPath
pkgs.desktop-file-utils
pkgs.dbus
pkgs.unzip
pkgs.libxslt
pkgs.getconf
nixosConfigurations.test-vm-persistence.config.system.clan.vm.create
nixosConfigurations.test-vm-deployment.config.system.clan.vm.create
];
};
}

4
pkgs/clan-cli/flake-module.nix
View File

@@ -5,6 +5,8 @@
...
}:
{
imports = [ ./clan_cli/tests/flake-module.nix ];
perSystem =
{
self',
@@ -55,6 +57,7 @@
"age"
"git"
];
inherit (self) nixosConfigurations;
};
clan-cli-full = pkgs.callPackage ./default.nix {
inherit (inputs) nixpkgs nix-select;
@@ -64,6 +67,7 @@
templateDerivation = templateDerivation;
pythonRuntime = pkgs.python3;
includedRuntimeDeps = lib.importJSON ./clan_lib/nix/allowed-packages.json;
inherit (self) nixosConfigurations;
};
clan-cli-docs = pkgs.stdenv.mkDerivation {
name = "clan-cli-docs";