pkgs/clan: Add machine validator with suggestion logic

Add machine validator with suggestion logic to:
- `clan machines update`
- `clan machines delete`
- `clan machines update-hardware-config`
This commit is contained in:
a-kenji
2025-06-26 16:57:51 +02:00
parent 50918c268c
commit 5a81adfc68
6 changed files with 213 additions and 0 deletions

View File

@@ -3,6 +3,7 @@ import logging
from clan_lib.machines.delete import delete_machine
from clan_lib.machines.machines import Machine
from clan_lib.machines.suggestions import validate_machine_names
from clan_cli.completions import add_dynamic_completer, complete_machines
@@ -10,6 +11,7 @@ log = logging.getLogger(__name__)
def delete_command(args: argparse.Namespace) -> None:
validate_machine_names([args.name], args.flake)
delete_machine(Machine(flake=args.flake, name=args.name))

View File

@@ -7,6 +7,7 @@ from clan_lib.machines.hardware import (
generate_machine_hardware_info,
)
from clan_lib.machines.machines import Machine
from clan_lib.machines.suggestions import validate_machine_names
from clan_lib.ssh.remote import Remote
from clan_cli.completions import add_dynamic_completer, complete_machines
@@ -17,6 +18,7 @@ log = logging.getLogger(__name__)
def update_hardware_config_command(args: argparse.Namespace) -> None:
validate_machine_names([args.machine], args.flake)
host_key_check = args.host_key_check
machine = Machine(flake=args.flake, name=args.machine)
opts = HardwareGenerateOptions(

View File

@@ -5,6 +5,7 @@ import sys
from clan_lib.async_run import AsyncContext, AsyncOpts, AsyncRuntime
from clan_lib.errors import ClanError
from clan_lib.machines.machines import Machine
from clan_lib.machines.suggestions import validate_machine_names
from clan_lib.machines.update import deploy_machine
from clan_lib.nix import nix_config
from clan_lib.ssh.remote import Remote
@@ -45,6 +46,9 @@ def update_command(args: argparse.Namespace) -> None:
msg = f"No machines found with tags: {', '.join(args.tags)}"
raise ClanError(msg)
if args.machines:
validate_machine_names(args.machines, args.flake)
for machine_name in selected_machines:
machine = Machine(name=machine_name, flake=args.flake)
machines.append(machine)

View File

@@ -90,6 +90,57 @@ def test_machines_update_with_tags(
assert args.tags == ["vm"]
@pytest.mark.impure
def test_machines_update_nonexistent_machine(
test_flake_with_core: fixtures_flakes.FlakeForTest,
) -> None:
"""Test that update command gives helpful error messages for non-existent machines."""
from clan_lib.errors import ClanError
with pytest.raises(ClanError) as exc_info:
cli.run(
[
"machines",
"update",
"--flake",
str(test_flake_with_core.path),
"nonexistent-machine",
]
)
error_message = str(exc_info.value)
assert "nonexistent-machine" in error_message
assert "not found." in error_message
# Should suggest similar machines (vm1, vm2 exist in test flake)
assert "Did you mean:" in error_message or "Available machines:" in error_message
@pytest.mark.impure
def test_machines_update_typo_in_machine_name(
test_flake_with_core: fixtures_flakes.FlakeForTest,
) -> None:
"""Test that update command suggests similar machine names for typos."""
from clan_lib.errors import ClanError
with pytest.raises(ClanError) as exc_info:
cli.run(
[
"machines",
"update",
"--flake",
str(test_flake_with_core.path),
"v1", # typo of "vm1"
]
)
error_message = str(exc_info.value)
assert "v1" in error_message
assert "not found." in error_message
assert "Did you mean:" in error_message
# Should suggest vm1 as it's the closest match
assert "vm1" in error_message
@pytest.mark.with_core
def test_machine_delete(
monkeypatch: pytest.MonkeyPatch,

View File

@@ -0,0 +1,72 @@
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
def _levenshtein_distance(s1: str, s2: str) -> int:
"""
Calculate the Levenshtein distance between two strings.
"""
if len(s1) < len(s2):
return _levenshtein_distance(s2, s1)
if len(s2) == 0:
return len(s1)
previous_row = list(range(len(s2) + 1))
for i, c1 in enumerate(s1):
current_row = [i + 1]
for j, c2 in enumerate(s2):
insertions = previous_row[j + 1] + 1
deletions = current_row[j] + 1
substitutions = previous_row[j] + (c1 != c2)
current_row.append(min(insertions, deletions, substitutions))
previous_row = current_row
return previous_row[-1]
def _suggest_similar_names(
target: str, candidates: list[str], max_suggestions: int = 3
) -> list[str]:
if not candidates:
return []
distances = [
(candidate, _levenshtein_distance(target.lower(), candidate.lower()))
for candidate in candidates
]
distances.sort(key=lambda x: (x[1], x[0]))
return [candidate for candidate, _ in distances[:max_suggestions]]
def get_available_machines(flake: Flake) -> list[str]:
from clan_lib.machines.list import list_machines
machines = list_machines(flake)
return list(machines.keys())
def validate_machine_names(machine_names: list[str], flake: Flake) -> None:
if not machine_names:
return
available_machines = get_available_machines(flake)
invalid_machines = [
name for name in machine_names if name not in available_machines
]
if invalid_machines:
error_lines = []
for machine_name in invalid_machines:
suggestions = _suggest_similar_names(machine_name, available_machines)
if suggestions:
suggestion_text = f"Did you mean: {', '.join(suggestions)}?"
else:
suggestion_text = (
f"Available machines: {', '.join(sorted(available_machines))}"
)
error_lines.append(f"Machine '{machine_name}' not found. {suggestion_text}")
raise ClanError("\n".join(error_lines))

View File

@@ -0,0 +1,82 @@
from clan_lib.machines.suggestions import (
_levenshtein_distance,
_suggest_similar_names,
)
def test_identical_strings() -> None:
assert _levenshtein_distance("hello", "hello") == 0
def test_empty_string() -> None:
assert _levenshtein_distance("", "") == 0
assert _levenshtein_distance("hello", "") == 5
assert _levenshtein_distance("", "world") == 5
def test_single_character_difference() -> None:
assert _levenshtein_distance("hello", "hallo") == 1 # substitution
assert _levenshtein_distance("hello", "helloo") == 1 # insertion
assert _levenshtein_distance("hello", "hell") == 1 # deletion
def test_multiple_differences() -> None:
assert _levenshtein_distance("kitten", "sitting") == 3
assert _levenshtein_distance("saturday", "sunday") == 3
def test_case_sensitivity() -> None:
assert _levenshtein_distance("Hello", "hello") == 1
assert _levenshtein_distance("HELLO", "hello") == 5
def test_exact_match() -> None:
candidates = ["machine1", "machine2", "machine3"]
suggestions = _suggest_similar_names("machine1", candidates)
assert suggestions[0] == "machine1"
def test_close_match() -> None:
candidates = ["machine1", "machine2", "machine3"]
suggestions = _suggest_similar_names("machne1", candidates) # missing 'i'
assert "machine1" in suggestions[:2]
def test_case_insensitive_matching() -> None:
candidates = ["Machine1", "MACHINE2", "machine3"]
suggestions = _suggest_similar_names("machine1", candidates)
assert "Machine1" in suggestions[:2]
def test_max_suggestions_limit() -> None:
candidates = ["aa", "ab", "ac", "ad", "ae"]
suggestions = _suggest_similar_names("a", candidates, max_suggestions=3)
assert len(suggestions) <= 3
def test_empty_candidates() -> None:
suggestions = _suggest_similar_names("test", [])
assert suggestions == []
def test_realistic_machine_names() -> None:
candidates = ["web-server", "database", "worker-01", "worker-02", "api-gateway"]
# Test typo in web-server
suggestions = _suggest_similar_names("web-sever", candidates)
assert suggestions[0] == "web-server"
# Test partial match
suggestions = _suggest_similar_names("work", candidates)
worker_suggestions = [s for s in suggestions if "worker" in s]
assert len(worker_suggestions) >= 2
def test_sorting_by_distance_then_alphabetically() -> None:
candidates = ["zebra", "apple", "apricot"]
suggestions = _suggest_similar_names("app", candidates)
# Both "apple" and "apricot" have same distance (2),
# but "apple" should come first alphabetically
apple_index = suggestions.index("apple")
apricot_index = suggestions.index("apricot")
assert apple_index < apricot_index