Add basic CLI for test infra

Makefile

@@ -25,7 +25,7 @@ REPO_NAME := $(or ${REPO_NAME}, "")
 PULL_NUMBER := $(or ${PULL_NUMBER}, "")
 
 # lint
-LINT_CODE_STYLING_DIRS := src/tests src/triggers src/assisted_test_infra/test_infra src/assisted_test_infra/download_logs src/service_client src/consts src/virsh_cleanup
+LINT_CODE_STYLING_DIRS := src/tests src/triggers src/assisted_test_infra/test_infra src/assisted_test_infra/download_logs src/service_client src/consts src/virsh_cleanup src/cli
 
 # assisted-service
 SERVICE_BRANCH := $(or $(SERVICE_BRANCH), "master")
@@ -388,3 +388,7 @@ deploy_capi_env: start_minikube
 #########
 test_kube_api_parallel:
 	TEST=./src/tests/test_kube_api.py make test_parallel
+
+cli:
+	$(MAKE) start_load_balancer START_LOAD_BALANCER=true
+	JUNIT_REPORT_DIR=$(REPORTS) LOGGING_LEVEL="error" skipper run -i "python3 ${DEBUG_FLAGS} -m src.cli"

cli-requirements.txt

@@ -0,0 +1 @@
+assisted-service-client

requirements.txt

@@ -32,3 +32,5 @@ tqdm==4.62.3
 urllib3==1.26.8
 pyvmomi>=7.0.2
 waiting>=1.4.1
+prompt_toolkit==3.0.28
+tabulate==0.8.9

scripts/install_environment.sh

