]> git.ipfire.org Git - thirdparty/knot-resolver.git/commitdiff
python: client: command: move the getting of completion words to function
authorAleš Mrázek <ales.mrazek@nic.cz>
Mon, 16 Dec 2024 17:04:47 +0000 (18:04 +0100)
committerAleš Mrázek <ales.mrazek@nic.cz>
Fri, 20 Dec 2024 21:24:22 +0000 (22:24 +0100)
- #dirnames# and #filenames# words to indicate that we want complete also files and dirs

python/knot_resolver/client/command.py
python/knot_resolver/client/commands/cache.py
python/knot_resolver/client/commands/completion.py
python/knot_resolver/client/commands/config.py
python/knot_resolver/client/commands/convert.py
python/knot_resolver/client/commands/debug.py
python/knot_resolver/client/commands/help.py
python/knot_resolver/client/commands/metrics.py
python/knot_resolver/client/commands/schema.py
python/knot_resolver/client/commands/validate.py
utils/shell-completion/client.bash

index 5a0d11a7575795f0ac82b0ebd613f2f0bf60b445..e4eddf0877706aebb0cc47877288f56a8c192e78 100644 (file)
@@ -1,7 +1,7 @@
 import argparse
 from abc import ABC, abstractmethod  # pylint: disable=[no-name-in-module]
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple, Type, TypeVar
+from typing import Dict, List, Optional, Tuple, Type, TypeVar
 from urllib.parse import quote
 
 from knot_resolver.constants import API_SOCK_FILE, CONFIG_FILE
@@ -14,33 +14,96 @@ T = TypeVar("T", bound=Type["Command"])
 
 CompWords = Dict[str, Optional[str]]
 
+COMP_DIRNAMES = "#dirnames#"
+COMP_FILENAMES = "#filenames#"
+
 _registered_commands: List[Type["Command"]] = []
 
 
-def get_subparsers_words(subparser_actions: List[argparse.Action]) -> CompWords:
+def get_parser_action(name: str, parser_actions: List[argparse.Action]) -> Optional[argparse.Action]:
+    for action in parser_actions:
+        if (action.choices and name in action.choices) or (action.option_strings and name in action.option_strings):
+            return action
+    return None
+
+
+def get_subparser_command(subparser: argparse.ArgumentParser) -> Optional["Command"]:
+    if "command" in subparser._defaults:  # noqa: SLF001
+        return subparser._defaults["command"]  # noqa: SLF001
+    return None
+
+
+def comp_get_actions_words(parser_actions: List[argparse.Action]) -> CompWords:
     words: CompWords = {}
-    for action in subparser_actions:
+    for action in parser_actions:
         if isinstance(action, argparse._SubParsersAction) and action.choices:  # noqa: SLF001
             for choice, parser in action.choices.items():
-                words[choice] = parser.description
-        else:
+                words[choice] = parser.description if isinstance(parser, argparse.ArgumentParser) else None
+        elif action.option_strings:
             for opt in action.option_strings:
                 words[opt] = action.help
+        elif not action.option_strings and action.choices:
+            for choice in action.choices:
+                words[choice] = action.help
+        elif not action.option_strings and not action.choices:
+            words[COMP_DIRNAMES] = None
+            words[COMP_FILENAMES] = None
     return words
 
 
-def get_action_by_name(name: str, parser_actions: List[argparse.Action]) -> Optional[argparse.Action]:
-    for action in parser_actions:
-        if (action.choices and name in action.choices) or (action.option_strings and name in action.option_strings):
-            return action
-    return None
-
+def comp_get_words(args: List[str], parser_actions: List[argparse.Action]) -> CompWords:  # noqa: PLR0912
+    words: CompWords = comp_get_actions_words(parser_actions)
+    nargs = len(args)
+
+    skip_arg = False
+    for i, arg in enumerate(args):
+        action: Optional[argparse.Action] = get_parser_action(arg, parser_actions)  # noqa: SLF001
+
+        if skip_arg:
+            skip_arg = False
+            continue
+
+        if not action:
+            words = comp_get_actions_words(parser_actions)
+            continue
+
+        if i + 1 >= nargs:
+            return words
+
+        # if not action or action is HelpAction or VersionAction
+        if isinstance(action, (argparse._HelpAction, argparse._VersionAction)):  # noqa: SLF001
+            words = {}
+            break
+
+        # if action is StoreTrueAction or StoreFalseAction
+        if isinstance(action, argparse._StoreConstAction):  # noqa: SLF001
+            continue
+
+        # if action is StoreAction
+        if isinstance(action, argparse._StoreAction):  # noqa: SLF001
+            choices = {}
+            if action.choices:
+                for choice in action.choices:
+                    choices[choice] = action.help
+            else:
+                choices[COMP_DIRNAMES] = None
+                choices[COMP_FILENAMES] = None
+            words = choices
+            skip_arg = True
+            continue
+
+        # if action is SubParserAction
+        if isinstance(action, argparse._SubParsersAction):  # noqa: SLF001
+            subparser = action.choices[arg]
+            cmd = get_subparser_command(subparser)
+            return cmd.completion(args[i + 1 :], subparser) if cmd else {}
+
+    # delete already used args
+    for arg in args:
+        if arg in words.keys():
+            del words[arg]
 
