test: Add more unit tests for conditionals

tests/data/workflows/unit_conditional_adder_diamond_skip_with_join_strong_dep_fails.yml

@@ -0,0 +1,64 @@
+title: Adder diamond skip with join with strong depedency
+#    A
+#    /\
+#   B  [C] <- C should not run
+#    \ /
+#     D <- D should fail, because it tries to access C's result
+description: |
+  Initial conditions:
+  - C fails
+  - D has a strong dependency on C, meaning it's both structurally dependent and data dependent
+
+  Expect:
+  - Condition returns false
+  - C does not run
+  - D fails
+config:
+  scheduler: dynamic
+entrypoint: a
+inputs:
+  another_url: http://api:8000
+  value: 1
+
+triggers:
+  - type: webhook
+    ref: my_webhook
+    id: wh-XXXXXX
+    entrypoint: a # This can be any
+    args:
+      url: http://api:8000/test/items/1
+      method: GET
+
+actions:
+  - ref: a
+    action: example.passthrough
+    args:
+      value: "1"
+
+  - ref: b
+    action: example.add
+    args:
+      lhs: ${{ ACTIONS.a.result -> int }}
+      rhs: 1
+    depends_on:
+      - a
+
+  - ref: c
+    action: example.add
+    args:
+      lhs: 3
+      rhs: ${{ ACTIONS.a.result -> int }}
+    depends_on:
+      - a
+    # This expression returns false and the task should not run, as a.result is "1"
+    run_if: ${{ FNS.is_equal(ACTIONS.a.result, str(2)) }}
+
+  # This task should still run, as B runs
+  - ref: d
+    action: example.add
+    args:
+      lhs: ${{ ACTIONS.c.result -> int }}
+      rhs: 4
+    depends_on:
+      - b
+      - c

tests/data/workflows/unit_conditional_adder_diamond_skip_with_join_weak_dep.yml

@@ -0,0 +1,64 @@
+title: Adder diamond skip with join with weak depedency
+#    A
+#    /\
+#   B  [C] <- C should not run
+#    \ /
+#     D <- D should still run, because it is not entirely dependent on C
+description: |
+  Initial conditions:
+  - C fails
+  - D has a weak dependency on C, meaning it's only structurally dependent but not data dependent
+
+  Expect:
+  - Condition returns false
+  - C does not run
+  - D still runs
+config:
+  scheduler: dynamic
+entrypoint: a
+inputs:
+  another_url: http://api:8000
+  value: 1
+
+triggers:
+  - type: webhook
+    ref: my_webhook
+    id: wh-XXXXXX
+    entrypoint: a # This can be any
+    args:
+      url: http://api:8000/test/items/1
+      method: GET
+
+actions:
+  - ref: a
+    action: example.passthrough
+    args:
+      value: "1"
+
+  - ref: b
+    action: example.add
+    args:
+      lhs: ${{ ACTIONS.a.result -> int }}
+      rhs: 1
+    depends_on:
+      - a
+
+  - ref: c
+    action: example.add
+    args:
+      lhs: 3
+      rhs: ${{ ACTIONS.a.result -> int }}
+    depends_on:
+      - a
+    # This expression returns false and the task should not run, as a.result is "1"
+    run_if: ${{ FNS.is_equal(ACTIONS.a.result, str(2)) }}
+
+  # This task should still run, as B runs
+  - ref: d
+    action: example.add
+    args:
+      lhs: ${{ ACTIONS.b.result -> int }}
+      rhs: 4
+    depends_on:
+      - b
+      - c

tests/data/workflows/unit_conditional_adder_diamond_skip_with_join_weak_dep_expected.yml

@@ -0,0 +1,13 @@
+ACTIONS:
+  a:
+    result: "1"
+    result_typename: "str"
+  b:
+    result: 2
+    result_typename: "int"
+  d:
+    result: 6
+    result_typename: "int"
+INPUTS:
+  another_url: "http://api:8000"
+  value: 1

tests/data/workflows/unit_conditional_adder_tree_skip_propagates.yml

@@ -0,0 +1,56 @@
+title: Adder tree skips with propagation
+#    A
+#    /\
+#   B  [C] <- C should not run
+#       |
+#      [D] <- D should not run as its entirely dependent on C
+description: Test that the condition returns false and C, D does not run
+config:
+  scheduler: dynamic
+entrypoint: a
+inputs:
+  another_url: http://api:8000
+  value: 1
+
+triggers:
+  - type: webhook
+    ref: my_webhook
+    id: wh-XXXXXX
+    entrypoint: a # This can be any
+    args:
+      url: http://api:8000/test/items/1
+      method: GET
+
+actions:
+  - ref: a
+    action: example.passthrough
+    args:
+      value: "1"
+
+  - ref: b
+    action: example.add
+    args:
+      # Demonstrate casting
+      lhs: ${{ ACTIONS.a.result -> int }}
+      rhs: 1
+    depends_on:
+      - a
+
+  - ref: c
+    action: example.add
+    args:
+      lhs: 3
+      rhs: ${{ ACTIONS.a.result -> int }}
+    depends_on:
+      - a
+    # This expression returns false and the task should not run, as a.result is "1"
+    run_if: ${{ FNS.is_equal(ACTIONS.a.result, str(2)) }}
+
+  # This task should not run, as C does not run
+  - ref: d
+    action: example.add
+    args:
+      lhs: 4
+      rhs: ${{ ACTIONS.c.result -> int }}
+    depends_on:
+      - c