@@ -130,7 +130,7 @@ function install_runtime_container() {
 
 function install_packages() {
     echo "Installing dnf packages"
-    sudo dnf install -y make python3 python3-pip git jq bash-completion xinetd
+    sudo dnf install -y make python3.9 python3-pip python39-devel git jq bash-completion xinetd
     sudo systemctl enable --now xinetd
 
     echo "Installing python packages"

skipper.env

@@ -69,6 +69,7 @@ ES_PASS
 BACKUP_DESTINATION
 OLM_OPERATORS
 KUBE_API
+ENABLE_KUBEAPI
 INSTALLER_KUBECONFIG
 DEBUG
 FOCUS
@@ -114,4 +115,5 @@ HYPERSHIFT_BRANCH
 HYPERSHIFT_IMAGE
 DEPLOY_CAPI_PROVIDER
 HOLD_INSTALLATION
+PYTEST_JUNIT_FILE
 LOGGER_NAME

src/assisted_test_infra/test_infra/utils/env_var.py

@@ -1,4 +1,3 @@
-import functools
 from typing import Any, Callable, List, Optional
 
 from assisted_test_infra.test_infra.utils.utils import get_env
@@ -35,11 +34,15 @@ def __add__(self, other: "EnvVar"):
     def __str__(self):
         return f"{f'{self.__var_keys[0]}=' if len(self.__var_keys) > 0 else ''}{self.__value}"
 
+    @property
+    def var_keys(self):
+        return self.__var_keys
+
     @property
     def is_user_set(self):
         return self.__is_user_set
 
-    @functools.cached_property
+    @property
     def value(self):
         value = self.__default
         for key in self.__var_keys:

src/cli/__main__.py

@@ -0,0 +1,5 @@
+from cli.application import CliApplication
+
+if __name__ == "__main__" or __name__ == "src.cli.__main__":
+    cli = CliApplication()
+    cli.run()

src/cli/application.py

@@ -0,0 +1,40 @@
+import os
+
+from prompt_toolkit.shortcuts import message_dialog, set_title
+
+from tests.global_variables import DefaultVariables
+
+from .commands.commands_factory import InvalidCommandError
+from .prompt_handler import PromptHandler
+
+
+class CliApplication:
+    def __init__(self) -> None:
+        self._global_variables = DefaultVariables()
+        self._prompt_handler = PromptHandler(self._global_variables)
+
+    def _init_window(self):
+        os.system("clear")
+        set_title("Test Infra CLI")
+
+        if not self._global_variables.pull_secret:
+            message_dialog(
+                title="Pull Secret", text="Cant find PULL_SECRET, some functionality might not work as expected"
+            ).run()
+
+    def run(self):
+        self._init_window()
+        while True:
+            try:
+                command = self._prompt_handler.get_prompt_results()
+            except InvalidCommandError:
+                print("\033[1;31mError, invalid command!\033[0m")
+                continue
+            if command is None:
+                break
+            if not command.is_valid:
+                continue
+
+            command.handle()
+
+        print("Exiting ....")

src/cli/cli_utils.py

@@ -0,0 +1,39 @@
+import functools
+import json
+import subprocess
+from typing import List
+
+from tests.global_variables.default_variables import DefaultVariables
+
+__global_variables = DefaultVariables()
+
+
+def get_namespace() -> List[str]:
+    res = subprocess.check_output("kubectl get ns --output=json", shell=True)
+    try:
+        namespaces = json.loads(res)
+    except json.JSONDecodeError:
+        return []
+
+    return [ns["metadata"]["name"] for ns in namespaces["items"]]
+
+
+@functools.cache
+def get_boolean_keys():
+    bool_env_vars = [
+        __global_variables.get_env(k).var_keys
+        for k in __global_variables.__dataclass_fields__
+        if isinstance(getattr(__global_variables, k), bool)
+    ]
+    return [item for sublist in bool_env_vars for item in sublist]
+
+
+@functools.cache
+def get_env_args_keys():
+    env_vars = [__global_variables.get_env(k).var_keys for k in __global_variables.__dataclass_fields__]
+    return [item for sublist in env_vars for item in sublist]
+
+
+@functools.cache
+def inventory_client():
+    return __global_variables.get_api_client()

src/cli/commands/__init__.py

@@ -0,0 +1,3 @@
+from .commands_factory import CommandFactory
+
+__all__ = ["CommandFactory"]

src/cli/commands/command.py

@@ -0,0 +1,52 @@
+from abc import ABC, abstractmethod
+
+from prompt_toolkit.completion import DummyCompleter
+
+from service_client import log
+
+
+class Command(ABC):
+    _log_default_level = log.level
+
+    def __init__(self, text: str):
+        self._text = text
+        self._args = None
+
+    @property
+    def text(self):
+        return self._text
+
+    @property
+    def args(self):
+        return self._args
+
+    @property
+    @abstractmethod
+    def is_valid(self):
+        pass
+
+    @args.setter
+    def args(self, args: str):
+        self._args = [arg for arg in args.split(" ") if arg] if args else []
+
+    @classmethod
+    @abstractmethod
+    def get_completer(cls):
+        pass
+
+    @abstractmethod
+    def handle(self):
+        pass
+
+
+class DummyCommand(Command):
+    @property
+    def is_valid(self):
+        return False
+
+    @classmethod
+    def get_completer(cls):
+        return DummyCompleter()
+
+    def handle(self):
+        pass

src/cli/commands/commands_factory.py

@@ -0,0 +1,47 @@
+import functools
+from typing import Union
+
+from prompt_toolkit.completion import merge_completers
+
+from tests.global_variables import DefaultVariables
+
+from .. import cli_utils
+from ..completers import DynamicNestedCompleter
+from .command import Command, DummyCommand
+from .env_command import EnvCommand
+from .help_command import HelpCommand
+from .test_command import TestCommand
+
+
+class InvalidCommandError(Exception):
+    pass
+
+
+class CommandFactory:
+    _supported_commands = {
+        "": DummyCommand,
+        "test": TestCommand,
+        "list": EnvCommand,
+        "clear": EnvCommand,
+        "help": HelpCommand,
+    }
+
+    @classmethod
+    def get_command(cls, text: str) -> Union[Command, None]:
+        text = text if text else ""
+        factory = cls._supported_commands.get(text.split(" ")[0])
+        try:
+            return factory(text)
+        except TypeError as e:
+            raise InvalidCommandError(f"Error, invalid command {text}") from e
+
+    @classmethod
+    @functools.cache
+    def get_completers(cls):
+        commands = [c.get_completer() for c in {cmd for k, cmd in cls._supported_commands.items() if k}]
+        return merge_completers(commands)
+
+    @classmethod
+    def env_vars_completers(cls, global_variables: DefaultVariables):
+        keys = cli_utils.get_env_args_keys()
+        return DynamicNestedCompleter.from_nested_dict({k: None for k in keys}, global_variables=global_variables)

src/cli/commands/env_command.py

@@ -0,0 +1,38 @@
+import os
+
+from prompt_toolkit.completion import NestedCompleter
+from tabulate import tabulate
+
+from .. import cli_utils
+from .command import Command
+
+
+class EnvCommand(Command):
+    ENV_COMMAND_CLUSTERS = "clusters"
+    ENV_COMMAND_CLEAR = "clear"
+
+    @classmethod
+    def get_completer(cls):
+        return NestedCompleter.from_nested_dict({cls.ENV_COMMAND_CLEAR: None, "list": {"clusters": None}})
+
+    def command_list(self):
+        if self.args and self.args[0] == "clusters":
+            clusters = cli_utils.inventory_client().clusters_list()
+            clusters_data = [(f"{i + 1})", clusters[i]["id"], clusters[i]["name"]) for i in range(len(clusters))]
+
+            table = tabulate(clusters_data, headers=["", "Cluster ID", "Name"], tablefmt="fancy_grid")
+            print(table, "\n")
+
+    def handle(self):
+        command, *args = self.text.strip().split(" ")
+        self._args = args
+
+        if command == "clear":
+            os.system("clear")
+
+        elif command == "list":
+            self.command_list()
+
+    @property
+    def is_valid(self):
+        return self._text is not None

src/cli/commands/help_command.py

@@ -0,0 +1,29 @@
+from prompt_toolkit.completion import NestedCompleter
+from prompt_toolkit.shortcuts import message_dialog
+from tabulate import tabulate
+
+from .command import Command
+
+
+class HelpCommand(Command):
+    HELP_COMMAND = "help"
+
+    @classmethod
+    def get_completer(cls):
+        return NestedCompleter.from_nested_dict({cls.HELP_COMMAND: None})
+
+    def handle(self):
+        headers = ("", "Key", "Single Press", "Double Press")
+        keys = [
+            ("1", "Control + C", "Clear the text if exist else exit the cli"),
+            ("2", "Control + Q", "Exit the cli"),
+            ("3", "Tab", "Enter and navigate the autocomplete menu"),
+            ("4", "Right", "Step right or autocomplete from history"),
+        ]
+        table = tabulate(keys, headers=headers, tablefmt="fancy_grid")
+
+        message_dialog(title="Help", text=str(table)).run()
+
+    @property
+    def is_valid(self):
+        return self._text is not None

src/cli/commands/test_command.py

@@ -0,0 +1,65 @@
+import os
+import re
+import subprocess
+import uuid
+from copy import deepcopy
+
+import pytest
+from prompt_toolkit.completion import Completer, NestedCompleter
+
+from service_client import log
+
+from .command import Command
+
+
+class TestCommand(Command):
+    @classmethod
+    def get_completer(cls) -> Completer:
+        output = subprocess.check_output("python3 -m pytest --collect-only -q", shell=True, timeout=30)
+        pattern = r"((?P<file>src/tests/.*\.py)::(?P<class>.*)::(?P<func>.*))"
+        groups = [
+            match.groupdict() for match in [re.match(pattern, line) for line in output.decode().split("\n")] if match
+        ]
+
+        groups_set = set()
+        for group in groups:
+            groups_set.add((group["file"], group.get("func", "").split("[")[0]))
+
+        test_options = {}
+        for file, func in groups_set:
+            if file not in test_options:
+                test_options[file] = {}
+
+            test_options[file][func] = None
+
+        return NestedCompleter.from_nested_dict({"test": test_options})
+
+    def handle(self):
+        if not self._text:
+            return
+
+        environ_bak = deepcopy(os.environ)
+        try:
+            for arg_str in self._args:
+                var = re.match(r"(?P<key>.*)=(?P<value>.*)", arg_str).groupdict()
+                os.environ[var["key"]] = var["value"]
+
+            command = self._text.split(" ")
+            _command, file, func, *_ = [var for var in command if var]
+            junit_report_path = f"unittest_{str(uuid.uuid4())[:8]}.xml"
+            log.setLevel(self._log_default_level)
+            pytest.main([file, "-k", func, "--verbose", "-s", f"--junit-xml={junit_report_path}"])
+
+        except BaseException:
+            """Ignore any exception that might happen during test execution"""
+
+        finally:
+            from tests.config import reset_global_variables
+
+            os.environ.clear()
+            os.environ.update(environ_bak)
+            reset_global_variables()  # reset the config to its default state
+
+    @property
+    def is_valid(self):
+        return self._text is not None and self._args is not None