feat: added more fixtures

Author: raphael-goetz

crates/taurus-core/src/fixtures.rs

@@ -31,7 +31,20 @@ pub struct Case {
 pub struct RemoteFixture {
     pub target_service: String,
     pub function_identifier: String,
-    pub result_parameter: String,
+    /// Only meaningful when `sub_flow_calls` is empty: the literal-valued
+    /// parameter to echo straight back as this remote call's own result
+    /// (see `0012_remote_function_subflow.json`).
+    #[serde(default)]
+    pub result_parameter: Option<String>,
+    /// Positional values to drive one `ExecutionEngine::execute_sub_flow`
+    /// call per entry against this request's `SubFlow`-valued parameter, in
+    /// order -- simulates an action invoking a minted sub-flow reference
+    /// the same number of times a real action would (e.g. once per element
+    /// for a remotely-dispatched `for_each`'s consumer callback). When
+    /// non-empty, the remote call itself resolves to `null` once every call
+    /// has run, mirroring a `void`-signature remote function.
+    #[serde(default)]
+    pub sub_flow_calls: Vec<serde_json::Value>,
 }
 
 #[derive(Clone, Deserialize)]

crates/taurus-core/src/runtime/engine/sub_flow_registry.rs

@@ -153,13 +153,18 @@ mod tests {
         let flow = flow_with_node(7);
         let activity = Arc::new(Notify::new());
 
+        // caller_node_id and caller_parameter_index are deliberately distinct
+        // (3, 5) so a field mix-up in `mint`/`PendingSubFlow` would be
+        // caught by the assertions below instead of accidentally matching.
         let id = registry
-            .mint(&flow, 7, "parent-1", activity, 1, 1)
+            .mint(&flow, 7, "parent-1", activity, 3, 5)
             .expect("node 7 exists in flow");
 
         let pending = registry.get(&id).expect("entry should exist after mint");
         assert_eq!(pending.start_idx, 0);
         assert_eq!(pending.parent_execution_id, "parent-1");
+        assert_eq!(pending.caller_node_id, 3);
+        assert_eq!(pending.caller_parameter_index, 5);
 
         // Looking up again does not remove the entry.
         assert!(registry.get(&id).is_some());

crates/taurus-tests/src/main.rs

@@ -8,21 +8,29 @@ use taurus_core::fixtures::{Case, Cases, Input, RemoteFixture, print_failure, pr
 use taurus_core::runtime::engine::ExecutionEngine;
 use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime};
 use taurus_core::types::errors::runtime_error::RuntimeError;
-use tucana::aquila::action_node_value;
+use taurus_core::types::signal::Signal;
+use tucana::aquila::{ActionNodeSubFlowValue, action_node_value};
 use tucana::shared::node_execution_result::{
     Id as NodeExecutionResultId, Result as NodeExecutionOutcome,
 };
+use tucana::shared::value::Kind;
 use tucana::shared::{
-    NodeExecutionResult,
+    NodeExecutionResult, Value,
     helper::value::{from_json_value, to_json_value},
 };
 
-struct FixtureRemoteRuntime {
+struct FixtureRemoteRuntime<'a> {
     fixture: RemoteFixture,
+    // Needed to simulate an action invoking a minted `SubFlow` reference
+    // back into `ExecutionEngine::execute_sub_flow` (see `sub_flow_calls`)
+    // -- the same engine instance the flow itself is running on, so the
+    // callback resolves against the registry entry the flow's own remote
+    // call minted.
+    engine: &'a ExecutionEngine,
 }
 
 #[async_trait::async_trait]
-impl RemoteRuntime for FixtureRemoteRuntime {
+impl RemoteRuntime for FixtureRemoteRuntime<'_> {
     async fn execute_remote(
         &self,
         execution: RemoteExecution,
@@ -48,6 +56,29 @@ impl RemoteRuntime for FixtureRemoteRuntime {
             ));
         }
 
+        // A `SubFlow`-valued parameter means this request carries a minted
+        // callback reference (see `resolve_remote_args`/`SubFlowRegistry`)
+        // -- drive it via `sub_flow_calls` regardless of how many entries
+        // that list has (zero is a legitimate, real case: an empty input
+        // list means the reference is minted but never actually invoked).
+        // Only fall through to the literal-echo path below when no
+        // `SubFlow` parameter is present at all.
+        if let Some(execution_identifier) = execution
+            .request
+            .parameters
+            .iter()
+            .find_map(|parameter| match parameter.value.as_ref()? {
+                action_node_value::Value::SubFlow(ActionNodeSubFlowValue {
+                    execution_identifier,
+                }) => Some(execution_identifier.clone()),
+                _ => None,
+            })
+        {
+            return self
+                .drive_sub_flow_calls(&execution, &execution_identifier)
+                .await;
+        }
+
         // Parameters are positional on the wire now (no key), so fixtures
         // with a `resultParameter` only make sense with a single remote
         // parameter — take it directly rather than looking it up by name.
@@ -65,7 +96,7 @@ impl RemoteRuntime for FixtureRemoteRuntime {
                     "T-TEST-000003",
                     "RemoteParameterMissing",
                     format!(
-                        "Remote parameter {} was not provided",
+                        "Remote parameter {:?} was not provided",
                         self.fixture.result_parameter
                     ),
                 )
@@ -83,6 +114,58 @@ impl RemoteRuntime for FixtureRemoteRuntime {
     }
 }
 
+impl FixtureRemoteRuntime<'_> {
+    /// Simulates an action driving `ActionSubFlowExecutionRequest` traffic
+    /// against `execution_identifier` (the request's minted `SubFlow`
+    /// reference): one `execute_sub_flow` call per `sub_flow_calls` entry,
+    /// in order -- exactly as `hercules`'s `for_each` implementation calls
+    /// back once per list element via `executeSubFlow`. An empty
+    /// `sub_flow_calls` is valid and drives zero calls (e.g. an empty input
+    /// list: the reference is minted but never actually invoked). Fails the
+    /// whole remote call the moment any individual sub-flow run fails,
+    /// since a real action would abort the same way. Resolves to `null`,
+    /// matching a `void`-signature consumer-driving remote function (e.g.
+    /// `for_each` itself).
+    async fn drive_sub_flow_calls(
+        &self,
+        execution: &RemoteExecution,
+        execution_identifier: &str,
+    ) -> Result<NodeExecutionResult, RuntimeError> {
+        for call in &self.fixture.sub_flow_calls {
+            let value = from_json_value(call.clone());
+            let report = self
+                .engine
+                .execute_sub_flow(&execution_identifier, vec![value], Some(self), false)
+                .await
+                .ok_or_else(|| {
+                    RuntimeError::new(
+                        "T-TEST-000005",
+                        "SubFlowNotFound",
+                        format!(
+                            "Sub flow {} was not minted (or already resolved)",
+                            execution_identifier
+                        ),
+                    )
+                })?;
+            if let Signal::Failure(err) = report.signal {
+                return Err(err);
+            }
+        }
+
+        Ok(NodeExecutionResult {
+            started_at: 0,
+            finished_at: 0,
+            parameter_results: Vec::new(),
+            id: Some(NodeExecutionResultId::FunctionIdentifier(
+                execution.request.function_identifier.clone(),
+            )),
+            result: Some(NodeExecutionOutcome::Success(Value {
+                kind: Some(Kind::NullValue(0)),
+            })),
+        })
+    }
+}
+
 pub enum CaseResult {
     Success,
     Failure(Input, serde_json::Value),
@@ -107,7 +190,10 @@ impl Testable for Case {
         let remote = self
             .remote
             .clone()
-            .map(|fixture| FixtureRemoteRuntime { fixture });
+            .map(|fixture| FixtureRemoteRuntime {
+                fixture,
+                engine: &engine,
+            });
 
         for input in self.inputs.clone() {
             let flow_input = input.clone().input.map(from_json_value);

crates/taurus/src/app/worker.rs

@@ -827,11 +827,12 @@ mod tests {
     /// keyed here against the node/parameter that minted the sub-flow
     /// reference in the first place (node `caller_node_id`'s parameter at
     /// `caller_parameter_index`).
-    fn first_input_type_param(
+    fn input_type_param(
         database_id: i64,
         runtime_parameter_id: &str,
         caller_node_id: i64,
         caller_parameter_index: i64,
+        input_index: i64,
     ) -> NodeParameter {
         NodeParameter {
             database_id,
@@ -843,7 +844,7 @@ mod tests {
                             tucana::shared::InputType {
                                 node_id: caller_node_id,
                                 parameter_index: caller_parameter_index,
-                                input_index: 0,
+                                input_index,
                             },
                         )),
                         paths: vec![],
@@ -854,6 +855,21 @@ mod tests {
         }
     }
 
+    fn first_input_type_param(
+        database_id: i64,
+        runtime_parameter_id: &str,
+        caller_node_id: i64,
+        caller_parameter_index: i64,
+    ) -> NodeParameter {
+        input_type_param(
+            database_id,
+            runtime_parameter_id,
+            caller_node_id,
+            caller_parameter_index,
+            0,
+        )
+    }
+
     /// Captures the minted sub-flow UUID from the outgoing remote request's
     /// `SubFlow` parameter, then blocks until `release` is notified --
     /// standing in for the parent remote call staying outstanding while the
@@ -1006,6 +1022,83 @@ mod tests {
         }
     }
 
+    /// A sub-flow's action-supplied parameters are seeded one `InputType`
+    /// slot per positional value (`input_index` 0, 1, ...), not just the
+    /// first. Drives a two-argument call and reads both slots back through
+    /// `std::number::add`, so a regression that only seeds `input_index: 0`
+    /// (as a naive port of the single-argument `for_each` case might) would
+    /// fail this by treating the second argument as missing.
+    #[tokio::test]
+    async fn sub_flow_execution_seeds_every_positional_input_index_independently() {
+        let engine = Arc::new(ExecutionEngine::new());
+        let minted_id = Arc::new(StdMutex::new(None));
+        let release = Arc::new(Notify::new());
+        let remote = BlockingMintCapturingRuntime {
+            result: NodeExecutionResult {
+                started_at: 1,
+                finished_at: 2,
+                parameter_results: Vec::new(),
+                id: Some(tucana::shared::node_execution_result::Id::NodeId(1)),
+                result: Some(tucana::shared::node_execution_result::Result::Success(
+                    int_value(1),
+                )),
+            },
+            minted_id: Arc::clone(&minted_id),
+            release: Arc::clone(&release),
+        };
+
+        let remote_node = node(
+            1,
+            "remote::open_stream",
+            vec![sub_flow_param(100, "on_message", 2)],
+            None,
+            Some("action.svc"),
+        );
+        // Reads both action-supplied arguments back independently via their
+        // own `input_index`, so the assertion below proves neither slot
+        // leaked into or overwrote the other.
+        let sub_flow_target = node(
+            2,
+            "std::number::add",
+            vec![
+                input_type_param(200, "first", 1, 0, 0),
+                input_type_param(201, "second", 1, 0, 1),
+            ],
+            None,
+            None,
+        );
+
+        let flow = ExecutionFlow {
+            flow_id: 1,
+            project_id: 7,
+            starting_node_id: 1,
+            node_functions: vec![remote_node, sub_flow_target],
+            input_value: None,
+        };
+
+        let parent_engine = Arc::clone(&engine);
+        let parent = tokio::spawn(async move {
+            parent_engine
+                .execute_flow_report_async("parent-1", flow, Some(&remote), false)
+                .await
+        });
+
+        let execution_identifier = wait_for_minted_id(&minted_id).await;
+
+        let request = ActionSubFlowExecutionRequest {
+            execution_identifier: execution_identifier.clone(),
+            parameters: vec![int_value(3), int_value(4)],
+        };
+        let result = build_sub_flow_execution_result(request, &engine, None, false).await;
+        match result.result {
+            Some(execution_result::Result::Success(value)) => assert_eq!(value, int_value(7)),
+            other => panic!("expected success result, got {:?}", other),
+        }
+
+        release.notify_one();
+        parent.await.expect("parent task should not panic");
+    }
+
     #[tokio::test]
     async fn sub_flow_execution_result_reports_not_found_for_unknown_id() {
         let engine = ExecutionEngine::new();