[FEATURE] KedroCLI structure get (kedro-org#1266)

Signed-off-by: Laurens Vijnck <[email protected]>

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
+        )

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