Skip to content

Commit 7a951aa

Browse files
Merge pull request #444 from conductor-oss/fix/dynamic-fork-tasks-param
fix: restore dynamic_fork_tasks_param in DynamicForkTask (re-fixes #349)
2 parents c44e7fa + 79141a7 commit 7a951aa

2 files changed

Lines changed: 33 additions & 1 deletion

File tree

src/conductor/client/workflow/task/dynamic_fork_task.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ def __init__(self, task_ref_name: str, tasks_param: str = "dynamicTasks", tasks_
2020

2121
def to_workflow_task(self) -> WorkflowTask:
2222
wf_task = super().to_workflow_task()
23-
wf_task.dynamic_fork_join_tasks_param = self.tasks_param
23+
wf_task.dynamic_fork_tasks_param = self.tasks_param
2424
wf_task.dynamic_fork_tasks_input_param_name = self.tasks_input_param_name
2525
tasks = [
2626
wf_task,
Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
import unittest
2+
3+
from conductor.client.workflow.task.dynamic_fork_task import DynamicForkTask
4+
from conductor.client.workflow.task.join_task import JoinTask
5+
6+
7+
class TestDynamicForkTask(unittest.TestCase):
8+
def test_to_workflow_task_uses_dynamic_fork_tasks_param(self):
9+
task = DynamicForkTask(
10+
task_ref_name="fork",
11+
tasks_param="myTasks",
12+
tasks_input_param_name="myTasksInput",
13+
)
14+
tasks = task.to_workflow_task()
15+
wf_task = tasks[0]
16+
self.assertEqual(wf_task.dynamic_fork_tasks_param, "myTasks")
17+
self.assertEqual(wf_task.dynamic_fork_tasks_input_param_name, "myTasksInput")
18+
self.assertIsNone(wf_task.dynamic_fork_join_tasks_param)
19+
20+
def test_to_workflow_task_with_join(self):
21+
join = JoinTask("join", join_on=[])
22+
task = DynamicForkTask(task_ref_name="fork", join_task=join)
23+
tasks = task.to_workflow_task()
24+
self.assertEqual(len(tasks), 2)
25+
self.assertEqual(tasks[0].dynamic_fork_tasks_param, "dynamicTasks")
26+
self.assertEqual(tasks[0].dynamic_fork_tasks_input_param_name, "dynamicTasksInputs")
27+
self.assertIsNone(tasks[0].dynamic_fork_join_tasks_param)
28+
self.assertEqual(tasks[1].task_reference_name, "join")
29+
30+
31+
if __name__ == "__main__":
32+
unittest.main()

0 commit comments

Comments
 (0)