-def get_subparser_command(subparser: argparse.ArgumentParser) -> "Command":
-    defaults: Dict[str, Any] = subparser._defaults  # noqa: SLF001
-    if "command" in defaults:
-        return defaults["command"]
-    raise ValueError(f"missing 'command' default for '{subparser.prog}' parser")
+    return words
 
 
 def register_command(cls: T) -> T:
index d11165808a5e84c96e78186a2c935e3fdb39d2f4..66ee77c96c4eee576e22eca3f18d2a209eca31fe 100644 (file)
@@ -3,7 +3,7 @@ import sys
 from enum import Enum
 from typing import Any, Dict, List, Optional, Tuple, Type
 
-from knot_resolver.client.command import Command, CommandArgs, CompWords, get_subparsers_words, register_command
+from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
 from knot_resolver.datamodel.cache_schema import CacheClearRPCSchema
 from knot_resolver.utils.modeling.exceptions import AggregateDataValidationError, DataValidationError
 from knot_resolver.utils.modeling.parsing import DataFormat, parse_json
@@ -99,7 +99,7 @@ class CacheCommand(Command):
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        return get_subparsers_words(parser._actions)  # noqa: SLF001
+        return comp_get_words(args, parser._actions)  # noqa: SLF001
 
     def run(self, args: CommandArgs) -> None:
         if not self.operation:
index 1da8c5164b6ca894b18e9449e3e224fdd4c89a4a..f3101c1977cc0b49440674fd61f9c755be13517e 100644 (file)
@@ -1,14 +1,12 @@
 import argparse
 from enum import Enum
-from typing import List, Optional, Tuple, Type
+from typing import List, Tuple, Type
 
 from knot_resolver.client.command import (
     Command,
     CommandArgs,
     CompWords,
-    get_action_by_name,
-    get_subparser_command,
-    get_subparsers_words,
+    comp_get_words,
     register_command,
 )
 
@@ -23,12 +21,8 @@ class CompletionCommand(Command):
     def __init__(self, namespace: argparse.Namespace) -> None:
         super().__init__(namespace)
         self.shell: Shells = namespace.shell
-        self.space = namespace.space
         self.args: List[str] = namespace.args
 
-        if self.space:
-            self.args.append("")
-
     @staticmethod
     def register_args_subparser(
         subparser: "argparse._SubParsersAction[argparse.ArgumentParser]",
@@ -37,13 +31,6 @@ class CompletionCommand(Command):
             "completion",
             help="commands auto-completion",
         )
-        completion.add_argument(
-            "--space",
-            help="space after last word, returns all possible folowing options",
-            dest="space",
-            action="store_true",
-            default=False,
-        )
 
         shells_dest = "shell"
         shells = completion.add_mutually_exclusive_group()
@@ -56,58 +43,14 @@ class CompletionCommand(Command):
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        return get_subparsers_words(parser._actions)  # noqa: SLF001
+        return comp_get_words(args, parser._actions)  # noqa: SLF001
 
     def run(self, args: CommandArgs) -> None:  # noqa: PLR0912
-        subparsers = args.parser._subparsers  # noqa: SLF001
         words: CompWords = {}
 
+        subparsers = args.parser._subparsers  # noqa: SLF001
         if subparsers:
