fix(3078, 3079): Resolve issues in Operations API Manager and Plugin Hooks (#3187)

* Pull from PR#3101

*

* fix!: Typo

* fix!: Conditional to Catch None Attribute Error Running Operations

* Planning Services attempts to fcall() and assume all abilities utilizes Builder Plugin and attempts to build payload.

* fix!: Caldera Operations Api Manager creates Builder Payloads during Runtime

* Conditional required to validate that only abilities utilizing builder plugin is used.

* Required encoded_command before building of ability and executor

* Update planning_svc.py

Added EOF new line to fix linting issue

* Update conftest.py

no longer able to use event_loop replaced all instances of event_loop with loop

* Fixes Flake8 Linter Run Errors

* fix: Direct Executer Function Call from Map

* Revert "Update conftest.py"

This reverts commit 3ecb69c.

* fix: additional executor checks

* Python Linting

Co-authored-by: Copilot <[email protected]>

* fix: flake8 linting error due to conditionals

* Pull Workflow from Upstream

---------

Co-authored-by: deacon-mp <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

app/api/v2/managers/operation_api_manager.py

@@ -102,6 +102,7 @@ async def create_potential_link(self, operation_id: str, data: dict, access: Bas
                                               file_svc=self.services['file_svc']))
         executor = self.build_executor(data=data.pop('executor', {}), agent=agent)
         ability = self.build_ability(data=data.pop('ability', {}), executor=executor)
+        await self._call_ability_plugin_hooks(ability, executor)
         link = Link.load(dict(command=encoded_command, plaintext_command=encoded_command, paw=agent.paw, ability=ability, executor=executor,
                               status=operation.link_status(), score=data.get('score', 0), jitter=data.get('jitter', 0),
                               cleanup=data.get('cleanup', 0), pin=data.get('pin', 0),
@@ -171,6 +172,13 @@ async def _construct_and_dump_source(self, source_id: str):
             source = (await self.services['data_svc'].locate('sources', match=dict(name='basic')))
         return SourceSchema().dump(source[0])
 
+    async def _call_ability_plugin_hooks(self, ability, executor):
+        """Calls any plugin hooks (at runtime) that exist for the ability and executor."""
+        if (hasattr(executor, 'HOOKS') and executor.HOOKS and
+                hasattr(executor, 'language') and executor.language and
+                executor.language in executor.HOOKS):
+            await executor.HOOKS[executor.language](ability, executor)
+
     async def validate_operation_state(self, data: dict, existing: Operation = None):
         if not existing:
             if data.get('state') in Operation.get_finished_states():

app/service/planning_svc.py

@@ -351,9 +351,7 @@ async def _generate_new_links(self, operation, agent, abilities, link_status):
             executor = await agent.get_preferred_executor(ability)
             if not executor:
                 continue
-
-            if executor.HOOKS and executor.language and executor.language in executor.HOOKS:
-                await executor.HOOKS[executor.language](ability, executor)
+            await self._call_ability_plugin_hooks(ability, executor)
             if executor.command:
                 link = Link.load(dict(command=self.encode_string(executor.test), paw=agent.paw, score=0,
                                       ability=ability, executor=executor, status=link_status,
@@ -392,6 +390,13 @@ async def _generate_cleanup_links(self, operation, agent, link_status):
                     links.append(lnk)
         return links
 
+    async def _call_ability_plugin_hooks(self, ability, executor):
+        """Calls any plugin hooks (at runtime) that exist for the ability and executor."""
+        if (hasattr(executor, 'HOOKS') and executor.HOOKS and
+                hasattr(executor, 'language') and executor.language and
+                executor.language in executor.HOOKS):
+            await executor.HOOKS[executor.language](ability, executor)
+
     @staticmethod
     async def _apply_adjustments(operation, links):
         """Apply operation source ability adjustments to links