From 811728b508a6b91abb16c4f2047c2a4e1f82883a Mon Sep 17 00:00:00 2001 From: Jiri Klein <44288863+jiriklein@users.noreply.github.com> Date: Wed, 3 Nov 2021 10:53:47 +0000 Subject: [PATCH] [FEATURE] `KedroCLI` structure get (#1266) Signed-off-by: Laurens Vijnck --- tests/tools/test_cli.py | 154 ++++++++++++++++++++++++++++++++++++++++ tools/cli.py | 62 ++++++++++++++++ 2 files changed, 216 insertions(+) create mode 100644 tests/tools/test_cli.py create mode 100644 tools/cli.py diff --git a/tests/tools/test_cli.py b/tests/tools/test_cli.py new file mode 100644 index 0000000000..b6c46adf85 --- /dev/null +++ b/tests/tools/test_cli.py @@ -0,0 +1,154 @@ +"""Testing module for CLI tools""" +import shutil +from collections import namedtuple +from pathlib import Path + +import pytest + +from kedro import __version__ as kedro_version +from kedro.framework.cli.cli import KedroCLI, cli +from kedro.framework.startup import ProjectMetadata +from tools.cli import get_cli_structure + +REPO_NAME = "cli_tools_dummy_project" +PACKAGE_NAME = "cli_tools_dummy_package" +DEFAULT_KEDRO_COMMANDS = [ + "activate-nbstripout", + "build-docs", + "build-reqs", + "catalog", + "install", + "ipython", + "jupyter", + "lint", + "new", + "package", + "pipeline", + "registry", + "run", + "starter", + "test", +] + + +@pytest.fixture +def fake_root_dir(tmp_path): + try: + yield Path(tmp_path).resolve() + finally: + shutil.rmtree(tmp_path, ignore_errors=True) + + +@pytest.fixture +def fake_metadata(fake_root_dir): + metadata = ProjectMetadata( + fake_root_dir / REPO_NAME / "pyproject.toml", + PACKAGE_NAME, + "CLI Tools Testing Project", + fake_root_dir / REPO_NAME, + kedro_version, + fake_root_dir / REPO_NAME / "src", + ) + return metadata + + +class TestCLITools: + def test_get_cli_structure_raw(self, mocker, fake_metadata): + Module = namedtuple("Module", ["cli"]) + mocker.patch( + "kedro.framework.cli.cli.importlib.import_module", + return_value=Module(cli=cli), + ) + mocker.patch( + "kedro.framework.cli.cli._is_project", + return_value=True, + ) + mocker.patch( + "kedro.framework.cli.cli.bootstrap_project", + return_value=fake_metadata, + ) + kedro_cli = KedroCLI(fake_metadata.project_path) + raw_cli_structure = get_cli_structure(kedro_cli, get_help=False) + + # raw CLI structure tests + assert isinstance(raw_cli_structure, dict) + assert isinstance(raw_cli_structure["kedro"], dict) + + for k, v in raw_cli_structure["kedro"].items(): + assert isinstance(k, str) + assert isinstance(v, dict) + + assert sorted(list(raw_cli_structure["kedro"])) == sorted( + DEFAULT_KEDRO_COMMANDS + ) + + def test_get_cli_structure_depth(self, mocker, fake_metadata): + Module = namedtuple("Module", ["cli"]) + mocker.patch( + "kedro.framework.cli.cli.importlib.import_module", + return_value=Module(cli=cli), + ) + mocker.patch( + "kedro.framework.cli.cli._is_project", + return_value=True, + ) + mocker.patch( + "kedro.framework.cli.cli.bootstrap_project", + return_value=fake_metadata, + ) + kedro_cli = KedroCLI(fake_metadata.project_path) + raw_cli_structure = get_cli_structure(kedro_cli, get_help=False) + assert type(raw_cli_structure["kedro"]["new"]) == dict + assert sorted(list(raw_cli_structure["kedro"]["new"].keys())) == sorted( + [ + "--verbose", + "-v", + "--config", + "-c", + "--starter", + "-s", + "--checkout", + "--directory", + "--help", + ] + ) + # now check that once params and args are reached, the values are None + assert raw_cli_structure["kedro"]["new"]["--starter"] is None + assert raw_cli_structure["kedro"]["new"]["--checkout"] is None + assert raw_cli_structure["kedro"]["new"]["--help"] is None + assert raw_cli_structure["kedro"]["new"]["-c"] is None + + def test_get_cli_structure_help(self, mocker, fake_metadata): + Module = namedtuple("Module", ["cli"]) + mocker.patch( + "kedro.framework.cli.cli.importlib.import_module", + return_value=Module(cli=cli), + ) + mocker.patch( + "kedro.framework.cli.cli._is_project", + return_value=True, + ) + mocker.patch( + "kedro.framework.cli.cli.bootstrap_project", + return_value=fake_metadata, + ) + kedro_cli = KedroCLI(fake_metadata.project_path) + help_cli_structure = get_cli_structure(kedro_cli, get_help=True) + + assert isinstance(help_cli_structure, dict) + assert isinstance(help_cli_structure["kedro"], dict) + + for k, v in help_cli_structure["kedro"].items(): + assert isinstance(k, str) + if isinstance(v, dict): + for sub_key in v: + assert isinstance(help_cli_structure["kedro"][k][sub_key], str) + assert help_cli_structure["kedro"][k][sub_key].startswith( + "Usage: [OPTIONS]" + ) + elif isinstance(v, str): + assert v.startswith("Usage: [OPTIONS]") + + assert sorted(list(help_cli_structure["kedro"])) == sorted( + DEFAULT_KEDRO_COMMANDS + ) diff --git a/tools/cli.py b/tools/cli.py new file mode 100644 index 0000000000..98f1d88a51 --- /dev/null +++ b/tools/cli.py @@ -0,0 +1,62 @@ +from typing import Any, Dict, Union + +import click + + +def _recurse_cli( + cli_element: Union[click.Command, click.Group, click.CommandCollection], + ctx: click.Context, + io_dict: Dict[str, Any], + get_help: bool = False, +) -> None: + """ + Recursive function that checks the type of the command (key) and decides: + 1. In case of `click.Group` or `click.CommandCollection` (aggregate commands), + the function collects the name and recurses one layer deeper + for each sub-command. + 2. In case of `click.Command`, the terminus command has been reached. The function + collects the name, parameters and args, flattens them and saves them as + dictionary keys. + Args: + cli_element: CLI Collection as input for recursion, typically `KedroCLI`. + ctx: Click Context, created by the wrapper function. + io_dict: Input-output dictionary, mutated during the recursion. + get_help: Boolean fork - allows either: + raw structure - nested dictionary until final value of `None` + help structure - nested dictionary where leaves are `--help` cmd output + + Returns: + None (underlying `io_dict` is mutated by the recursion) + """ + if isinstance(cli_element, (click.Group, click.CommandCollection)): + element_name = cli_element.name or "kedro" + io_dict[element_name] = {} + for command_name in cli_element.list_commands(ctx): + _recurse_cli( # type: ignore + cli_element.get_command(ctx, command_name), + ctx, + io_dict[element_name], + get_help, + ) + + elif isinstance(cli_element, click.Command): + if get_help: # gets formatted CLI help incl params for printing + io_dict[cli_element.name] = cli_element.get_help(ctx) + else: # gets params for structure purposes + nested_parameter_list = [option.opts for option in cli_element.get_params(ctx)] + io_dict[cli_element.name] = dict.fromkeys( + [item for sublist in nested_parameter_list for item in sublist], None + ) + + +def get_cli_structure( + cli_obj: Union[click.Command, click.Group, click.CommandCollection], + get_help: bool = False, +) -> Dict[str, Any]: + """Convenience wrapper function for `_recurse_cli` to work within + `click.Context` and return a `dict`. + """ + output: Dict[str, Any] = dict() + with click.Context(cli_obj) as ctx: # type: ignore + _recurse_cli(cli_obj, ctx, output, get_help) + return output