-            words = get_subparsers_words(subparsers._actions)  # noqa: SLF001
-
-            args_iter = iter(self.args)
-            for arg in args_iter:
-                action: Optional[argparse.Action] = get_action_by_name(arg, subparsers._actions)  # noqa: SLF001
-
-                # if action is SubParserAction; complete using the command
-                if isinstance(action, argparse._SubParsersAction) and arg in action.choices:  # noqa: SLF001
-                    # remove from words
-                    for choice in action.choices:
-                        del words[choice]
-
-                    subparser = action.choices[arg]
-                    cmd = get_subparser_command(subparser)
-
-                    nargs = len(self.args)
-                    index = self.args.index(arg) + 1
-                    # check that index is not out of args length
-                    if index > nargs:
-                        break
-
-                    # complete using the command
-                    words = cmd.completion(self.args[index:], subparser)
-                    break
-
-                # if action is StoreAction; skip number of arguments
-                if isinstance(action, argparse._StoreAction) and arg in action.option_strings:  # noqa: SLF001
-                    # remove from words
-                    for option_string in action.option_strings:
-                        del words[option_string]
-
-                    if action.nargs and isinstance(action.nargs, int):
-                        for _ in range(action.nargs):
-                            next(args_iter)
-                    continue
-
-                # remove other options from words
-                if action and action.option_strings:
-                    for option_string in action.option_strings:
-                        del words[option_string]
-
-                # if 'arg' is not found in actions
-                # there is nothing to complete
-                if not action:
-                    break
+            words = comp_get_words(self.args, subparsers._actions)  # noqa: SLF001
 
         # print completion words
         # based on required bash/fish shell format
index dbf6ccaa16d0f51ffb67fab36bef4b3d92fdf36d..3ebc96ec57b87b7142994b0b0184c555390a7ec8 100644 (file)
@@ -3,7 +3,7 @@ import sys
 from enum import Enum
 from typing import Any, Dict, List, Literal, Optional, Tuple, Type
 
-from knot_resolver.client.command import Command, CommandArgs, CompWords, get_subparsers_words, register_command
+from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
 from knot_resolver.datamodel import KresConfig
 from knot_resolver.utils.modeling.parsing import DataFormat, parse_json, try_to_parse
 from knot_resolver.utils.requests import request
@@ -170,7 +170,7 @@ class ConfigCommand(Command):
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        words = get_subparsers_words(parser._actions)  # noqa: SLF001
+        words = comp_get_words(args, parser._actions)  # noqa: SLF001
         if args is None:
             return words
 
index b72af758b8d65a5a653db63c9e74f4148255a366..28a5be957e3d513f68a36b23f9afb187f8f0a3ee 100644 (file)
@@ -3,7 +3,7 @@ import sys
 from pathlib import Path
 from typing import List, Optional, Tuple, Type
 
-from knot_resolver.client.command import Command, CommandArgs, CompWords, get_subparsers_words, register_command
+from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
 from knot_resolver.datamodel import KresConfig
 from knot_resolver.datamodel.globals import Context, reset_global_validation_context, set_global_validation_context
 from knot_resolver.utils.modeling import try_to_parse
@@ -39,7 +39,6 @@ class ConvertCommand(Command):
             type=str,
             help="File with configuration in YAML or JSON format.",
         )
-
         convert.add_argument(
             "output_file",
             type=str,
@@ -47,12 +46,11 @@ class ConvertCommand(Command):
             help="Optional, output file for converted configuration in Lua script. If not specified, converted configuration is printed.",
             default=None,
         )
-
         return convert, ConvertCommand
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        return get_subparsers_words(parser._actions)  # noqa: SLF001
+        return comp_get_words(args, parser._actions)  # noqa: SLF001
 
     def run(self, args: CommandArgs) -> None:
         with open(self.input_file, "r") as f:
index 5d9a81df0b1f84dceabc050f7908623750ac2cea..00853c0344920390a275ad680f9436509ce242d3 100644 (file)
@@ -5,7 +5,7 @@ import sys
 from pathlib import Path
 from typing import List, Optional, Tuple, Type
 
-from knot_resolver.client.command import Command, CommandArgs, CompWords, register_command
+from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
 from knot_resolver.utils import which
 from knot_resolver.utils.requests import request
 
@@ -30,13 +30,6 @@ class DebugCommand(Command):
             "debug",
             help="Run GDB on the manager's subprocesses",
         )
-        debug.add_argument(
-            "proc_type",
-            help="Optional, the type of process to debug. May be 'kresd' (default), 'gc', or 'all'.",
-            type=str,
-            nargs="?",
-            default="kresd",
-        )
         debug.add_argument(
             "--sudo",
             dest="sudo",
@@ -56,11 +49,19 @@ class DebugCommand(Command):
             action="store_true",
             default=False,
         )
+        debug.add_argument(
+            "proc_type",
+            help="Optional, the type of process to debug. May be 'kresd', 'gc', or 'all'.",
+            choices=["kresd", "gc", "all"],
+            type=str,
+            nargs="?",
+            default="kresd",
+        )
         return debug, DebugCommand
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        return {}
+        return comp_get_words(args, parser._actions)  # noqa: SLF001
 
     def run(self, args: CommandArgs) -> None:  # noqa: PLR0912, PLR0915
         if self.gdb is None:
