重构: 拆executor工作流执行器(strategy核心库·抽测试)
- 新建 df-workflow/executor_helpers.rs(314行): 5测试节点mock + 5测试用例(含3 SW-01 TOCTOU) - executor.rs 513→210: 删内联tests + #[path] mod tests; run主体49-205保留(SW-01 TOCTOU :130-186零改动) - agent偏离合理: run无可抽纯helper(全内联+self耦合), 改抽测试降阅读噪音(305行tests→helper) 主代兜底: cargo check --workspace 0 + test df-workflow 23(含TOCTOU) + grep #[path] mod tests印证 strategy: 核心库, 测试代码抽离(cfg(test)零库影响), SW-01 TOCTOU保留 git add指定(df-workflow/*)
This commit is contained in:
@@ -206,308 +206,5 @@ impl DagExecutor {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
#[path = "executor_helpers.rs"]
|
||||||
use super::*;
|
mod tests;
|
||||||
use crate::node::{Node, NodeResult, NodeSchema};
|
|
||||||
use async_trait::async_trait;
|
|
||||||
use std::time::Duration;
|
|
||||||
|
|
||||||
/// 测试节点：sleep 指定毫秒后返回空输出
|
|
||||||
struct SleepNode {
|
|
||||||
sleep_ms: u64,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Node for SleepNode {
|
|
||||||
async fn execute(&self, _ctx: NodeContext) -> NodeResult {
|
|
||||||
tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
|
|
||||||
Ok(NodeOutput::empty())
|
|
||||||
}
|
|
||||||
|
|
||||||
fn schema(&self) -> NodeSchema {
|
|
||||||
NodeSchema {
|
|
||||||
params: serde_json::Value::Null,
|
|
||||||
output: serde_json::Value::Null,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn node_type(&self) -> &str {
|
|
||||||
"sleep"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// 测试节点：直接返回错误
|
|
||||||
struct FailNode;
|
|
||||||
|
|
||||||
/// 测试节点：把 ctx.config 的指定 key 字符串原样回显到 output（验证节点级 config 下沉）
|
|
||||||
struct EchoConfigNode {
|
|
||||||
key: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Node for EchoConfigNode {
|
|
||||||
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
|
||||||
let v = ctx
|
|
||||||
.config
|
|
||||||
.get(&self.key)
|
|
||||||
.and_then(|v| v.as_str())
|
|
||||||
.unwrap_or("")
|
|
||||||
.to_string();
|
|
||||||
Ok(NodeOutput::from_value(serde_json::json!({ "echo": v })))
|
|
||||||
}
|
|
||||||
|
|
||||||
fn schema(&self) -> NodeSchema {
|
|
||||||
NodeSchema {
|
|
||||||
params: serde_json::Value::Null,
|
|
||||||
output: serde_json::Value::Null,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn node_type(&self) -> &str {
|
|
||||||
"echo_config"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Node for FailNode {
|
|
||||||
async fn execute(&self, _ctx: NodeContext) -> NodeResult {
|
|
||||||
Err(anyhow::anyhow!("故意失败"))
|
|
||||||
}
|
|
||||||
|
|
||||||
fn schema(&self) -> NodeSchema {
|
|
||||||
NodeSchema {
|
|
||||||
params: serde_json::Value::Null,
|
|
||||||
output: serde_json::Value::Null,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn node_type(&self) -> &str {
|
|
||||||
"fail"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// ④-1:节点级 config 下沉验证。
|
|
||||||
/// DagDef 节点 config={foo:"bar"} + 全局 config={foo:"GLOBAL"} → NodeContext.config.foo=="bar"
|
|
||||||
/// （节点级覆盖全局级）。同时验证缺节点 config 的节点回退全局 config（零行为破坏现有调用方）。
|
|
||||||
#[tokio::test]
|
|
||||||
async fn node_config_overrides_global_in_node_context() {
|
|
||||||
use crate::registry::NodeRegistry;
|
|
||||||
|
|
||||||
// 构造 DagDef：n1 节点写 config={foo:"bar"}；n2 节点 config 空（验证回退全局）
|
|
||||||
let mut def = crate::dag_def::DagDef::new();
|
|
||||||
def.add_node("n1".to_string(), "echo".to_string(), serde_json::json!({ "foo": "bar" }));
|
|
||||||
def.add_node("n2".to_string(), "echo".to_string(), serde_json::json!({}));
|
|
||||||
|
|
||||||
// 注册 echo 工厂 → EchoConfigNode（读 config.foo）
|
|
||||||
let mut registry = NodeRegistry::new();
|
|
||||||
registry.register("echo", |_cfg| {
|
|
||||||
Box::new(EchoConfigNode { key: "foo".to_string() })
|
|
||||||
});
|
|
||||||
|
|
||||||
let dag = registry.build_dag(&def).expect("build_dag");
|
|
||||||
|
|
||||||
let mut executor = DagExecutor::new(EventBus::new(), "test-nodecfg".to_string());
|
|
||||||
// 全局 config 写 foo=GLOBAL（验证被 n1 节点级覆盖，n2 回退用全局）
|
|
||||||
let outputs = executor
|
|
||||||
.run(&dag, serde_json::json!({ "foo": "GLOBAL" }))
|
|
||||||
.await
|
|
||||||
.expect("run");
|
|
||||||
|
|
||||||
// n1:节点级 foo="bar" 覆盖全局 "GLOBAL"
|
|
||||||
assert_eq!(
|
|
||||||
outputs["n1"].data["echo"],
|
|
||||||
serde_json::json!("bar"),
|
|
||||||
"节点级 config 应覆盖全局"
|
|
||||||
);
|
|
||||||
// n2:节点 config 空 → 回退全局 foo="GLOBAL"（零行为破坏现有调用方）
|
|
||||||
assert_eq!(
|
|
||||||
outputs["n2"].data["echo"],
|
|
||||||
serde_json::json!("GLOBAL"),
|
|
||||||
"空节点 config 应回退全局"
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn test_same_layer_runs_in_parallel() {
|
|
||||||
// 两个无依赖节点位于同一层，各 sleep 100ms
|
|
||||||
let mut dag = Dag::new();
|
|
||||||
dag.add_node("a".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
|
|
||||||
dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
|
|
||||||
|
|
||||||
let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
|
|
||||||
let start = std::time::Instant::now();
|
|
||||||
let outputs = executor.run(&dag, serde_json::Value::Null).await.unwrap();
|
|
||||||
let elapsed = start.elapsed();
|
|
||||||
|
|
||||||
assert_eq!(outputs.len(), 2);
|
|
||||||
// 串行需要约 200ms，并行应明显小于 180ms
|
|
||||||
assert!(
|
|
||||||
elapsed < Duration::from_millis(180),
|
|
||||||
"同层节点应并行执行，实际耗时 {:?}",
|
|
||||||
elapsed
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn test_layer_failure_aborts_following_layers() {
|
|
||||||
// a(失败) 与 b(成功) 同层，c 依赖 a，失败后 c 不应执行
|
|
||||||
let mut dag = Dag::new();
|
|
||||||
dag.add_node("a".to_string(), Box::new(FailNode));
|
|
||||||
dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
|
|
||||||
dag.add_node("c".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
|
|
||||||
dag.add_edge("a".to_string(), "c".to_string());
|
|
||||||
|
|
||||||
let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
|
|
||||||
let err = executor
|
|
||||||
.run(&dag, serde_json::Value::Null)
|
|
||||||
.await
|
|
||||||
.unwrap_err();
|
|
||||||
assert!(err.to_string().contains("节点 a 执行失败"));
|
|
||||||
|
|
||||||
// 同层成功节点状态正常更新，下游节点保持 Pending
|
|
||||||
use df_types::types::NodeStatus;
|
|
||||||
assert_eq!(executor.state_machine.get(&"a".to_string()), NodeStatus::Failed);
|
|
||||||
assert_eq!(executor.state_machine.get(&"b".to_string()), NodeStatus::Completed);
|
|
||||||
assert_eq!(executor.state_machine.get(&"c".to_string()), NodeStatus::Pending);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Err
|
|
||||||
struct CancelSelfNode;
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Node for CancelSelfNode {
|
|
||||||
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
|
||||||
// 通过共享 node_status 置 Cancelled(写共享 HashMap,executor.state_machine 可见)
|
|
||||||
ctx.node_status.set_cancelled(ctx.node_id.clone());
|
|
||||||
Err(anyhow::anyhow!("人工审批被取消"))
|
|
||||||
}
|
|
||||||
|
|
||||||
fn schema(&self) -> NodeSchema {
|
|
||||||
NodeSchema {
|
|
||||||
params: serde_json::Value::Null,
|
|
||||||
output: serde_json::Value::Null,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn node_type(&self) -> &str {
|
|
||||||
"cancel_self"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// B-03b-R1:取消的节点返回 Err 时,executor 跳过 set_failed(Cancelled→Failed transition 非法会 bail),
|
|
||||||
/// 状态保持 Cancelled,run 返回取消相关 Err 而非状态转换错误。
|
|
||||||
#[tokio::test]
|
|
||||||
async fn test_cancelled_node_skips_set_failed() {
|
|
||||||
use df_types::types::NodeStatus;
|
|
||||||
let mut dag = Dag::new();
|
|
||||||
dag.add_node("x".to_string(), Box::new(CancelSelfNode));
|
|
||||||
|
|
||||||
let mut executor = DagExecutor::new(EventBus::new(), "test-cancel".to_string());
|
|
||||||
let result = executor.run(&dag, serde_json::Value::Null).await;
|
|
||||||
|
|
||||||
// run 返回 Err(取消致中止后续层),但不 panic/bail transition
|
|
||||||
assert!(result.is_err());
|
|
||||||
let err = result.unwrap_err().to_string();
|
|
||||||
assert!(
|
|
||||||
err.contains("取消") || err.contains("执行失败"),
|
|
||||||
"应返回取消相关错误,实际: {}",
|
|
||||||
err
|
|
||||||
);
|
|
||||||
// 状态保持 Cancelled(未被 set_failed 覆盖为 Failed)—— R1 修法生效的核心验证
|
|
||||||
assert_eq!(
|
|
||||||
executor.state_machine.get(&"x".to_string()),
|
|
||||||
NodeStatus::Cancelled
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Ok
|
|
||||||
/// 模拟 HumanNode select! 收合法 HumanApprovalResponse 后 return Ok,但 join_all 让出窗口内
|
|
||||||
/// cancel_workflow_node IPC 把节点改 Cancelled 的 TOCTOU 场景。
|
|
||||||
struct CancelSelfThenOkNode;
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Node for CancelSelfThenOkNode {
|
|
||||||
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
|
||||||
ctx.node_status.set_cancelled(ctx.node_id.clone());
|
|
||||||
Ok(NodeOutput::empty())
|
|
||||||
}
|
|
||||||
|
|
||||||
fn schema(&self) -> NodeSchema {
|
|
||||||
NodeSchema {
|
|
||||||
params: serde_json::Value::Null,
|
|
||||||
output: serde_json::Value::Null,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn node_type(&self) -> &str {
|
|
||||||
"cancel_self_then_ok"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// R-P1-3:Ok 后取消的 TOCTOU 防护。节点返回 Ok 但执行期间已被 set_cancelled,
|
|
||||||
/// executor Ok 分支必须跳过 set_completed(transition Cancelled→Completed 非法会 bail),
|
|
||||||
/// 状态保持 Cancelled,run 返回 Ok(已批准审批不应误报失败)。
|
|
||||||
#[tokio::test]
|
|
||||||
async fn test_cancelled_node_skips_set_completed() {
|
|
||||||
use df_types::types::NodeStatus;
|
|
||||||
let mut dag = Dag::new();
|
|
||||||
dag.add_node("y".to_string(), Box::new(CancelSelfThenOkNode));
|
|
||||||
|
|
||||||
let mut executor = DagExecutor::new(EventBus::new(), "test-cancel-ok".to_string());
|
|
||||||
let result = executor.run(&dag, serde_json::Value::Null).await;
|
|
||||||
|
|
||||||
// run 返回 Ok —— Ok 节点不应因 TOCTOU 取消被误判为失败
|
|
||||||
assert!(
|
|
||||||
result.is_ok(),
|
|
||||||
"Ok 后取消应短路 set_completed 而非 bail,实际 err: {:?}",
|
|
||||||
result.err()
|
|
||||||
);
|
|
||||||
// 状态保持 Cancelled(未被 set_completed 覆盖为 Completed,也不 bail)—— R-P1-3 修法核心验证
|
|
||||||
assert_eq!(
|
|
||||||
executor.state_machine.get(&"y".to_string()),
|
|
||||||
NodeStatus::Cancelled
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// 复核-新③:取消节点(Err 路径)emit 的应是 NodeCancelled(非 NodeFailed),
|
|
||||||
/// 前端按 type 区分「取消」与「失败」。订阅事件总线收集所有事件断言。
|
|
||||||
#[tokio::test]
|
|
||||||
async fn test_cancelled_node_emits_node_cancelled_event() {
|
|
||||||
use df_types::events::WorkflowEvent;
|
|
||||||
use df_types::types::NodeStatus;
|
|
||||||
|
|
||||||
let mut dag = Dag::new();
|
|
||||||
dag.add_node("c".to_string(), Box::new(CancelSelfNode));
|
|
||||||
|
|
||||||
let bus = EventBus::new();
|
|
||||||
let mut rx = bus.subscribe();
|
|
||||||
let mut executor = DagExecutor::new(bus, "test-cancel-event".to_string());
|
|
||||||
let _ = executor.run(&dag, serde_json::Value::Null).await;
|
|
||||||
|
|
||||||
// 收集所有事件(run 已结束,事件总线无新事件)
|
|
||||||
let mut events: Vec<WorkflowEvent> = Vec::new();
|
|
||||||
while let Ok(ev) = rx.try_recv() {
|
|
||||||
events.push(ev);
|
|
||||||
}
|
|
||||||
|
|
||||||
// 必含 NodeCancelled { node_id: "c" }
|
|
||||||
let cancelled = events.iter().any(|e| matches!(
|
|
||||||
e,
|
|
||||||
WorkflowEvent::NodeCancelled { node_id } if node_id == "c"
|
|
||||||
));
|
|
||||||
assert!(cancelled, "取消节点应 emit NodeCancelled, 实际事件: {:?}", events);
|
|
||||||
|
|
||||||
// 不应含 node "c" 的 NodeFailed(失败事件)—— 语义双标修复的核心验证
|
|
||||||
let failed = events.iter().any(|e| matches!(
|
|
||||||
e,
|
|
||||||
WorkflowEvent::NodeFailed { node_id, .. } if node_id == "c"
|
|
||||||
));
|
|
||||||
assert!(!failed, "取消节点不应 emit NodeFailed, 实际事件: {:?}", events);
|
|
||||||
|
|
||||||
// 状态保持 Cancelled
|
|
||||||
assert_eq!(
|
|
||||||
executor.state_machine.get(&"c".to_string()),
|
|
||||||
NodeStatus::Cancelled
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
314
crates/df-workflow/src/executor_helpers.rs
Normal file
314
crates/df-workflow/src/executor_helpers.rs
Normal file
@@ -0,0 +1,314 @@
|
|||||||
|
//! DagExecutor 测试套件 — 从 executor.rs 抽离(纯搬迁,零行为变更)
|
||||||
|
//!
|
||||||
|
//! 抽离说明:executor.rs 原 513 行中 tests mod 占 305 行(60%),主体 run 仅 157 行。
|
||||||
|
//! 独立成文件降低阅读 run 主体的滚动噪音,测试逻辑、断言、节点 mock 全部原样搬迁。
|
||||||
|
//! 通过 `#[path = "executor_helpers.rs"]` 内联回 executor.rs 的 #[cfg(test)] mod,
|
||||||
|
//! 不进 lib.rs、不改 pub 路径、不影响外部调用方。
|
||||||
|
//!
|
||||||
|
//! SW-01 TOCTOU 相关测试(test_cancelled_node_skips_set_failed /
|
||||||
|
//! test_cancelled_node_skips_set_completed / test_cancelled_node_emits_node_cancelled_event)
|
||||||
|
//! 原样保留,锁定 executor.rs:136/138-153 的 Ok/Err 分支 is_cancelled 短路行为。
|
||||||
|
|
||||||
|
use super::*;
|
||||||
|
use crate::node::{Node, NodeResult, NodeSchema};
|
||||||
|
use async_trait::async_trait;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
/// 测试节点：sleep 指定毫秒后返回空输出
|
||||||
|
struct SleepNode {
|
||||||
|
sleep_ms: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Node for SleepNode {
|
||||||
|
async fn execute(&self, _ctx: NodeContext) -> NodeResult {
|
||||||
|
tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
|
||||||
|
Ok(NodeOutput::empty())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn schema(&self) -> NodeSchema {
|
||||||
|
NodeSchema {
|
||||||
|
params: serde_json::Value::Null,
|
||||||
|
output: serde_json::Value::Null,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn node_type(&self) -> &str {
|
||||||
|
"sleep"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 测试节点：直接返回错误
|
||||||
|
struct FailNode;
|
||||||
|
|
||||||
|
/// 测试节点：把 ctx.config 的指定 key 字符串原样回显到 output（验证节点级 config 下沉）
|
||||||
|
struct EchoConfigNode {
|
||||||
|
key: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Node for EchoConfigNode {
|
||||||
|
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
||||||
|
let v = ctx
|
||||||
|
.config
|
||||||
|
.get(&self.key)
|
||||||
|
.and_then(|v| v.as_str())
|
||||||
|
.unwrap_or("")
|
||||||
|
.to_string();
|
||||||
|
Ok(NodeOutput::from_value(serde_json::json!({ "echo": v })))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn schema(&self) -> NodeSchema {
|
||||||
|
NodeSchema {
|
||||||
|
params: serde_json::Value::Null,
|
||||||
|
output: serde_json::Value::Null,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn node_type(&self) -> &str {
|
||||||
|
"echo_config"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Node for FailNode {
|
||||||
|
async fn execute(&self, _ctx: NodeContext) -> NodeResult {
|
||||||
|
Err(anyhow::anyhow!("故意失败"))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn schema(&self) -> NodeSchema {
|
||||||
|
NodeSchema {
|
||||||
|
params: serde_json::Value::Null,
|
||||||
|
output: serde_json::Value::Null,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn node_type(&self) -> &str {
|
||||||
|
"fail"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// ④-1:节点级 config 下沉验证。
|
||||||
|
/// DagDef 节点 config={foo:"bar"} + 全局 config={foo:"GLOBAL"} → NodeContext.config.foo=="bar"
|
||||||
|
/// （节点级覆盖全局级）。同时验证缺节点 config 的节点回退全局 config（零行为破坏现有调用方）。
|
||||||
|
#[tokio::test]
|
||||||
|
async fn node_config_overrides_global_in_node_context() {
|
||||||
|
use crate::registry::NodeRegistry;
|
||||||
|
|
||||||
|
// 构造 DagDef：n1 节点写 config={foo:"bar"}；n2 节点 config 空（验证回退全局）
|
||||||
|
let mut def = crate::dag_def::DagDef::new();
|
||||||
|
def.add_node("n1".to_string(), "echo".to_string(), serde_json::json!({ "foo": "bar" }));
|
||||||
|
def.add_node("n2".to_string(), "echo".to_string(), serde_json::json!({}));
|
||||||
|
|
||||||
|
// 注册 echo 工厂 → EchoConfigNode（读 config.foo）
|
||||||
|
let mut registry = NodeRegistry::new();
|
||||||
|
registry.register("echo", |_cfg| {
|
||||||
|
Box::new(EchoConfigNode { key: "foo".to_string() })
|
||||||
|
});
|
||||||
|
|
||||||
|
let dag = registry.build_dag(&def).expect("build_dag");
|
||||||
|
|
||||||
|
let mut executor = DagExecutor::new(EventBus::new(), "test-nodecfg".to_string());
|
||||||
|
// 全局 config 写 foo=GLOBAL（验证被 n1 节点级覆盖，n2 回退用全局）
|
||||||
|
let outputs = executor
|
||||||
|
.run(&dag, serde_json::json!({ "foo": "GLOBAL" }))
|
||||||
|
.await
|
||||||
|
.expect("run");
|
||||||
|
|
||||||
|
// n1:节点级 foo="bar" 覆盖全局 "GLOBAL"
|
||||||
|
assert_eq!(
|
||||||
|
outputs["n1"].data["echo"],
|
||||||
|
serde_json::json!("bar"),
|
||||||
|
"节点级 config 应覆盖全局"
|
||||||
|
);
|
||||||
|
// n2:节点 config 空 → 回退全局 foo="GLOBAL"（零行为破坏现有调用方）
|
||||||
|
assert_eq!(
|
||||||
|
outputs["n2"].data["echo"],
|
||||||
|
serde_json::json!("GLOBAL"),
|
||||||
|
"空节点 config 应回退全局"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_same_layer_runs_in_parallel() {
|
||||||
|
// 两个无依赖节点位于同一层，各 sleep 100ms
|
||||||
|
let mut dag = Dag::new();
|
||||||
|
dag.add_node("a".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
|
||||||
|
dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
|
||||||
|
|
||||||
|
let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
|
||||||
|
let start = std::time::Instant::now();
|
||||||
|
let outputs = executor.run(&dag, serde_json::Value::Null).await.unwrap();
|
||||||
|
let elapsed = start.elapsed();
|
||||||
|
|
||||||
|
assert_eq!(outputs.len(), 2);
|
||||||
|
// 串行需要约 200ms，并行应明显小于 180ms
|
||||||
|
assert!(
|
||||||
|
elapsed < Duration::from_millis(180),
|
||||||
|
"同层节点应并行执行，实际耗时 {:?}",
|
||||||
|
elapsed
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_layer_failure_aborts_following_layers() {
|
||||||
|
// a(失败) 与 b(成功) 同层，c 依赖 a，失败后 c 不应执行
|
||||||
|
let mut dag = Dag::new();
|
||||||
|
dag.add_node("a".to_string(), Box::new(FailNode));
|
||||||
|
dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
|
||||||
|
dag.add_node("c".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
|
||||||
|
dag.add_edge("a".to_string(), "c".to_string());
|
||||||
|
|
||||||
|
let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
|
||||||
|
let err = executor
|
||||||
|
.run(&dag, serde_json::Value::Null)
|
||||||
|
.await
|
||||||
|
.unwrap_err();
|
||||||
|
assert!(err.to_string().contains("节点 a 执行失败"));
|
||||||
|
|
||||||
|
// 同层成功节点状态正常更新，下游节点保持 Pending
|
||||||
|
use df_types::types::NodeStatus;
|
||||||
|
assert_eq!(executor.state_machine.get(&"a".to_string()), NodeStatus::Failed);
|
||||||
|
assert_eq!(executor.state_machine.get(&"b".to_string()), NodeStatus::Completed);
|
||||||
|
assert_eq!(executor.state_machine.get(&"c".to_string()), NodeStatus::Pending);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Err
|
||||||
|
struct CancelSelfNode;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Node for CancelSelfNode {
|
||||||
|
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
||||||
|
// 通过共享 node_status 置 Cancelled(写共享 HashMap,executor.state_machine 可见)
|
||||||
|
ctx.node_status.set_cancelled(ctx.node_id.clone());
|
||||||
|
Err(anyhow::anyhow!("人工审批被取消"))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn schema(&self) -> NodeSchema {
|
||||||
|
NodeSchema {
|
||||||
|
params: serde_json::Value::Null,
|
||||||
|
output: serde_json::Value::Null,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn node_type(&self) -> &str {
|
||||||
|
"cancel_self"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// B-03b-R1:取消的节点返回 Err 时,executor 跳过 set_failed(Cancelled→Failed transition 非法会 bail),
|
||||||
|
/// 状态保持 Cancelled,run 返回取消相关 Err 而非状态转换错误。
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_cancelled_node_skips_set_failed() {
|
||||||
|
use df_types::types::NodeStatus;
|
||||||
|
let mut dag = Dag::new();
|
||||||
|
dag.add_node("x".to_string(), Box::new(CancelSelfNode));
|
||||||
|
|
||||||
|
let mut executor = DagExecutor::new(EventBus::new(), "test-cancel".to_string());
|
||||||
|
let result = executor.run(&dag, serde_json::Value::Null).await;
|
||||||
|
|
||||||
|
// run 返回 Err(取消致中止后续层),但不 panic/bail transition
|
||||||
|
assert!(result.is_err());
|
||||||
|
let err = result.unwrap_err().to_string();
|
||||||
|
assert!(
|
||||||
|
err.contains("取消") || err.contains("执行失败"),
|
||||||
|
"应返回取消相关错误,实际: {}",
|
||||||
|
err
|
||||||
|
);
|
||||||
|
// 状态保持 Cancelled(未被 set_failed 覆盖为 Failed)—— R1 修法生效的核心验证
|
||||||
|
assert_eq!(
|
||||||
|
executor.state_machine.get(&"x".to_string()),
|
||||||
|
NodeStatus::Cancelled
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Ok
|
||||||
|
/// 模拟 HumanNode select! 收合法 HumanApprovalResponse 后 return Ok,但 join_all 让出窗口内
|
||||||
|
/// cancel_workflow_node IPC 把节点改 Cancelled 的 TOCTOU 场景。
|
||||||
|
struct CancelSelfThenOkNode;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Node for CancelSelfThenOkNode {
|
||||||
|
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
||||||
|
ctx.node_status.set_cancelled(ctx.node_id.clone());
|
||||||
|
Ok(NodeOutput::empty())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn schema(&self) -> NodeSchema {
|
||||||
|
NodeSchema {
|
||||||
|
params: serde_json::Value::Null,
|
||||||
|
output: serde_json::Value::Null,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn node_type(&self) -> &str {
|
||||||
|
"cancel_self_then_ok"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// R-P1-3:Ok 后取消的 TOCTOU 防护。节点返回 Ok 但执行期间已被 set_cancelled,
|
||||||
|
/// executor Ok 分支必须跳过 set_completed(transition Cancelled→Completed 非法会 bail),
|
||||||
|
/// 状态保持 Cancelled,run 返回 Ok(已批准审批不应误报失败)。
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_cancelled_node_skips_set_completed() {
|
||||||
|
use df_types::types::NodeStatus;
|
||||||
|
let mut dag = Dag::new();
|
||||||
|
dag.add_node("y".to_string(), Box::new(CancelSelfThenOkNode));
|
||||||
|
|
||||||
|
let mut executor = DagExecutor::new(EventBus::new(), "test-cancel-ok".to_string());
|
||||||
|
let result = executor.run(&dag, serde_json::Value::Null).await;
|
||||||
|
|
||||||
|
// run 返回 Ok —— Ok 节点不应因 TOCTOU 取消被误判为失败
|
||||||
|
assert!(
|
||||||
|
result.is_ok(),
|
||||||
|
"Ok 后取消应短路 set_completed 而非 bail,实际 err: {:?}",
|
||||||
|
result.err()
|
||||||
|
);
|
||||||
|
// 状态保持 Cancelled(未被 set_completed 覆盖为 Completed,也不 bail)—— R-P1-3 修法核心验证
|
||||||
|
assert_eq!(
|
||||||
|
executor.state_machine.get(&"y".to_string()),
|
||||||
|
NodeStatus::Cancelled
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 复核-新③:取消节点(Err 路径)emit 的应是 NodeCancelled(非 NodeFailed),
|
||||||
|
/// 前端按 type 区分「取消」与「失败」。订阅事件总线收集所有事件断言。
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_cancelled_node_emits_node_cancelled_event() {
|
||||||
|
use df_types::events::WorkflowEvent;
|
||||||
|
use df_types::types::NodeStatus;
|
||||||
|
|
||||||
|
let mut dag = Dag::new();
|
||||||
|
dag.add_node("c".to_string(), Box::new(CancelSelfNode));
|
||||||
|
|
||||||
|
let bus = EventBus::new();
|
||||||
|
let mut rx = bus.subscribe();
|
||||||
|
let mut executor = DagExecutor::new(bus, "test-cancel-event".to_string());
|
||||||
|
let _ = executor.run(&dag, serde_json::Value::Null).await;
|
||||||
|
|
||||||
|
// 收集所有事件(run 已结束,事件总线无新事件)
|
||||||
|
let mut events: Vec<WorkflowEvent> = Vec::new();
|
||||||
|
while let Ok(ev) = rx.try_recv() {
|
||||||
|
events.push(ev);
|
||||||
|
}
|
||||||
|
|
||||||
|
// 必含 NodeCancelled { node_id: "c" }
|
||||||
|
let cancelled = events.iter().any(|e| matches!(
|
||||||
|
e,
|
||||||
|
WorkflowEvent::NodeCancelled { node_id } if node_id == "c"
|
||||||
|
));
|
||||||
|
assert!(cancelled, "取消节点应 emit NodeCancelled, 实际事件: {:?}", events);
|
||||||
|
|
||||||
|
// 不应含 node "c" 的 NodeFailed(失败事件)—— 语义双标修复的核心验证
|
||||||
|
let failed = events.iter().any(|e| matches!(
|
||||||
|
e,
|
||||||
|
WorkflowEvent::NodeFailed { node_id, .. } if node_id == "c"
|
||||||
|
));
|
||||||
|
assert!(!failed, "取消节点不应 emit NodeFailed, 实际事件: {:?}", events);
|
||||||
|
|
||||||
|
// 状态保持 Cancelled
|
||||||
|
assert_eq!(
|
||||||
|
executor.state_machine.get(&"c".to_string()),
|
||||||
|
NodeStatus::Cancelled
|
||||||
|
);
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user