Implement --temp for stop/start

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,10 @@ 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)
+    backend.commands.start(service_name, temp)
 
 
 @cli.command()
@@ -246,9 +247,10 @@ 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)
+    backend.commands.stop(service_name, temp)
 
 
 @cli.command()

src/edge_containers_cli/cmds/argo_commands.py

@@ -52,12 +52,12 @@ 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)
         cmd = f"argocd app set {self.target} -p services.{service_name}.enabled=true"
         shell.run_command(cmd, skip_on_dryrun=True)
 
-    def stop(self, service_name):
+    def stop(self, service_name, temp):
         self._check_service(service_name)
         cmd = f"argocd app set {self.target} -p services.{service_name}.enabled=false"
         shell.run_command(cmd, skip_on_dryrun=True)

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)
 

tests/data/argocd.yaml

@@ -37,9 +37,17 @@ restart:
     rsp: ""
 
 start:
-  - cmd: argocd app set namespace/bl01t-ea-test-01 -p global.enabled=true
+  - cmd: argocd app set namespace/bl01t -p services.bl01t-ea-test-01.enabled=true
+    rsp: ""
+
+start_temp:
+  - cmd: argocd app set namespace/bl01t -p services.bl01t-ea-test-01.enabled=true
     rsp: ""
 
 stop:
-  - cmd: argocd app set namespace/bl01t-ea-test-01 -p global.enabled=false
+  - cmd: argocd app set namespace/bl01t -p services.bl01t-ea-test-01.enabled=false
+    rsp: ""
+
+stop_temp:
+  - cmd: argocd app set namespace/bl01t -p services.bl01t-ea-test-01.enabled=false
     rsp: ""

tests/test_argocd.py

@@ -18,11 +18,21 @@ def test_start(mock_run, ARGOCD):
     mock_run.run_cli("start bl01t-ea-test-01")
 
 
+def test_start_temp(mock_run, ARGOCD):
+    mock_run.set_seq(ARGOCD.checks + ARGOCD.start_temp)
+    mock_run.run_cli("start bl01t-ea-test-01 --temp")
+
+
 def test_stop(mock_run, ARGOCD):
     mock_run.set_seq(ARGOCD.checks + ARGOCD.stop)
     mock_run.run_cli("stop bl01t-ea-test-01")
 
 
+def test_stop_temp(mock_run, ARGOCD):
+    mock_run.set_seq(ARGOCD.checks + ARGOCD.stop_temp)
+    mock_run.run_cli("stop bl01t-ea-test-01 --temp")
+
+
 def test_ps(mock_run, ARGOCD):
     expect = (
         "| name             | version | ready | deployed             |\n"