Change enable to act on root app (#165)

* Change enable to act on root app

* Implement stop/start using git and --temp for direct communication to argocd API via patches

* Fix tests, some linting

* Code clean up

src/edge_containers_cli/__main__.py

@@ -23,15 +23,22 @@ def version_callback(value: bool):
 
 def backend_callback(ctx: typer.Context, backend: ECBackends):
     init_backend(backend)
+
     # Dynamically drop any method not implemented
     not_implemented = [
-        mthd.replace("_", "-") for mthd in ec_backend.get_notimplemented()
+        mthd.replace("_", "-") for mthd in ec_backend.get_notimplemented_cmds()
     ]
+    typer_commands = ctx.command.commands  # type: ignore
     for command in not_implemented:
-        typer_commands = ctx.command.commands  # type: ignore
         if command in typer_commands:
             typer_commands.pop(command)
 
+    # Dynamically drop any cli options as specified
+    for cmd_name, drop_params in ec_backend.get_notimplemented_params().items():
+        for index, param in enumerate(typer_commands[cmd_name].params):
+            if param.name in drop_params:
+                typer_commands[cmd_name].params.pop(index)
+
     return backend.value
 
 

src/edge_containers_cli/backend.py

@@ -20,6 +20,13 @@ def __init__(self) -> None:
         self._Commands: type | None = None
         self._commands: Commands | None = None
 
+    @property
+    def Commands(self):
+        if self._Commands is None:
+            raise BackendError("Backend commands not set")
+        else:
+            return self._Commands
+
     @property
     def commands(self):
         if self._commands is None:
@@ -44,7 +51,7 @@ def set_context(self, context: ECContext):
             self._cxt = context
             self._commands = self._Commands(context)
 
-    def get_notimplemented(self) -> list[str]:
+    def get_notimplemented_cmds(self) -> list[str]:
         notimplemented = []
         if self._Commands is None:
             return []
@@ -54,6 +61,9 @@ def get_notimplemented(self) -> list[str]:
                     notimplemented.append(command)
         return notimplemented
 
+    def get_notimplemented_params(self) -> dict[str, list[str]]:
+        return self.Commands.params_opt_out
+
 
 backend = Backend()
 

src/edge_containers_cli/cli.py

@@ -234,9 +234,14 @@ def start(
     service_name: str = typer.Argument(
         ..., help="Name of the service container to start", autocompletion=all_svc
     ),
+    temp: bool = typer.Option(False, help="Directly overrides the controller values"),
 ):
     """Start a service"""
-    backend.commands.start(service_name)
+    try:
+        backend.commands.start(service_name, temp)
+    except GitError as e:
+        msg = f"{str(e)} - Try 'ec start <service> --temp' to bypass the git server"
+        raise GitError(msg) from e
 
 
 @cli.command()
@@ -246,9 +251,14 @@ def stop(
         help="Name of the service container to stop",
         autocompletion=running_svc,
     ),
+    temp: bool = typer.Option(False, help="Directly overrides the controller values"),
 ):
     """Stop a service"""
-    backend.commands.stop(service_name)
+    try:
+        backend.commands.stop(service_name, temp)
+    except GitError as e:
+        msg = f"{str(e)} - Try ec stop <service> --temp to bypass the git server"
+        raise GitError(msg) from e
 
 
 @cli.command()

src/edge_containers_cli/cmds/argo_commands.py

@@ -6,12 +6,14 @@
 
 import webbrowser
 from datetime import datetime
+from pathlib import Path
 
 import polars
 from ruamel.yaml import YAML
 
 from edge_containers_cli.cmds.commands import CommandError, Commands, ServicesDataFrame
 from edge_containers_cli.definitions import ECContext
+from edge_containers_cli.git import set_values
 from edge_containers_cli.globals import TIME_FORMAT
 from edge_containers_cli.shell import ShellError, shell
 
@@ -21,6 +23,36 @@ def extract_ns_app(target: str) -> tuple[str, str]:
     return namespace, app
 
 
