feat: add support for configuration

src/armonik_cli/cli.py

@@ -1,15 +1,21 @@
 import rich_click as click
 
 from armonik_cli import commands, __version__
+from armonik_cli.core import Configuration, console
 
 
 @click.group(name="armonik")
 @click.version_option(version=__version__, prog_name="armonik")
-def cli() -> None:
+@click.pass_context
+def cli(ctx: click.Context) -> None:
     """
     ArmoniK CLI is a tool to monitor and manage ArmoniK clusters.
     """
-    pass
+    if "help" not in ctx.args:
+        if not Configuration.default_path.exists():
+            console.print(f"Created configuration file at {Configuration.default_path}")
+            Configuration.create_default_if_not_exists()
 
 
-cli.add_command(commands.sessions)
+cli.add_command(commands.session)
+cli.add_command(commands.config)

src/armonik_cli/commands/__init__.py

@@ -1,4 +1,5 @@
-from .sessions import sessions
+from armonik_cli.commands.config import config
+from armonik_cli.commands.session import session
 
 
-__all__ = ["sessions"]
+__all__ = ["config", "session"]

src/armonik_cli/commands/config.py

@@ -0,0 +1,52 @@
+import json
+
+import rich_click as click
+
+from armonik_cli.core import console, Configuration, MutuallyExclusiveOption
+
+
+key_argument = click.argument("key", required=True, type=str, metavar="KEY")
+
+
+@click.group()
+def config():
+    """Display or change configuration settings for ArmoniK CLI."""
+    pass
+
+
+@config.command()
+@key_argument
+def get(key: str) -> None:
+    """Retrieve the value of a configuration setting by its KEY."""
+    config = Configuration.load_default()
+    if config.has(key):
+        return console.formatted_print({key: config.get(key)}, format="json")
+    return console.print(f"Warning: '{key}' is not a known configuration key.")
+
+# The function cannot be called 'set' directly, as this causes a conflict with the constructor of the built-in set object.
+@config.command("set")
+@key_argument
+@click.argument("value", required=True, type=str, metavar="VALUE")
+def set_(key: str, value: str) -> None:
+    """Update a configuration setting with a VALUE for the given KEY."""
+    config = Configuration.load_default()
+    if config.has(key):
+        config.set(key, value)
+        return console.print(f"Updated '{key}' configuration with value '{value}'.")
+    return console.print(f"Warning: '{key}' is not a known configuration key.")
+
+
+@config.command()
+def list() -> None:
+    """Display all configuration settings."""
+    config = Configuration.load_default()
+    console.formatted_print(config.to_dict(), format="json")
+
+
+@config.command()
+@click.option('--local', is_flag=True, help="Set to a local cluster without TLS enabled.", cls=MutuallyExclusiveOption, mutual=["interactive", "source"])
+@click.option('--interactive', is_flag=True, help="Use interactive prompts.", cls=MutuallyExclusiveOption, mutual=["local", "source"])
+@click.option('--source', type=click.Path(exists=True, file_okay=False, dir_okay=True), help="Use the deployment generated folder to retrieve connection details.", cls=MutuallyExclusiveOption, mutual=["local", "interactive"])
+def set_connection(local: bool, interactive: bool, source: str) -> None:
+    """Update all cluster connection configuration settings at once."""
+    raise NotImplementedError()

src/armonik_cli/commands/sessions.py → src/armonik_cli/commands/session.py

@@ -15,12 +15,12 @@
 
 
 @click.group(name="session")
-def sessions() -> None:
+def session() -> None:
     """Manage cluster sessions."""
     pass
 
 
-@sessions.command()
+@session.command()
 @base_command
 def list(endpoint: str, output: str, debug: bool) -> None:
     """List the sessions of an ArmoniK cluster."""
@@ -36,7 +36,7 @@ def list(endpoint: str, output: str, debug: bool) -> None:
     # console.print(f"\n{total} sessions found.")
 
 
-@sessions.command()
+@session.command()
 @session_argument
 @base_command
 def get(endpoint: str, output: str, session_id: str, debug: bool) -> None:
@@ -48,7 +48,7 @@ def get(endpoint: str, output: str, session_id: str, debug: bool) -> None:
         console.formatted_print(session, format=output, table_cols=SESSION_TABLE_COLS)
 
 
