Merge pull request 'pkgs/cli: Fix dynamic shell completions' (#5599) from ke-cli-completion-fix into main

Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/5599
This commit is contained in:
Mic92
2025-10-21 14:36:10 +00:00
2 changed files with 129 additions and 13 deletions

View File

@@ -23,6 +23,9 @@ from .secrets.users import list_users
This module provides dynamic completions.
The completions should feel fast.
We target a maximum of 1second on our average machine.
Note: All completion functions have 'prefix' as their first parameter (required
by argcomplete's API) but don't use it internally.
"""
@@ -44,7 +47,7 @@ def clan_dir(flake: str | None) -> str | None:
def complete_machines(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -77,7 +80,7 @@ def complete_machines(
def complete_services_for_machine(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -121,7 +124,7 @@ def complete_services_for_machine(
def complete_backup_providers_for_machine(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -164,7 +167,7 @@ def complete_backup_providers_for_machine(
def complete_state_services_for_machine(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -207,7 +210,7 @@ def complete_state_services_for_machine(
def complete_secrets(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -225,7 +228,7 @@ def complete_secrets(
def complete_users(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -243,7 +246,7 @@ def complete_users(
def complete_groups(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -262,7 +265,7 @@ def complete_groups(
def complete_templates_disko(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -283,7 +286,7 @@ def complete_templates_disko(
def complete_templates_clan(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -304,7 +307,7 @@ def complete_templates_clan(
def complete_templates_machine(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -325,7 +328,7 @@ def complete_templates_machine(
def complete_vars_for_machine(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -366,7 +369,7 @@ def complete_vars_for_machine(
def complete_target_host(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:
@@ -407,7 +410,7 @@ def complete_target_host(
def complete_tags(
_prefix: str,
prefix: str, # noqa: ARG001
parsed_args: argparse.Namespace,
**_kwargs: Any,
) -> Iterable[str]:

View File

@@ -0,0 +1,113 @@
"""Tests for dynamic completion functionality.
These tests verify that dynamic completions work correctly for various commands.
The tests directly call completion functions rather than simulating shell completion.
"""
import argparse
import inspect
import pytest
from clan_lib.flake import Flake
from .completions import complete_machines
from .tests import fixtures_flakes
from .tests.helpers import cli
@pytest.mark.with_core
def test_complete_machines(
test_flake_with_core: fixtures_flakes.FlakeForTest,
) -> None:
"""Test that complete_machines returns the correct machine names."""
parsed_args = argparse.Namespace(flake=Flake(str(test_flake_with_core.path)))
completions = complete_machines(prefix="", parsed_args=parsed_args)
completion_list = (
list(completions.keys()) if isinstance(completions, dict) else list(completions)
)
assert "vm1" in completion_list, (
f"Expected 'vm1' in completions, got: {completion_list}"
)
assert "vm2" in completion_list, (
f"Expected 'vm2' in completions, got: {completion_list}"
)
@pytest.mark.with_core
def test_complete_machines_with_prefix(
test_flake_with_core: fixtures_flakes.FlakeForTest,
) -> None:
parsed_args = argparse.Namespace(flake=Flake(str(test_flake_with_core.path)))
completions = complete_machines(prefix="vm", parsed_args=parsed_args)
completion_list = (
list(completions.keys()) if isinstance(completions, dict) else list(completions)
)
assert len(completion_list) > 0, "Should return machine names"
@pytest.mark.with_core
def test_complete_machines_without_flake(
test_flake_with_core: fixtures_flakes.FlakeForTest,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test that complete_machines works when flake is not explicitly set.
When no --flake argument is provided, completions should use the current
directory or CLAN_DIR environment variable.
"""
monkeypatch.chdir(test_flake_with_core.path)
parsed_args = argparse.Namespace()
completions = complete_machines(prefix="", parsed_args=parsed_args)
completion_list = (
list(completions.keys()) if isinstance(completions, dict) else list(completions)
)
assert "vm1" in completion_list
assert "vm2" in completion_list
@pytest.mark.with_core
def test_complete_machines_with_created_machine(
test_flake_with_core: fixtures_flakes.FlakeForTest,
) -> None:
"""Test that completions include dynamically created machines."""
# Create a new machine
cli.run(
[
"machines",
"create",
"--flake",
str(test_flake_with_core.path),
"test-machine",
],
)
parsed_args = argparse.Namespace(flake=Flake(str(test_flake_with_core.path)))
completions = complete_machines(prefix="", parsed_args=parsed_args)
completion_list = (
list(completions.keys()) if isinstance(completions, dict) else list(completions)
)
# Should include the newly created machine
assert "test-machine" in completion_list, (
f"Expected 'test-machine' in {completion_list}"
)
assert "vm1" in completion_list
assert "vm2" in completion_list
def test_complete_machines_signature() -> None:
"""Test that complete_machines has the correct signature for argcomplete."""
sig = inspect.signature(complete_machines)
params = list(sig.parameters.keys())
assert params[0] == "prefix", (
f"First parameter should be 'prefix' for argcomplete compatibility, got: {params[0]}"
)