index 87306c2ab70a15362db8b96c6d1280e829b237d2..1db83fb6106f6b4ae540f4e246b99a9fd9bf735a 100644 (file)
@@ -1,7 +1,7 @@
 import argparse
 from typing import List, Tuple, Type
 
-from knot_resolver.client.command import Command, CommandArgs, CompWords, register_command
+from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
 
 
 @register_command
@@ -14,7 +14,7 @@ class HelpCommand(Command):
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        return {}
+        return comp_get_words(args, parser._actions)  # noqa: SLF001
 
     @staticmethod
     def register_args_subparser(
index 85cff2583db3aaecd6273c3e25882ae3fb8e8b52..eaf83090dfe9e5f2f62f89a55002a7a09930dff2 100644 (file)
@@ -2,7 +2,7 @@ import argparse
 import sys
 from typing import List, Optional, Tuple, Type
 
-from knot_resolver.client.command import Command, CommandArgs, CompWords, get_subparsers_words, register_command
+from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
 from knot_resolver.utils.modeling.parsing import DataFormat, parse_json
 from knot_resolver.utils.requests import request
 
@@ -44,7 +44,7 @@ class MetricsCommand(Command):
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        return get_subparsers_words(parser._actions)  # noqa: SLF001
+        return comp_get_words(args, parser._actions)  # noqa: SLF001
 
     def run(self, args: CommandArgs) -> None:
         response = request(args.socket, "GET", "metrics/prometheus" if self.prometheus else "metrics/json")
index fa7465c178f22654c724c65207d631b51045f80c..fdb65bdee49eedfa06693bb66078ffe580720e69 100644 (file)
@@ -3,7 +3,7 @@ import json
 import sys
 from typing import List, Optional, Tuple, Type
 
-from knot_resolver.client.command import Command, CommandArgs, CompWords, get_subparsers_words, register_command
+from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
 from knot_resolver.datamodel import kres_config_json_schema
 from knot_resolver.utils.requests import request
 
@@ -35,7 +35,7 @@ class SchemaCommand(Command):
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        return get_subparsers_words(parser._actions)  # noqa: SLF001
+        return comp_get_words(args, parser._actions)  # noqa: SLF001
 
     def run(self, args: CommandArgs) -> None:
         if self.live:
index 141a2003047304140047efd91cfb44c21b6366e1..2347519ee6dfb8f9a0f3963e4474a910e3e92e4a 100644 (file)
@@ -3,7 +3,7 @@ import sys
 from pathlib import Path
 from typing import List, Tuple, Type
 
-from knot_resolver.client.command import Command, CommandArgs, CompWords, get_subparsers_words, register_command
+from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
 from knot_resolver.datamodel import KresConfig
 from knot_resolver.datamodel.globals import Context, reset_global_validation_context, set_global_validation_context
 from knot_resolver.utils.modeling import try_to_parse
@@ -41,7 +41,7 @@ class ValidateCommand(Command):
 
     @staticmethod
     def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
-        return get_subparsers_words(parser._actions)  # noqa: SLF001
+        return comp_get_words(args, parser._actions)  # noqa: SLF001
 
     def run(self, args: CommandArgs) -> None:
         if self.input_file:
index a1c8290b054c0230ecbf5206956e676662ed22a1..0df57f316146f84b58ce5f9ef82506dd26e6a3eb 100644 (file)
@@ -1,24 +1,27 @@
-#/usr/bin/env bash
+#!/usr/bin/env bash
 
 _kresctl_completion()
 {
     COMPREPLY=()
+    local args=""
     local words=""
-    local space_arg=""
     local cur="${COMP_WORDS[COMP_CWORD]}"
+    local opts=$(kresctl completion --bash --args "${COMP_WORDS[@]:1}")
 
-    # if the current word is empty
-    # we need to inform the kresctl client about it
-    if [[ -z "$cur" ]]; then
-        space_arg="--space"
+    # filter special opts
+    for opt in $opts
+    do
+    if [[ "$opt" == "#dirnames#" ]]; then
+        args="$args${args:+ }-d"
+    elif [[ "$opt" == "#filenames#" ]]; then
+        args="$args${args:+ }-f"
+    else
+        words="$words${words:+ }$opt"
     fi
+    done
 
-    # get words from the kresctl client
-    words=$(kresctl completion --bash ${space_arg} --args "${COMP_WORDS[@]:1}")
-
-    COMPREPLY=($(compgen  -W "${words}" -- "${cur}"))
-
+    COMPREPLY=($(compgen $args -W "${words}" -- "${cur}"))
     return 0
 }
 
-complete -o filenames -o dirnames -o nosort -F _kresctl_completion kresctl
+complete -o nosort -F _kresctl_completion kresctl