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,86 @@
+import json
+
+import rich_click as click
+
+from pathlib import Path
+
+from rich.prompt import Prompt
+
+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."""
+    endpoint = None
+    if local:
+        endpoint = "172.17.119.85:5001"
+    elif interactive:
+        endpoint = Prompt.get_input(console, "Endpoint: ", password=False)
+    elif source:
+        with (Path(source) / "armonik-output.json").open() as output_file:
+            endpoint = json.loads(output_file.read())["armonik"]["control_plane_url"]
+    else:
+        return
+    config = Configuration.load_default()
+    config.set("endpoint", endpoint)

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,14 @@
+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,49 @@
+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)
+            self._save()
+
+    def to_dict(self) -> Dict[str, str]:
+        return {key: getattr(self, key) for key in self._config_keys}
+
+    def _save(self):
+        with self.default_path.open("w") as config_file:
+            config_file.write(json.dumps(self.to_dict(), indent=4))

src/armonik_cli/core/params.py

@@ -99,3 +99,22 @@ 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,43 @@
+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, "Warning: 'not' is not a known configuration key."),
+    ],
+)
+def test_config_get(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=has_return) == output
+
+
+@pytest.mark.parametrize(
+    ("cmd", "has_return", "output"),
+    [
+        ("config set endpoint value", True, "Updated 'endpoint' configuration with value 'value'."),
+        ("config set not value", False, "Warning: 'not' is not a known configuration key."),
+    ],
+)
+def test_config_set(mocker, cmd, has_return, output):
+    mocker.patch.object(Configuration, "has", return_value=has_return)
+    result = run_cmd_and_assert_exit_code(cmd)
+    assert reformat_cmd_output(result.output) == output
+
+
+def test_config_list(mocker):
+    mock_config = {"config": "config"}
+    mocker.patch.object(Configuration, "to_dict", return_value=mock_config)
+    result = run_cmd_and_assert_exit_code("config list")
+    assert reformat_cmd_output(result.output, deserialize=True) == mock_config
+
+
+def test_config_set_connection():
+    pass

tests/core/test_params.py

@@ -39,3 +39,29 @@ def test_timedelta_parm_success(input, output):
 def test_timedelta_parm_fail(input):
     with pytest.raises(click.BadParameter):
         assert TimeDeltaParam().convert(input, None, None)
+
+
+# Define a simple Click command using the custom MutuallyExclusiveOption
+@click.command()
+@click.option('--option-a', cls=MutuallyExclusiveOption, mutual=['option-b'])
+@click.option('--option-b', cls=MutuallyExclusiveOption, mutual=['option-a'])
+def test_command(option_a, option_b):
+    pass
+
+def test_no_conflict():
+    runner = CliRunner()
+    result = runner.invoke(test_command, ['--option-a', 'value_a'])
+    assert result.exit_code == 0
+
+def test_mutual_exclusion_error():
+    runner = CliRunner()
+    result = runner.invoke(test_command, ['--option-a', 'value_a', '--option-b', 'value_b'])
+    assert result.exit_code != 0
+    assert "Illegal usage: `option-a` cannot be used together with 'option-b'." in result.output
+
+def test_other_non_conflicting_option():
+    """Test that the command works when using the other non-conflicting option."""
+    runner = CliRunner()
+    result = runner.invoke(test_command, ['--option-b', 'value_b'])
+    assert result.exit_code == 0
+    assert "option_a: None, option_b: value_b" in result.output