-@sessions.command()
+@session.command()
 @click.option(
     "--max-retries",
     type=int,
@@ -156,7 +156,7 @@ def create(
         console.formatted_print(session, format=output, table_cols=SESSION_TABLE_COLS)
 
 
-@sessions.command()
+@session.command()
 @click.confirmation_option("--confirm", prompt="Are you sure you want to cancel this session?")
 @session_argument
 @base_command
@@ -169,7 +169,7 @@ def cancel(endpoint: str, output: str, session_id: str, debug: bool) -> None:
         console.formatted_print(session, format=output, table_cols=SESSION_TABLE_COLS)
 
 
-@sessions.command()
+@session.command()
 @session_argument
 @base_command
 def pause(endpoint: str, output: str, session_id: str, debug: bool) -> None:
@@ -181,7 +181,7 @@ def pause(endpoint: str, output: str, session_id: str, debug: bool) -> None:
         console.formatted_print(session, format=output, table_cols=SESSION_TABLE_COLS)
 
 
-@sessions.command()
+@session.command()
 @session_argument
 @base_command
 def resume(endpoint: str, output: str, session_id: str, debug: bool) -> None:
@@ -193,7 +193,7 @@ def resume(endpoint: str, output: str, session_id: str, debug: bool) -> None:
         console.formatted_print(session, format=output, table_cols=SESSION_TABLE_COLS)
 
 
-@sessions.command()
+@session.command()
 @click.confirmation_option("--confirm", prompt="Are you sure you want to close this session?")
 @session_argument
 @base_command
@@ -206,7 +206,7 @@ def close(endpoint: str, output: str, session_id: str, debug: bool) -> None:
         console.formatted_print(session, format=output, table_cols=SESSION_TABLE_COLS)
 
 
-@sessions.command()
+@session.command()
 @click.confirmation_option("--confirm", prompt="Are you sure you want to purge this session?")
 @session_argument
 @base_command
@@ -219,7 +219,7 @@ def purge(endpoint: str, output: str, session_id: str, debug: bool) -> None:
         console.formatted_print(session, format=output, table_cols=SESSION_TABLE_COLS)
 
 
-@sessions.command()
+@session.command()
 @click.confirmation_option("--confirm", prompt="Are you sure you want to delete this session?")
 @session_argument
 @base_command
@@ -232,7 +232,7 @@ def delete(endpoint: str, output: str, session_id: str, debug: bool) -> None:
         console.formatted_print(session, format=output, table_cols=SESSION_TABLE_COLS)
 
 
-@sessions.command()
+@session.command()
 @click.option(
     "--clients-only",
     is_flag=True,

src/armonik_cli/core/__init__.py

@@ -1,6 +1,7 @@
+from armonik_cli.core.config import Configuration
 from armonik_cli.core.console import console
 from armonik_cli.core.decorators import base_command
-from armonik_cli.core.params import KeyValuePairParam, TimeDeltaParam
+from armonik_cli.core.params import KeyValuePairParam, TimeDeltaParam, MutuallyExclusiveOption
 
 
-__all__ = ["base_command", "KeyValuePairParam", "TimeDeltaParam", "console"]
+__all__ = ["base_command", "KeyValuePairParam", "TimeDeltaParam", "console", "Configuration", "MutuallyExclusiveOption"]

src/armonik_cli/core/config.py

@@ -0,0 +1,44 @@
+import json
+
+import rich_click as click
+
+from pathlib import Path
+from typing import Union, Dict
+
+
+class Configuration:
+    default_path = Path(click.get_app_dir("armonik_cli")) / "config"
+    _config_keys = ["endpoint"]
+    _default_config = {"endpoint": None}
+
+    def __init__(self, endpoint: str) -> None:
+        self.endpoint = endpoint
+
+    @classmethod
+    def create_default_if_not_exists(cls) -> None:
+        cls.default_path.parent.mkdir(exist_ok=True)
+        if not (cls.default_path.is_file() and cls.default_path.exists()):
+            with cls.default_path.open("w") as config_file:
+                config_file.write(json.dumps(cls._default_config, indent=4))
+
+    @classmethod
+    def load_default(cls) -> "Configuration":
+        with cls.default_path.open("r") as config_file:
+            return cls(**json.loads(config_file.read()))
+
+    def has(self, key: str) -> bool:
+        if key in self._config_keys:
+            return True
+        return False
+
+    def get(self, key: str) -> Union[str, None]:
+        if self.has(key):
+            return getattr(self, key)
+        return None
+
+    def set(self, key: str, value: str) -> None:
+        if self.has(key):
+            setattr(self, key, value)
+
+    def to_dict(self) -> Dict[str, str]:
+        return {key: getattr(self, key) for key in self._config_keys}

src/armonik_cli/core/params.py

@@ -99,3 +99,18 @@ def _parse_time_delta(time_str: str) -> timedelta:
             seconds=int(sec),
             milliseconds=int(microseconds.ljust(3, "0")),  # Ensure 3 digits for milliseconds
         )
+
+
+class MutuallyExclusiveOption(click.Option):
+    def __init__(self, *args, **kwargs):
+        self.mutual = set(kwargs.pop('mutual', []))
+        if self.mutual:
+            kwargs['help'] = f"{kwargs.get('help', '')} This option cannot be used together with {' or '.join(self.mutual)}."
+        super().__init__(*args, **kwargs)
+
+    def handle_parse_result(self, ctx, opts, args):
+        mutex = self.mutual.intersection(opts)
+        if mutex and self.name in opts:
+            raise click.UsageError(f"Illegal usage: `{self.name}` cannot be used together with '{''.join(mutex)}'.")
+
+        return super().handle_parse_result(ctx, opts, args)

tests/commands/test_config.py

@@ -0,0 +1,19 @@
+import pytest
+
+from armonik_cli.core import Configuration
+
+from conftest import run_cmd_and_assert_exit_code, reformat_cmd_output
+
+
+@pytest.mark.parametrize(
+    ("cmd", "has_return", "get_return", "output"),
+    [
+        ("config get endpoint", True, "endpoint", {"endpoint": "endpoint"}),
+        # ("config get not", False, None, ""),
+    ],
+)
+def test_session_list(mocker, cmd, has_return, get_return, output):
+    mocker.patch.object(Configuration, "has", return_value=has_return)
+    mocker.patch.object(Configuration, "get", return_value=get_return)
+    result = run_cmd_and_assert_exit_code(cmd)
+    assert reformat_cmd_output(result.output, deserialize=True) == output