Refactoring/260 added unit tests for cli param wrappers (#261)

* Added unit tests for SCS CLI parameter wrappers
* Updated exasol-transformers-extension

Author: ckunki

doc/changes/unreleased.md

@@ -11,3 +11,4 @@ This release marks most of the NC's dependencies in file `pyproject.toml` as _op
 ## Refactorings
 
 * #253: Made dependencies optional in file `pyproject.toml`
+* #260: Added unit tests for CLI param wrappers

exasol/nb_connector/cli/options/common.py

@@ -2,6 +2,7 @@
 
 import click
 
+from exasol.nb_connector.ai_lab_config import AILabConfig as CKey
 from exasol.nb_connector.cli.param_wrappers import (
     ScsArgument,
     ScsOption,
@@ -31,5 +32,6 @@
         metavar="DB_SCHEMA",
         type=str,
         help="Database schema for installing UDFs of Exasol extensions",
+        scs_key=CKey.db_schema,
     )
 ]

exasol/nb_connector/cli/param_wrappers.py

@@ -2,32 +2,71 @@
 Wrappers for adding custom properties to click parameters, e.g. SCS key.
 """
 
+import getpass
+import os
+import re
+from abc import abstractmethod
+from typing import Any
+
 import click
 
 from exasol.nb_connector.ai_lab_config import AILabConfig as CKey
+from exasol.nb_connector.cli import reporting as report
+from exasol.nb_connector.secret_store import Secrets
 
 
-class ScsArgument:
+class ScsParam:
     """
-    Represents a CLI argument for the SCS command.
+    Abstract base class for ScsArgument and ScsOption.
     """
 
-    def __init__(self, *args, scs_key: CKey | None = None, **kwargs):
-        self._args = args
+    def __init__(self, scs_key: CKey | None = None, **kwargs):
         self.scs_key = scs_key
         self._kwargs = kwargs
 
+    @property
+    def arg_name(self) -> str:
+        return ""
+
+    def needs_entry(self, scs: Secrets) -> bool:
+        return False
+
+    @property
+    def default(self) -> Any:
+        return self._kwargs.get("default")
+
+    def displayed_value(self, scs: Secrets) -> str | None:
+        return None
+
+    @abstractmethod
+    def decorate(self, func):
+        """
+        This method is to be called when decorating the functions in the
+        actual CLI declaration.
+        """
+        pass
+
+
+class ScsArgument(ScsParam):
+    """
+    Represents a CLI argument for the SCS command.
+    """
+
+    def __init__(self, name: str, scs_key: CKey | None = None, **kwargs):
+        super().__init__(scs_key, **kwargs)
+        self.name = name
+
     def decorate(self, func):
         """
         This method is to be called when decorating the functions in the
         actual CLI declaration. Hence, ScsArgument calls click.argument()
         under the hood.
         """
-        decorator = click.argument(*self._args, **self._kwargs)
+        decorator = click.argument(self.name, **self._kwargs)
         return decorator(func)
 
 
-class ScsOption(ScsArgument):
+class ScsOption(ScsParam):
     """
     CLI option for saving and checking values to the Secure Configuration
     Storage (SCS).
@@ -53,30 +92,70 @@ class ScsOption(ScsArgument):
 
     def __init__(
         self,
+        cli_option,
         *args,
         scs_key: CKey | None = None,
         scs_alternative_key: CKey | None = None,
         scs_required: bool = True,
         get_default_from: str | None = None,
         **kwargs,
     ):
-        super().__init__(*args, scs_key=scs_key, **kwargs)
+        super().__init__(scs_key=scs_key, **kwargs)
+        self._cli_option = cli_option
+        self._args = args
         self.scs_alternative_key = scs_alternative_key
         self.scs_required = scs_required
         self.get_default_from = get_default_from
 
