236 lines
9.0 KiB
Rust
236 lines
9.0 KiB
Rust
//! human 节点端到端集成测试 — DagDef → NodeRegistry.build_dag → DagExecutor.run
|
|
//!
|
|
//! df-workflow 不依赖 df-nodes(反向依赖),此处用等价阻塞节点模拟 HumanNode 的
|
|
//! 「发审批请求 → 挂起等待 → 外部 approve → 返回结果」链路,覆盖
|
|
//! DagDef 序列化 → 注册表 → 执行器全链路 + 事件序列断言(R6 send 缺 await /
|
|
//! R7 契约失配的存活土壤)。
|
|
|
|
use std::time::Duration;
|
|
|
|
use async_trait::async_trait;
|
|
use serde_json::json;
|
|
|
|
use df_types::events::{SelectType, WorkflowEvent};
|
|
use df_types::types::NodeStatus;
|
|
use df_workflow::dag_def::DagDef;
|
|
use df_workflow::eventbus::{EventBus, EventSubscriber};
|
|
use df_workflow::executor::DagExecutor;
|
|
use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema};
|
|
use df_workflow::registry::NodeRegistry;
|
|
|
|
/// 前驱节点:sleep 后返回空输出(仿 executor 单测 SleepNode)。
|
|
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: json!(null),
|
|
output: json!(null),
|
|
}
|
|
}
|
|
|
|
fn node_type(&self) -> &str {
|
|
"sleep"
|
|
}
|
|
}
|
|
|
|
/// 模拟 HumanNode 的阻塞审批节点:先 subscribe 再发 HumanApprovalRequest(broadcast 不回放),
|
|
/// select! 等待匹配 execution_id + node_id 的 HumanApprovalResponse,收到则返回 decision。
|
|
struct ApprovalNode;
|
|
|
|
#[async_trait]
|
|
impl Node for ApprovalNode {
|
|
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
|
let mut rx = ctx.event_bus.subscribe();
|
|
let _ = ctx
|
|
.event_bus
|
|
.send(WorkflowEvent::HumanApprovalRequest {
|
|
execution_id: ctx.execution_id.clone(),
|
|
node_id: ctx.node_id.clone(),
|
|
title: "确认发布".to_string(),
|
|
description: String::new(),
|
|
options: vec!["同意".into(), "拒绝".into()],
|
|
select_type: SelectType::Single,
|
|
})
|
|
.await;
|
|
let timeout_secs = ctx
|
|
.config
|
|
.get("timeout_secs")
|
|
.and_then(|v| v.as_u64())
|
|
.unwrap_or(2);
|
|
let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
|
|
loop {
|
|
tokio::select! {
|
|
recv = rx.recv() => match recv {
|
|
Ok(WorkflowEvent::HumanApprovalResponse {
|
|
execution_id, node_id, decision, decisions, ..
|
|
}) if execution_id == ctx.execution_id && node_id == ctx.node_id => {
|
|
let primary = if decision.is_empty() {
|
|
decisions.first().cloned().unwrap_or_default()
|
|
} else {
|
|
decision
|
|
};
|
|
return Ok(NodeOutput::from_value(json!({ "decision": primary })));
|
|
}
|
|
Ok(_) => continue,
|
|
Err(_) => return Err(anyhow::anyhow!("事件总线关闭,审批无法完成")),
|
|
},
|
|
_ = tokio::time::sleep_until(deadline) => {
|
|
return Err(anyhow::anyhow!("人工审批等待超时({timeout_secs}s)"));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn schema(&self) -> NodeSchema {
|
|
NodeSchema {
|
|
params: json!(null),
|
|
output: json!(null),
|
|
}
|
|
}
|
|
|
|
fn is_blocking(&self) -> bool {
|
|
true
|
|
}
|
|
|
|
fn node_type(&self) -> &str {
|
|
"human"
|
|
}
|
|
}
|
|
|
|
/// 从事件流中等待一条 HumanApprovalRequest(跳过 NodeStarted/NodeCompleted 等其他事件)。
|
|
async fn wait_approval_request(mut rx: EventSubscriber) -> WorkflowEvent {
|
|
tokio::time::timeout(Duration::from_millis(1000), async move {
|
|
loop {
|
|
if let Ok(ev @ WorkflowEvent::HumanApprovalRequest { .. }) = rx.recv().await {
|
|
return ev;
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
.expect("应收到 HumanApprovalRequest,实际超时")
|
|
}
|
|
|
|
/// 端到端主链路:前驱 sleep(a) → human(b)。
|
|
/// 执行到 human 后挂起等审批 → 外部 approve(模拟 approve_human_approval IPC) → 返回结果、工作流完成。
|
|
#[tokio::test]
|
|
async fn human_workflow_end_to_end_approval() {
|
|
let bus = EventBus::new();
|
|
let rx = bus.subscribe(); // 先 subscribe 再执行(broadcast 不回放)
|
|
let exec_id = "exec-e2e-approve";
|
|
|
|
// DagDef(可序列化定义)→ NodeRegistry.build_dag → 运行时 Dag
|
|
let mut def = DagDef::new();
|
|
def.add_node("a", "sleep", json!({}));
|
|
def.add_node("b", "human", json!({}));
|
|
def.add_edge("a", "b");
|
|
let mut registry = NodeRegistry::new();
|
|
registry.register("sleep", |_| Box::new(SleepNode { sleep_ms: 10 }));
|
|
registry.register("human", |_| Box::new(ApprovalNode));
|
|
let dag = registry.build_dag(&def).expect("build_dag 应成功");
|
|
|
|
let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
|
|
let sm = executor.state_machine(); // 共享状态机(spawn 后仍可读)
|
|
let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });
|
|
|
|
// a 层完成、b 层 human subscribe + send Request 后应收到审批请求
|
|
let request = wait_approval_request(rx).await;
|
|
match request {
|
|
WorkflowEvent::HumanApprovalRequest {
|
|
node_id, title, options, ..
|
|
} => {
|
|
assert_eq!(node_id, "b");
|
|
assert_eq!(title, "确认发布");
|
|
assert_eq!(options, vec!["同意".to_string(), "拒绝".to_string()]);
|
|
}
|
|
other => panic!("期望 HumanApprovalRequest,收到 {:?}", other),
|
|
}
|
|
|
|
// 外部 approve(模拟 approve_human_approval IPC 发送 HumanApprovalResponse 到总线)
|
|
bus.send(WorkflowEvent::HumanApprovalResponse {
|
|
execution_id: exec_id.into(),
|
|
node_id: "b".to_string(),
|
|
decision: "同意".into(),
|
|
decisions: vec![],
|
|
comment: Some("可以发布".into()),
|
|
})
|
|
.await;
|
|
|
|
let outputs = run_handle
|
|
.await
|
|
.expect("run 不应 panic")
|
|
.expect("run 应成功");
|
|
assert!(outputs.contains_key("a"), "outputs 应含前驱节点 a");
|
|
assert_eq!(outputs["b"].data["decision"], json!("同意"));
|
|
assert_eq!(sm.get(&"a".to_string()), NodeStatus::Completed);
|
|
assert_eq!(sm.get(&"b".to_string()), NodeStatus::Completed);
|
|
}
|
|
|
|
/// 取消路径:外部 set_cancelled(模拟 cancel_workflow_node IPC 写共享状态机)
|
|
/// → human select! 轮询 is_cancelled → Err → 状态保持 Cancelled。
|
|
#[tokio::test]
|
|
async fn human_workflow_cancel_via_shared_state_machine() {
|
|
let bus = EventBus::new();
|
|
let rx = bus.subscribe();
|
|
let exec_id = "exec-e2e-cancel";
|
|
|
|
let mut def = DagDef::new();
|
|
def.add_node("h", "human", json!({}));
|
|
let mut registry = NodeRegistry::new();
|
|
registry.register("human", |_| Box::new(ApprovalNode));
|
|
let dag = registry.build_dag(&def).expect("build_dag 应成功");
|
|
|
|
let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
|
|
let sm = executor.state_machine();
|
|
let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });
|
|
|
|
// 等 human 发出审批请求(确认已 subscribe + send + 进入 select! 轮询)
|
|
let _ = wait_approval_request(rx).await;
|
|
|
|
// 外部 cancel:写共享状态机置 Cancelled,humar select! 的 cancel_tick 会读到
|
|
sm.set_cancelled("h".to_string());
|
|
|
|
let result = run_handle.await.unwrap();
|
|
assert!(result.is_err(), "取消应致 run 返回 Err");
|
|
let err = result.unwrap_err().to_string();
|
|
assert!(
|
|
err.contains("取消") || err.contains("执行失败"),
|
|
"应返回取消相关错误,实际: {err}"
|
|
);
|
|
// 状态保持 Cancelled(未被 set_failed 覆盖为 Failed)—— 取消终态不被失败语义覆盖
|
|
assert_eq!(sm.get(&"h".to_string()), NodeStatus::Cancelled);
|
|
}
|
|
|
|
/// 挂起等待路径:收到请求后不 approve → human 挂起直到超时 → run 返 Err。
|
|
/// 证明节点真在等待外部响应(而非空转直接返回)。
|
|
#[tokio::test]
|
|
async fn human_workflow_waits_then_times_out_without_approve() {
|
|
let bus = EventBus::new();
|
|
let rx = bus.subscribe();
|
|
let exec_id = "exec-e2e-timeout";
|
|
|
|
let mut def = DagDef::new();
|
|
def.add_node("h", "human", json!({ "timeout_secs": 1 }));
|
|
let mut registry = NodeRegistry::new();
|
|
registry.register("human", |_| Box::new(ApprovalNode));
|
|
let dag = registry.build_dag(&def).expect("build_dag 应成功");
|
|
|
|
let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
|
|
let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });
|
|
|
|
let _ = wait_approval_request(rx).await;
|
|
let result = run_handle.await.unwrap();
|
|
assert!(result.is_err(), "无审批响应应超时失败");
|
|
// executor 包了「节点 h 执行失败」context,用 {:#} 展开整条错误链断言根因
|
|
let chain = format!("{:#}", result.unwrap_err());
|
|
assert!(chain.contains("超时"), "应返回超时错误,实际: {chain}");
|
|
}
|