Automation calls: Hand down flag indicating if called via automation …

…helper

We need to adjust the behaviour of some automation calls depending on the call
path.

Change-Id: Ie477386a10032e33af9a85a761354f88c44b86f6

cmk/base/automation_helper/_app.py

@@ -57,8 +57,13 @@ def redirect_stdin(stream: io.StringIO) -> Iterator[None]:
 
 
 class AutomationEngine(Protocol):
-    # TODO: remove `reload_config` when automation helper is fully integrated.
-    def execute(self, cmd: str, args: list[str], *, reload_config: bool) -> AutomationExitCode: ...
+    def execute(
+        self,
+        cmd: str,
+        args: list[str],
+        *,
+        called_from_automation_helper: bool,
+    ) -> AutomationExitCode: ...
 
 
 def get_application(
@@ -102,9 +107,10 @@ async def automation(request: Request, payload: AutomationPayload) -> Automation
             temporary_log_level(LOGGER, payload.log_level),
         ):
             try:
-                # TODO: remove `reload_config` when automation helper is fully integrated.
                 exit_code: int = engine.execute(
-                    payload.name, list(payload.args), reload_config=False
+                    payload.name,
+                    list(payload.args),
+                    called_from_automation_helper=True,
                 )
             except SystemExit as system_exit:
                 LOGGER.error("[automation] command raised a system exit exception.")

cmk/base/automations/__init__.py

@@ -48,16 +48,23 @@ def register(self, automation: "Automation") -> None:
             raise TypeError()
         self._automations[automation.cmd] = automation
 
-    # TODO: remove `reload_config` when automation helper is fully integrated.
     def execute(
-        self, cmd: str, args: list[str], *, reload_config: bool = True
+        self, cmd: str, args: list[str], *, called_from_automation_helper: bool = False
     ) -> AutomationExitCode:
         remaining_args, timeout = self._extract_timeout_from_args(args)
         with nullcontext() if timeout is None else Timeout(timeout, message="Action timed out."):
-            return self._execute(cmd, remaining_args, reload_config=reload_config)
+            return self._execute(
+                cmd,
+                remaining_args,
+                called_from_automation_helper=called_from_automation_helper,
+            )
 
     def _execute(
-        self, cmd: str, args: list[str], *, reload_config: bool = True
+        self,
+        cmd: str,
+        args: list[str],
+        *,
+        called_from_automation_helper: bool,
     ) -> AutomationExitCode:
         try:
             try:
@@ -68,7 +75,7 @@ def _execute(
                     f" (available: {', '.join(sorted(self._automations))})"
                 )
 
-            if reload_config and automation.needs_checks:
+            if not called_from_automation_helper and automation.needs_checks:
                 with (
                     tracer.start_as_current_span("load_all_plugins"),
                     redirect_stdout(open(os.devnull, "w")),
@@ -79,12 +86,12 @@ def _execute(
                         checks_dir=paths.checks_dir,
                     )
 
-            if reload_config and automation.needs_config:
+            if not called_from_automation_helper and automation.needs_config:
                 with tracer.start_as_current_span("load_config"):
                     config.load(validate_hosts=False)
 
             with tracer.start_as_current_span(f"execute_automation[{cmd}]"):
-                result = automation.execute(args)
+                result = automation.execute(args, called_from_automation_helper)
 
         except (MKAutomationError, MKTimeout) as e:
             console.error(f"{e}", file=sys.stderr)
@@ -123,7 +130,11 @@ class Automation(abc.ABC):
     needs_config = False
 
     @abc.abstractmethod
-    def execute(self, args: list[str]) -> ABCAutomationResult: ...
+    def execute(
+        self,
+        args: list[str],
+        called_from_automation_helper: bool,
+    ) -> ABCAutomationResult: ...
 
 
 #