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:
@@ -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))
|
||||
|
||||
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
72
pkgs/clan-cli/clan_lib/machines/suggestions.py
Normal file
72
pkgs/clan-cli/clan_lib/machines/suggestions.py
Normal 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))
|
||||
82
pkgs/clan-cli/clan_lib/machines/test_suggestions.py
Normal file
82
pkgs/clan-cli/clan_lib/machines/test_suggestions.py
Normal 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
|
||||
Reference in New Issue
Block a user