add integration tests

Signed-off-by: Troy Chiu <[email protected]>

tests/flytekit/integration/remote/test_remote.py

@@ -616,6 +616,20 @@ def test_execute_workflow_with_maptask(register):
     assert execution.outputs["o0"] == [4, 5, 6]
     assert len(execution.node_executions["n0"].task_executions) == 1
 
+def test_execution_workflow_with_maptask_in_dynamic(register):
+    remote = FlyteRemote(Config.auto(config_file=CONFIG), PROJECT, DOMAIN)
+    d: typing.List[int] = [1, 2, 3]
+    flyte_launch_plan = remote.fetch_launch_plan(name="basic.dynamic_array_map.workflow_with_maptask_in_dynamic", version=VERSION)
+    execution = remote.execute(
+        flyte_launch_plan,
+        inputs={"data": d},
+        version=VERSION,
+        wait=True,
+    )
+    assert execution.outputs["o0"] == [2, 3, 4]
+    assert len(execution.node_executions["n0"].subworkflow_node_executions["n0-0-dn0"].task_executions) == 1
+
+
 def test_executes_nested_workflow_dictating_interruptible(register):
     remote = FlyteRemote(Config.auto(config_file=CONFIG), PROJECT, DOMAIN)
     flyte_launch_plan = remote.fetch_launch_plan(name="basic.child_workflow.parent_wf", version=VERSION)

tests/flytekit/integration/remote/workflows/basic/dynamic_array_map.py

@@ -0,0 +1,16 @@
+from flytekit import workflow, task, map_task, dynamic
+
+
+@task
+def fn(x: int) -> int:
+    return x + 1
+
+
+@dynamic
+def dynamic(data: list[int]) -> list[int]:
+    return map_task(fn)(x=data)
+
+
+@workflow
+def workflow_with_maptask_in_dynamic(data: list[int]) -> list[int]:
+    return dynamic(data=data)