feat(rs): add utility functions for testing flow functions

Author: lemorage

src/ops/functions/mod.rs

@@ -2,3 +2,6 @@ pub mod embed_text;
 pub mod extract_by_llm;
 pub mod parse_json;
 pub mod split_recursively;
+
+#[cfg(test)]
+mod test_utils;

src/ops/functions/test_utils.rs

@@ -0,0 +1,61 @@
+use crate::builder::plan::AnalyzedValueMapping;
+use crate::ops::sdk::{
+    EnrichedValueType, FlowInstanceContext, OpArgSchema, OpArgsResolver, SimpleFunctionExecutor,
+    SimpleFunctionFactoryBase, Value,
+};
+use anyhow::Result;
+use serde::de::DeserializeOwned;
+use serde_json::Value as JsonValue;
+use std::sync::Arc;
+
+// This function provides a helper to create OpArgSchema for literal values.
+pub fn new_literal_op_arg_schema(
+    name: Option<&str>,
+    value: Value,
+    value_type: EnrichedValueType,
+) -> OpArgSchema {
+    OpArgSchema {
+        name: name.map_or(crate::base::spec::OpArgName(None), |n| {
+            crate::base::spec::OpArgName(Some(n.to_string()))
+        }),
+        value_type,
+        analyzed_value: AnalyzedValueMapping::Constant { value },
+    }
+}
+
+// This function tests a flow function by providing a spec, input argument schemas, and values.
+pub async fn test_flow_function<S, R, F>(
+    factory: Arc<F>,
+    spec_json: JsonValue,
+    input_arg_schemas: Vec<OpArgSchema>,
+    input_arg_values: Vec<Value>,
+    context: Arc<FlowInstanceContext>,
+) -> Result<Value>
+where
+    S: DeserializeOwned + Send + Sync + 'static,
+    R: Send + Sync + 'static,
+    F: SimpleFunctionFactoryBase<Spec = S, ResolvedArgs = R> + ?Sized,
+{
+    // 1. Deserialize Spec
+    let spec: S = serde_json::from_value(spec_json)?;
+
+    // 2. Resolve Schema & Args
+    // The caller of test_flow_function will be responsible for creating these schemas.
+    let mut args_resolver = OpArgsResolver::new(&input_arg_schemas)?;
+
+    let (resolved_args_from_schema, _output_schema): (R, EnrichedValueType) = factory
+        .resolve_schema(&spec, &mut args_resolver, &context)
+        .await?;
+
+    args_resolver.done()?;
+
+    // 3. Build Executor
+    let executor: Box<dyn SimpleFunctionExecutor> = factory
+        .build_executor(spec, resolved_args_from_schema, Arc::clone(&context))
+        .await?;
+
+    // 4. Evaluate
+    let result = executor.evaluate(input_arg_values).await?;
+
+    Ok(result)
+}