+def patch_value(target: str, key: str, value: str | bool | int):
+    cmd_temp_ = f"argocd app set {target} -p {key}={value}"
+    shell.run_command(cmd_temp_, skip_on_dryrun=True)
+    cmd_sync = f"argocd app sync {target} --apply-out-of-sync-only"
+    shell.run_command(cmd_sync, skip_on_dryrun=True)
+
+
+def push_value(target: str, key: str, value: str | bool | int):
+    # Get source details
+    app_resp = shell.run_command(
+        f"argocd app get {target} -o yaml",
+    )
+    app_dicts = YAML(typ="safe").load(app_resp)
+    repo_url = app_dicts["spec"]["source"]["repoURL"]
+    path = Path(app_dicts["spec"]["source"]["path"])
+
+    set_values(
+        repo_url,
+        path / "values.yaml",
+        key,
+        value,
+    )
+
+    # Sync & release a possible patched value
+    cmd_sync = f"argocd app sync {target} --apply-out-of-sync-only"
+    shell.run_command(cmd_sync, skip_on_dryrun=True)
+    cmd_unset = f"argocd app unset {target} -p {key}"
+    shell.run_command(cmd_unset, skip_on_dryrun=True)
+
+
 class ArgoCommands(Commands):
     """
     A class for implementing the Kubernetes based commands
@@ -52,17 +84,19 @@ def restart(self, service_name):
         )
         shell.run_command(cmd, skip_on_dryrun=True)
 
-    def start(self, service_name):
+    def start(self, service_name, temp):
         self._check_service(service_name)
-        namespace, app = extract_ns_app(self.target)
-        cmd = f"argocd app set {namespace}/{service_name} -p global.enabled=true"
-        shell.run_command(cmd, skip_on_dryrun=True)
+        if temp:
+            patch_value(self.target, f"ec_services.{service_name}.enabled", True)
+        else:
+            push_value(self.target, f"ec_services.{service_name}.enabled", True)
 
-    def stop(self, service_name):
+    def stop(self, service_name, temp):
         self._check_service(service_name)
-        namespace, app = extract_ns_app(self.target)
-        cmd = f"argocd app set {namespace}/{service_name} -p global.enabled=false"
-        shell.run_command(cmd, skip_on_dryrun=True)
+        if temp:
+            patch_value(self.target, f"ec_services.{service_name}.enabled", False)
+        else:
+            push_value(self.target, f"ec_services.{service_name}.enabled", False)
 
     def _get_logs(self, service_name, prev) -> str:
         namespace, app = extract_ns_app(self.target)

src/edge_containers_cli/cmds/commands.py

@@ -39,6 +39,8 @@ class Commands(ABC):
     Methods not exposed to the CLI should be private
     """
 
+    params_opt_out: dict[str, list[str]] = {}
+
     def __init__(self, ctx: ECContext):
         self._target = ctx.target
         self._target_valid = False
@@ -72,70 +74,70 @@ def log_url(self):
             log.debug("log_url = %s", self._log_url)
             return self._log_url
 
-    def attach(self, service_name: str):
+    def attach(self, service_name: str) -> None:
         raise NotImplementedError
 
-    def delete(self, service_name: str):
+    def delete(self, service_name: str) -> None:
         raise NotImplementedError
 
-    def deploy(self, service_name: str, version: str, args: str):
+    def deploy(self, service_name: str, version: str, args: str) -> None:
         raise NotImplementedError
 
-    def deploy_local(self, svc_instance: Path, args: str):
+    def deploy_local(self, svc_instance: Path, args: str) -> None:
         raise NotImplementedError
 
-    def exec(self, service_name: str):
+    def exec(self, service_name: str) -> None:
         raise NotImplementedError
 
-    def logs(self, service_name: str, prev: bool):
+    def logs(self, service_name: str, prev: bool) -> None:
         raise NotImplementedError
 
-    def log_history(self, service_name: str):
+    def log_history(self, service_name: str) -> None:
         raise NotImplementedError
 
-    def ps(self, running_only: bool):
+    def ps(self, running_only: bool) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def restart(self, service_name: str):
+    def restart(self, service_name: str) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def start(self, service_name: str):
+    def start(self, service_name: str, temp: bool) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def stop(self, service_name: str):
+    def stop(self, service_name: str, temp: bool) -> None:
         raise NotImplementedError
 
-    def template(self, svc_instance: Path, args: str):
+    def template(self, svc_instance: Path, args: str) -> None:
         raise NotImplementedError
 
     @abstractmethod
     def _get_services(self, running_only: bool) -> ServicesDataFrame:
         raise NotImplementedError
 
