IWF-467 add decider trigger type: ANY_COMMAND_COMBINATION (#90)

* IWF-467 add decider trigger type: ANY_COMMAND_COMBINATION

---------

Co-authored-by: Nathan Givens <[email protected]>

Author: N-give

iwf/command_request.py

@@ -3,6 +3,7 @@
 
 from iwf.errors import WorkflowDefinitionError
 from iwf.iwf_api.models import CommandWaitingType
+from iwf.iwf_api.models.command_combination import CommandCombination
 from iwf.iwf_api.models.command_request import (
     CommandRequest as IdlCommandRequest,
 )
@@ -21,8 +22,7 @@ class TimerCommand:
     @classmethod
     def by_seconds(cls, duration_seconds: int, command_id: Optional[str] = None):
         return TimerCommand(
-            command_id if command_id is not None else "",
-            duration_seconds,
+            command_id if command_id is not None else "", duration_seconds
         )
 
 
@@ -46,7 +46,8 @@ class SignalChannelCommand:
     @classmethod
     def by_name(cls, channel_name: str, command_id: Optional[str] = None):
         return SignalChannelCommand(
-            command_id if command_id is not None else "", channel_name
+            command_id if command_id is not None else "",
+            channel_name,
         )
 
 
@@ -57,20 +58,31 @@ def by_name(cls, channel_name: str, command_id: Optional[str] = None):
 class CommandRequest:
     commands: list[BaseCommand]
     command_waiting_type: CommandWaitingType
+    command_combinations: list[CommandCombination]
 
     @classmethod
     def for_any_command_completed(cls, *commands: BaseCommand):
         bc = [c for c in commands]
-        return CommandRequest(bc, CommandWaitingType.ANY_COMPLETED)
+        return CommandRequest(bc, CommandWaitingType.ANY_COMPLETED, [])
 
     @classmethod
     def for_all_command_completed(cls, *commands: BaseCommand):
         bc = [c for c in commands]
-        return CommandRequest(bc, CommandWaitingType.ALL_COMPLETED)
+        return CommandRequest(bc, CommandWaitingType.ALL_COMPLETED, [])
+
+    @classmethod
+    def for_any_command_combination_completed(
+        cls, command_combinations_list: list[list[str]], *commands: BaseCommand
+    ):
+        return CommandRequest(
+            list(commands),
+            CommandWaitingType.ANY_COMBINATION_COMPLETED,
+            [CommandCombination(c) for c in command_combinations_list],
+        )
 
     @classmethod
     def empty(cls):
-        return CommandRequest(list(), CommandWaitingType.ALL_COMPLETED)
+        return CommandRequest(list(), CommandWaitingType.ALL_COMPLETED, [])
 
 
 def _to_idl_command_request(request: CommandRequest) -> IdlCommandRequest:
@@ -99,4 +111,6 @@ def _to_idl_command_request(request: CommandRequest) -> IdlCommandRequest:
         req.inter_state_channel_commands = internal_channel_commands
     if len(signal_commands) > 0:
         req.signal_commands = signal_commands
+    if len(request.command_combinations) > 0:
+        req.command_combinations = request.command_combinations
     return req

iwf/tests/test_signal.py

@@ -3,11 +3,17 @@
 import unittest
 
 from iwf.client import Client
+from iwf.iwf_api.models.timer_result import TimerStatus
 from iwf.command_request import (
     CommandRequest,
     SignalChannelCommand,
+    TimerCommand,
+)
+from iwf.command_results import (
+    CommandResults,
+    SignalChannelCommandResult,
+    TimerCommandResult,
 )
-from iwf.command_results import CommandResults, SignalChannelCommandResult
 from iwf.communication import Communication
 from iwf.communication_schema import CommunicationMethod, CommunicationSchema
 from iwf.iwf_api.models import ChannelRequestStatus
@@ -25,8 +31,14 @@
 test_channel_str = "test-str"
 test_idle_channel_none = "test-idle"
 
+test_channel1 = "test-channel1"
+test_channel1_id = "test-channel1-id"
+test_channel2 = "test-channel2"
+test_channel2_id = "test-channel2-id"
+test_timer_id = "test-timer-id"
+
 
-class WaitState(WorkflowState[None]):
+class WaitState1(WorkflowState[None]):
     def wait_until(
         self,
         ctx: WorkflowContext,
@@ -61,7 +73,58 @@ def execute(
         assert sig3 == SignalChannelCommandResult(
             test_channel_str, "abc", ChannelRequestStatus.RECEIVED, ""
         )
-        return StateDecision.graceful_complete_workflow(sig3.value)
+        return StateDecision.single_next_state(WaitState2, sig3.value)
+
+
+class WaitState2(WorkflowState[str]):
+    def wait_until(
+        self,
+        ctx: WorkflowContext,
+        input: str,
+        persistence: Persistence,
+        communication: Communication,
+    ) -> CommandRequest:
+        return CommandRequest.for_any_command_combination_completed(
+            [
+                [
+                    test_channel1_id,
+                    test_timer_id,
+                ]
+            ],
+            SignalChannelCommand.by_name(test_channel1, test_channel1_id),
+            SignalChannelCommand.by_name(test_channel2, test_channel2_id),
+            TimerCommand.by_seconds(1, test_timer_id),
+        )
+
+    def execute(
+        self,
+        ctx: WorkflowContext,
+        input: str,
+        command_results: CommandResults,
+        persistence: Persistence,
+        communication: Communication,
+    ) -> StateDecision:
+        assert len(command_results.signal_channel_commands) == 2
+        assert (
+            len(
+                [
+                    r
+                    for r in command_results.signal_channel_commands
+                    if r.status == ChannelRequestStatus.RECEIVED
+                ]
+            )
+            == 1
+        )
+        sig1 = command_results.signal_channel_commands[0]
+        tim1 = command_results.timer_commands[0]
+        assert sig1 == SignalChannelCommandResult(
+            test_channel1, None, ChannelRequestStatus.RECEIVED, test_channel1_id
+        )
+        assert tim1 == TimerCommandResult(
+            TimerStatus.FIRED,
+            test_timer_id,
+        )
+        return StateDecision.graceful_complete_workflow(input)
 
 
 class WaitSignalWorkflow(ObjectWorkflow):
@@ -71,10 +134,12 @@ def get_communication_schema(self) -> CommunicationSchema:
             CommunicationMethod.signal_channel_def(test_channel_none, type(None)),
             CommunicationMethod.signal_channel_def(test_channel_str, str),
             CommunicationMethod.signal_channel_def(test_idle_channel_none, type(None)),
+            CommunicationMethod.signal_channel_def(test_channel1, type(None)),
+            CommunicationMethod.signal_channel_def(test_channel2, type(None)),
         )
 
     def get_workflow_states(self) -> StateSchema:
-        return StateSchema.with_starting_state(WaitState())
+        return StateSchema.with_starting_state(WaitState1(), WaitState2())
 
     @rpc()
     def get_idle_signal_channel_size(self, com: Communication):
@@ -90,11 +155,13 @@ def setUpClass(cls):
 
     def test_signal(self):
         wf_id = f"{inspect.currentframe().f_code.co_name}-{time.time_ns()}"
-        self.client.start_workflow(WaitSignalWorkflow, wf_id, 1)
+        self.client.start_workflow(WaitSignalWorkflow, wf_id, 10)
         self.client.signal_workflow(wf_id, test_channel_int, 123)
         self.client.signal_workflow(wf_id, test_channel_str, "abc")
         self.client.signal_workflow(wf_id, test_channel_none)
-        res = self.client.get_simple_workflow_result_with_wait(wf_id)
+
+        self.client.signal_workflow(wf_id, test_channel1)
+        res = self.client.wait_for_workflow_completion(wf_id, str)
         assert res == "abc"
 
     def test_signal_channel_size(self):