tests/data/workflows/unit_conditional_adder_tree_halt.yml → tests/data/workflows/unit_conditional_adder_tree_skips.yml

@@ -1,4 +1,4 @@
-title: Adder tree 1 Workflow
+title: Adder tree skips
 #    A
 #    /\
 #   B  [C] <- C should not run

tests/data/workflows/unit_conditional_adder_tree_skips_expected.yml

@@ -0,0 +1,10 @@
+ACTIONS:
+  a:
+    result: "1"
+    result_typename: "str"
+  b:
+    result: 2
+    result_typename: "int"
+INPUTS:
+  another_url: "http://api:8000"
+  value: 1

tests/unit/test_workflows.py

@@ -30,6 +30,7 @@
     DSLWorkflow,
     dsl_activities,
 )
+from tracecat.types.exceptions import TracecatExpressionError
 
 DATA_PATH = Path(__file__).parent.parent.joinpath("data/workflows")
 SHARED_TEST_DEFNS = list(DATA_PATH.glob("shared_*.yml"))
@@ -201,22 +202,27 @@ async def test_workflow_ordering_is_correct(
     "dsl,expected",
     [
         (
-            DATA_PATH / "unit_conditional_adder_tree_halt.yml",
-            DATA_PATH / "unit_conditional_adder_tree_halt_expected.yaml",
+            DATA_PATH / "unit_conditional_adder_tree_skips.yml",
+            DATA_PATH / "unit_conditional_adder_tree_skips_expected.yml",
         ),
         (
             DATA_PATH / "unit_conditional_adder_tree_continues.yml",
-            DATA_PATH / "unit_conditional_adder_tree_continues_expected.yaml",
+            DATA_PATH / "unit_conditional_adder_tree_continues_expected.yml",
+        ),
+        (
+            DATA_PATH / "unit_conditional_adder_tree_skip_propagates.yml",
+            DATA_PATH / "unit_conditional_adder_tree_skip_propagates_expected.yml",
+        ),
+        (
+            DATA_PATH / "unit_conditional_adder_diamond_skip_with_join_weak_dep.yml",
+            DATA_PATH
+            / "unit_conditional_adder_diamond_skip_with_join_weak_dep_expected.yml",
         ),
-        # (
-        #     DATA_PATH / "unit_conditional_adder_tree_halt_with_propagation.yml",
-        #     DATA_PATH / "unit_conditional_adder_tree_halt_with_propagation.yaml",
-        # ),
     ],
     indirect=True,
 )
 @pytest.mark.asyncio
-async def test_conditional_execution(
+async def test_conditional_execution_completes(
     dsl, expected, temporal_cluster, mock_registry, auth_sandbox
 ):
     """We need to test that the ordering of the workflow tasks is correct."""
@@ -240,3 +246,41 @@ async def test_conditional_execution(
             retry_policy=RetryPolicy(maximum_attempts=1),
         )
     assert result == expected
+
+
+@pytest.mark.parametrize(
+    "dsl",
+    [DATA_PATH / "unit_conditional_adder_diamond_skip_with_join_strong_dep_fails.yml"],
+    indirect=True,
+)
+@pytest.mark.asyncio
+@pytest.mark.skip
+async def test_conditional_execution_fails(
+    dsl, temporal_cluster, mock_registry, auth_sandbox
+):
+    client = await get_temporal_client()
+    async with Worker(
+        client,
+        task_queue=os.environ["TEMPORAL__CLUSTER_QUEUE"],
+        activities=dsl_activities,
+        workflows=[DSLWorkflow],
+        workflow_runner=new_sandbox_runner(),
+    ):
+        # NOTE: I can't seem to figure out how to catch the exception thrown by the workflow
+        # We need to figure out how to bubble up certain exceptions to the client
+        # Or allow certain exceptions to control workflow execution
+        with pytest.raises(TracecatExpressionError) as e:
+            await client.execute_workflow(
+                DSLWorkflow.run,
+                DSLRunArgs(dsl=dsl, role=ctx_role.get()),
+                id=gen_id(f"test_conditional_execution-{dsl.title}"),
+                task_queue=os.environ["TEMPORAL__CLUSTER_QUEUE"],
+                retry_policy=RetryPolicy(
+                    maximum_attempts=0,
+                    non_retryable_error_types=[
+                        "tracecat.types.exceptions.TracecatExpressionError"
+                        "TracecatValidationError"
+                    ],
+                ),
+            )
+        assert "Operand has no path" in str(e)