+    def cli_option(self, full=False) -> str:
+        raw = self._cli_option
+        return raw if full else re.sub(r"/--.*$", "", raw)
+
+    @property
+    def arg_name(self) -> str:
+        for arg in self._args:
+            if not arg.startswith("--"):
+                return arg
+        name = self.cli_option()
+        return name[2:].replace("-", "_")
+
     def decorate(self, func):
         """
         This method is to be called when decorating the functions in the
         actual CLI declaration. ScsOption calls click.option().
         """
         decorator = click.option(
+            self._cli_option,
             *self._args,
             **self._kwargs,
             show_default=True,
         )
         return decorator(func)
 
+    def displayed_value(self, scs: Secrets) -> str | None:
+        return scs.get(self.scs_key) if self.scs_key else None
+
+    def needs_entry(self, scs: Secrets) -> bool:
+        """
+        Return True, if the current option is configured to be saved to
+        the SCS but SCS does not yet contain a value.
+        """
+
+        def has_value() -> bool:
+            if not self.scs_key:
+                return False
+            if scs.get(self.scs_key) is not None:
+                return True
+            if alt := self.scs_alternative_key:
+                return scs.get(alt) is not None
+            return False
+
+        return bool(self.scs_key) and self.scs_required and not has_value()
+
+    def __repr__(self) -> str:
+        cls_name = type(self).__name__
+        return f"{cls_name}<{self.cli_option(full=True)}>"
+
 
 class ScsSecretOption(ScsOption):
     """
@@ -107,6 +186,22 @@ def __init__(
         self.prompt = prompt
         self.name = name
 
+    def displayed_value(self, scs: Secrets) -> str | None:
+        return "****" if self.scs_key and scs.get(self.scs_key) else None
+
+    def get_secret(self, interactive: Any) -> str:
+        """
+        If interactive is True and the related environment variable is not
+        set then ask for the secret interactively.
+        """
+        if value := os.getenv(self.envvar):
+            report.info(f"Reading {self.name} from environment variable {self.envvar}.")
+            return value
+        if not interactive:
+            return ""
+        prompt = f"{self.prompt} (option {self.name}): "
+        return getpass.getpass(prompt)
+
 
 def add_params(scs_options: list[ScsArgument]):
     def multi_decorator(func):

exasol/nb_connector/cli/reporting.py

@@ -0,0 +1,17 @@
+import click
+
+
+def info(text: str):
+    click.echo(click.style(text, fg="green"))
+
+
+def success(text: str):
+    click.echo(click.style(text, fg="green"))
+
+
+def error(text: str):
+    click.echo(click.style(f"Error: {text}", fg="bright_red"))
+
+
+def warning(text: str):
+    click.echo(click.style(f"Warning: {text}", fg="yellow"))

poetry.lock

@@ -116,6 +116,8 @@ force_grid_wrap = 2
 
 
 [tool.pylint.master]
+# Module exasol.ai.text is compiled via Cython, released as binary.
+# So pylint cannot verify types there.
 ignored-modules = [ "exasol.ai.text" ]
 fail-under = 8.0
 

pyproject.toml

@@ -0,0 +1,26 @@
+from exasol.nb_connector.ai_lab_config import AILabConfig as CKey
+from exasol.nb_connector.ai_lab_config import StorageBackend
+
+
+class ScsMock:
+    """
+    Instead of using a real Secure Configuration Storage, this mock
+    simpulates it using a simple dict().
+    """
+
+    def __init__(
+        self,
+        backend: StorageBackend | None = None,
+        use_itde: bool | None = None,
+    ):
+        self._dict = dict()
+        if backend:
+            self.save(CKey.storage_backend, backend.name)
+        if use_itde is not None:
+            self.save(CKey.use_itde, str(use_itde))
+
+    def save(self, key: CKey, value: str) -> None:
+        self._dict[key.name] = str(value)
+
+    def get(self, key: CKey, default: str | None = None) -> str | None:
+        return self._dict.get(key.name, default)