-    def _ps(self, running_only: bool):
+    def _ps(self, running_only: bool) -> None:
         services_df = self._get_services(running_only)
         print(services_df)
 
     @abstractmethod
     def _get_logs(self, service_name: str, prev: bool) -> str:
         raise NotImplementedError
 
-    def _logs(self, service_name: str, prev: bool):
+    def _logs(self, service_name: str, prev: bool) -> None:
         print(self._get_logs(service_name, prev))
 
-    def _validate_target(self):
+    def _validate_target(self) -> None:
         raise NotImplementedError
 
-    def _running_services(self):
+    def _running_services(self) -> list[str]:
         return self._get_services(running_only=True)["name"].to_list()
 
-    def _all_services(self):
+    def _all_services(self) -> list[str]:
         return self._get_services(running_only=False)["name"].to_list()
 
-    def _check_service(self, service_name: str):
+    def _check_service(self, service_name: str) -> None:
         services_list = self._get_services(running_only=False)["name"]
         if service_name in services_list:
             pass

src/edge_containers_cli/cmds/k8s_commands.py

@@ -23,6 +23,11 @@ class K8sCommands(Commands):
     A class for implementing the Kubernetes based commands
     """
 
+    params_opt_out = {
+        "stop": ["temp"],
+        "start": ["temp"],
+    }
+
     def __init__(
         self,
         ctx: ECContext,
@@ -84,14 +89,14 @@ def restart(self, service_name):
             f"kubectl delete -n {self.target} {pod_name}", skip_on_dryrun=True
         )
 
-    def start(self, service_name):
+    def start(self, service_name, temp):
         self._check_service(service_name)
         shell.run_command(
             f"kubectl scale -n {self.target} statefulset {service_name} --replicas=1",
             skip_on_dryrun=True,
         )
 
-    def stop(self, service_name):
+    def stop(self, service_name, temp):
         self._check_service(service_name)
         shell.run_command(
             f"kubectl scale -n {self.target} statefulset {service_name} --replicas=0 ",

src/edge_containers_cli/cmds/monitor.py

@@ -446,7 +446,7 @@ def action_start_ioc(self) -> None:
         def check_start(start: bool | None) -> None:
             """Called when StartScreen is dismissed."""
             if start:
-                self.commands.start(service_name)
+                self.commands.start(service_name, False)
 
         self.push_screen(StartScreen(service_name), check_start)
 
@@ -457,7 +457,7 @@ def action_stop_ioc(self) -> None:
         def check_stop(stop: bool | None) -> None:
             """Called when StopScreen is dismissed."""
             if stop:
-                self.commands.stop(service_name)
+                self.commands.stop(service_name, False)
 
         self.push_screen(StopScreen(service_name), check_stop)
 

src/edge_containers_cli/git.py

@@ -9,14 +9,40 @@
 from natsort import natsorted
 
 from edge_containers_cli.logging import log
-from edge_containers_cli.shell import shell
-from edge_containers_cli.utils import chdir, new_workdir
+from edge_containers_cli.shell import ShellError, shell
+from edge_containers_cli.utils import YamlFile, chdir, new_workdir
 
 
 class GitError(Exception):
     pass
 
 
+def set_values(repo_url: str, file: Path, key: str, value: str | bool | int) -> None:
+    """
+    sets an existing key value pair in a yaml file and push the changes
+    """
+    with new_workdir() as path:
+        try:
+            shell.run_command(f"git clone --depth=1 {repo_url} {path}")
+            with chdir(path):  # From python 3.11 can use contextlib.chdir(working_dir)
+                file_data = YamlFile(file)
+
+                value_repo = file_data.get_key(key)
+                if value_repo == value:
+                    log.debug(f"{key} already set as {value}")
+                else:
+                    file_data.set_key(key, value)
+                    file_data.dump_file()
+
+                    commit_msg = f"Set {key}={value} in {file}"
+                    shell.run_command("git add .")
+                    shell.run_command(f'git commit -m "{commit_msg}"')
+                    shell.run_command("git push", skip_on_dryrun=True)
+
+        except (FileNotFoundError, ShellError) as e:
+            raise GitError(str(e)) from e
+
+
 def create_version_map(
     repo: str, root_dir: Path, working_dir: Path, shared: list[str] | None = None
 ) -> dict[str, list[str]]: