use std::sync::Arc;
use std::time::{Duration, Instant};
use ek_core::main_engine::MainEngine;
use ek_core::service::PlugNotice;
use ek_event::EVENT_TIMER;
use ek_event::Event;
use ek_station::{StationConfig, boot};
use ek_ui::state::{ConfirmStep, UiCmd, UiShared, is_quarantined};
fn wait_until(tag: &str, secs: u64, f: impl Fn() -> bool) {
let deadline = Instant::now() + Duration::from_secs(secs);
while Instant::now() < deadline {
if f() {
return;
}
std::thread::sleep(Duration::from_millis(20));
}
panic!("wait_until timeout: {tag}");
}
#[test]
fn u0_cmd_queue_roundtrip() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
s.push_cmd(UiCmd::Start {
factory: "dm".to_string(),
name: String::new(),
});
s.push_cmd(UiCmd::Remove {
name: "dicy".to_string(),
});
assert_eq!(
s.drain_cmds(),
vec![
UiCmd::Start {
factory: "dm".to_string(),
name: String::new()
},
UiCmd::Remove {
name: "dicy".to_string()
},
]
);
assert!(s.drain_cmds().is_empty(), "drain 是消费性取走");
}
#[test]
fn u1_confirm_steps_and_cancel() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
assert_eq!(s.confirm_step("a"), ConfirmStep::Armed);
assert_eq!(s.pending_confirm().as_deref(), Some("a"));
assert_eq!(s.confirm_step("a"), ConfirmStep::Fired);
assert_eq!(s.pending_confirm(), None);
assert_eq!(s.confirm_step("b"), ConfirmStep::Armed);
assert_eq!(s.confirm_step("c"), ConfirmStep::Switched);
assert_eq!(s.pending_confirm().as_deref(), Some("c"));
s.confirm_cancel();
assert_eq!(s.pending_confirm(), None);
assert_eq!(s.confirm_step("b"), ConfirmStep::Armed); }
#[test]
fn u2_quarantined_detection() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
assert!(is_quarantined("Quarantined"));
assert!(is_quarantined("State::Quarantined(panic x3)"));
assert!(!is_quarantined("Running"));
assert!(!is_quarantined("Ready"));
}
#[test]
fn u5_cfg_readouts_and_delete_arm() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
assert_eq!(s.cfg_briefs(), None);
s.request_cfg_refresh();
assert!(s.take_cfg_due());
assert!(!s.take_cfg_due(), "consume = take once is cleared");
s.set_cfg_briefs(vec![ek_cfg::wire::NsBrief {
ns: "oms".to_owned(),
rev: 12,
keys: vec![ek_cfg::wire::KeyBrief {
key: "rate_limit".to_owned(),
secret: false,
}],
}]);
let b = s.cfg_briefs().unwrap();
assert_eq!(b.len(), 1);
assert_eq!(b[0].rev, 12);
assert!(!b[0].keys[0].secret);
assert_eq!(s.cfg_delete_armed(), None);
s.arm_cfg_delete(Some("oms".to_owned()));
assert_eq!(s.cfg_delete_armed().as_deref(), Some("oms"));
s.arm_cfg_delete(None);
assert_eq!(s.cfg_delete_armed(), None);
}
#[test]
fn t6_cfg_list_real_chain() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
boot(
&engine,
&ek_ui::ui_kinds(&shared, true),
&StationConfig {
plugins: vec![
"cfg".to_string(),
"dm".to_string(),
"oms".to_string(),
"ui".to_string(),
],
},
)
.unwrap();
engine.start().unwrap();
wait_until("cfg row", 4, || {
shared.snapshot().rows.iter().any(|r| r.name == "cfg")
});
shared.push_cmd(UiCmd::CfgList);
put_timer(&engine);
wait_until("cfg briefs", 4, || shared.cfg_briefs().is_some());
engine.close_all(Duration::from_secs(2));
}
#[test]
fn u3_list_cadence_status_flattening_and_pipeline_flag() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
let due: Vec<bool> = (0..9).map(|_| s.list_poll_due()).collect();
assert_eq!(
due,
vec![true, false, false, true, false, false, false, true, false],
"管制面快照 = 第一拍必问(界面不该空着等 4 秒),之后每 4 拍(第 4、第 8 拍)"
);
s.request_list_refresh();
assert!(s.list_poll_due(), "被标记的那一拍立刻重问");
assert!(!s.list_poll_due(), "标记是消费性的,不是一档状态");
let present: Vec<String> = (0..8).map(|i| format!("gw{i}")).collect();
let mut asked: Vec<String> = Vec::new();
for _ in 0..8 {
let batch = s.status_batch(&present);
assert!(
batch.len() <= ek_ui::state::STATUS_ASKS_PER_TICK,
"单拍问的账户数不许超过 STATUS_ASKS_PER_TICK,实测 {}",
batch.len()
);
asked.extend(batch);
}
for gw in &present {
assert!(
asked.iter().filter(|a| *a == gw).count() >= 2,
"8 拍里每个账户至少被问到一轮: {gw} 实测 {asked:?}"
);
}
assert!(s.status_batch(&[]).is_empty(), "没有账户就不问");
let dup = vec!["gwA".to_string(), "gwA".to_string(), "gwB".to_string()];
let b = s.status_batch(&dup);
assert!(
b.len() <= ek_ui::state::STATUS_ASKS_PER_TICK,
"去重不放宽单拍上限,实测 {b:?}"
);
assert_eq!(
b.iter().filter(|g| *g == "gwA").count(),
1,
"同一拍里同一个网关只许问一次,实测 {b:?}"
);
assert!(!s.take_pipeline_due(), "没标记就不问");
s.request_pipeline_refresh();
assert!(s.take_pipeline_due(), "标记一次就问一次");
assert!(!s.take_pipeline_due(), "标记是消费性的,不是一档状态");
assert!(!s.take_trade_reseed(), "没标记就不播种");
s.request_trade_reseed();
assert!(s.take_trade_reseed(), "标记一次就播种一次");
assert!(!s.take_trade_reseed(), "标记是消费性的,不是一档状态");
s.request_pipeline_refresh();
s.request_pipeline_refresh();
assert!(s.take_pipeline_due());
assert!(!s.take_pipeline_due(), "两次标记不变成两次问(合并成一次)");
}
fn put_timer(engine: &MainEngine) {
engine.event_engine().put(Event::new(EVENT_TIMER)).unwrap();
}
#[test]
fn t0_remove_then_restart_loop() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
boot(
&engine,
&ek_ui::ui_kinds(&shared, true),
&StationConfig {
plugins: vec!["cfg".to_string(), "dm".to_string(), "oms".to_string(), "ui".to_string()],
},
)
.unwrap();
engine.start().unwrap();
wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
assert!(shared.snapshot().rows.iter().any(|r| r.name == "dm"));
shared.push_cmd(UiCmd::Remove {
name: "dm".to_string(),
});
put_timer(&engine);
wait_until("dm removed", 4, || {
shared.snapshot().rows.iter().all(|r| r.name != "dm")
});
assert!(
shared
.logs_snapshot()
.0
.iter()
.any(|l| l.contains("[ops] svctl.remove dm 完成")),
"remove log missing"
);
shared.push_cmd(UiCmd::Start {
factory: "dm".to_string(),
name: String::new(),
});
put_timer(&engine);
wait_until("dm back", 4, || {
shared
.snapshot()
.rows
.iter()
.any(|r| r.name == "dm" && r.state == "Running")
});
assert!(
shared
.logs_snapshot()
.0
.iter()
.any(|l| l.contains("svctl.start dm → dm 上线")),
"start log missing"
);
engine.close_all(Duration::from_secs(2));
}
#[test]
fn t1_unknown_factory_rejected_logged() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
boot(
&engine,
&ek_ui::ui_kinds(&shared, true),
&StationConfig {
plugins: vec!["cfg".to_string(), "dm".to_string(), "oms".to_string(), "ui".to_string()],
},
)
.unwrap();
engine.start().unwrap();
wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
shared.push_cmd(UiCmd::Start {
factory: "ghost".to_string(),
name: String::new(),
});
put_timer(&engine);
wait_until("rejected log", 4, || {
shared
.logs_snapshot()
.0
.iter()
.any(|l| l.contains("start ghost 被拒"))
});
engine.close_all(Duration::from_secs(2));
}
#[test]
fn t2_self_removal_refused_on_both_faces() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
boot(
&engine,
&ek_ui::ui_kinds(&shared, true),
&StationConfig {
plugins: vec!["cfg".to_string(), "dm".to_string(), "oms".to_string(), "ui".to_string()],
},
)
.unwrap();
engine.start().unwrap();
wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
assert!(
shared.snapshot().rows.iter().any(|r| r.name == "svctl"),
"svctl 应在场(它是 svctl.* 的提供方)"
);
shared.push_cmd(UiCmd::Remove {
name: "ui".to_string(),
});
put_timer(&engine);
wait_until("ui refusal logged", 4, || {
shared
.logs_snapshot()
.0
.iter()
.any(|l| l.contains("拒绝移除 ui"))
});
assert!(
shared
.logs_snapshot()
.0
.iter()
.all(|l| !l.contains("svctl.remove ui 完成")),
"不得真摘 ui"
);
shared.push_cmd(UiCmd::Remove {
name: "svctl".to_string(),
});
put_timer(&engine);
wait_until("svctl refusal logged", 4, || {
shared
.logs_snapshot()
.0
.iter()
.any(|l| l.contains("remove svctl 被拒"))
});
put_timer(&engine);
wait_until("both faces still listed", 4, || {
let rows = &shared.snapshot().rows;
rows.iter().any(|r| r.name == "svctl") && rows.iter().any(|r| r.name == "ui")
});
assert!(
shared
.snapshot()
.rows
.iter()
.all(|r| r.name != "dm" || r.state == "Running"),
"dm 不该被顺手摘掉"
);
engine.close_all(Duration::from_secs(2));
}
#[test]
fn t3_pipeline_status_reaches_ui_state() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
boot(
&engine,
&ek_ui::ui_kinds(&shared, true),
&StationConfig {
plugins: vec!["cfg".to_string(), "dm".to_string(), "oms".to_string(), "ui".to_string()],
},
)
.unwrap();
engine.start().unwrap();
wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
put_timer(&engine);
wait_until("pipeline lines", 6, || !shared.pipeline_lines().is_empty());
let lines = shared.pipeline_lines();
assert!(
lines
.iter()
.any(|l| l == "stage:validate|split|risk|send|book"),
"五段管线没显形: {lines:?}"
);
assert!(
lines.iter().any(|l| l.starts_with("floor:field|rate(")),
"限速/撤单守卫没显形: {lines:?}"
);
assert!(
lines.iter().any(|l| l == "splitter:ek-converter"),
"拆单器没显形: {lines:?}"
);
for gate in [
"gate:ek-risk:contract",
"gate:ek-risk:volume",
"gate:ek-risk:position-limit",
"gate:ek-risk:price-band",
] {
assert!(
lines.iter().any(|l| l == gate),
"工厂注入的风控闸 {gate} 没显形: {lines:?}"
);
}
shared.push_cmd(UiCmd::Remove {
name: "oms".to_string(),
});
put_timer(&engine);
wait_until("oms gone from rows", 6, || {
!shared.snapshot().rows.iter().any(|r| r.name == "oms")
});
put_timer(&engine);
wait_until("pipeline lines cleared", 6, || {
shared.pipeline_lines().is_empty()
});
assert!(
shared.pipeline_lines().is_empty(),
"摘掉 oms 之后管线读数必须清空,实测 {:?}",
shared.pipeline_lines()
);
engine.close_all(Duration::from_secs(2));
}
fn spill_row(name: &str, spilled: u64) -> ek_station::wire::PlugRow {
water_row(name, spilled, 0, 0, ek_core::plugin::MAILBOX_SOFT_CAP)
}
fn water_row(
name: &str,
spilled: u64,
pending: u64,
peak: u64,
soft_cap: u64,
) -> ek_station::wire::PlugRow {
ek_station::wire::PlugRow {
name: name.to_string(),
version: String::new(),
state: "Running".to_string(),
deps: Vec::new(),
spilled,
panics: 0,
pending,
peak,
soft_cap,
}
}
fn order_row(
gw: &str,
id: &str,
status: ek_data::constant::Status,
volume: f64,
price: f64,
) -> ek_data::order::OrderData {
ek_data::order::OrderData {
gateway_name: gw.to_string(),
symbol: "rb2605".to_string(),
exchange: ek_data::constant::Exchange::SHFE,
orderid: id.to_string(),
direction: Some(ek_data::constant::Direction::LONG),
offset: ek_data::constant::Offset::OPEN,
price,
volume,
status,
name: "螺纹钢2605".to_string(),
..ek_data::order::OrderData::default()
}
}
fn trade_row(id: &str) -> ek_data::trade::TradeData {
ek_data::trade::TradeData {
gateway_name: "ctp".to_string(),
symbol: "rb2605".to_string(),
exchange: ek_data::constant::Exchange::SHFE,
orderid: "A1".to_string(),
tradeid: id.to_string(),
direction: Some(ek_data::constant::Direction::LONG),
offset: ek_data::constant::Offset::OPEN,
price: 102.5,
volume: 1.0,
..ek_data::trade::TradeData::default()
}
}
fn position_row(gw: &str, symbol: &str, volume: f64) -> ek_data::position::PositionData {
ek_data::position::PositionData {
gateway_name: gw.to_string(),
symbol: symbol.to_string(),
exchange: ek_data::constant::Exchange::SHFE,
direction: ek_data::constant::Direction::LONG,
volume,
can_use_volume: volume,
..ek_data::position::PositionData::default()
}
}
fn account_row(gw: &str, acct: &str, balance: f64) -> ek_data::account::AccountData {
ek_data::account::AccountData {
gateway_name: gw.to_string(),
accountid: acct.to_string(),
balance,
..ek_data::account::AccountData::default()
}
}
#[test]
fn u4_order_book_merges_by_en_orderid_and_keeps_immutable_columns() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
s.note_order(&order_row(
"ctp",
"A1",
ek_data::constant::Status::SUBMITTING,
10.0,
102.5,
));
let mut later = order_row("ctp", "A1", ek_data::constant::Status::PARTTRADED, 0.0, 0.0);
later.traded = 4.0;
later.traded_price = 102.5;
s.note_order(&later);
let ts = s.trade_snap();
assert_eq!(ts.orders.len(), 1, "同 en_orderid 是一条行,不是两条");
let o = &ts.orders[0];
assert_eq!(
o.status,
ek_data::constant::Status::PARTTRADED,
"状态列可变"
);
assert_eq!(o.traded, 4.0, "已成列可变");
assert_eq!(o.traded_price, 102.5, "成交均价列可变");
assert_eq!(o.volume, 10.0, "总量列不可变:回报给 0 也不许写进表");
assert_eq!(o.price, 102.5, "委托价列不可变");
assert_eq!(
o.direction,
Some(ek_data::constant::Direction::LONG),
"方向列不可变"
);
assert_eq!(o.name, "螺纹钢2605", "名称列可变(基线 update True)");
}
#[test]
fn u4b_newest_event_is_the_first_row_but_seeding_keeps_order() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
s.note_order(&order_row(
"ctp",
"A1",
ek_data::constant::Status::SUBMITTING,
1.0,
1.0,
));
s.note_order(&order_row(
"ctp",
"A2",
ek_data::constant::Status::SUBMITTING,
2.0,
2.0,
));
let ts = s.trade_snap();
assert_eq!(ts.orders.len(), 2);
assert_eq!(ts.orders[0].orderid, "A2", "后到的排在前(基线顶部插入)");
s.seed_trade_snap(ek_ui::state::TradeSnap {
orders: vec![
order_row("ctp", "B1", ek_data::constant::Status::SUBMITTING, 1.0, 1.0),
order_row("ctp", "B2", ek_data::constant::Status::SUBMITTING, 2.0, 2.0),
],
trades: Vec::new(),
positions: Vec::new(),
accounts: Vec::new(),
});
let ts = s.trade_snap();
assert_eq!(
ts.orders
.iter()
.map(|o| o.orderid.clone())
.collect::<Vec<_>>(),
vec!["B1".to_string(), "B2".to_string()],
"播种保序:那是「已有的一份账」,不是一批新事件"
);
}
#[test]
fn u4c_clear_gateway_rows_voids_only_that_gateway() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
s.note_order(&order_row(
"gwA",
"A1",
ek_data::constant::Status::SUBMITTING,
1.0,
1.0,
));
s.note_order(&order_row(
"gwB",
"B1",
ek_data::constant::Status::SUBMITTING,
1.0,
1.0,
));
s.note_position(&position_row("gwA", "rb2605", 3.0));
s.note_funds(&account_row("gwA", "220380", 100.0));
assert_eq!(s.clear_gateway_rows("gwA"), 3, "委托 + 持仓 + 资金三行");
let ts = s.trade_snap();
assert_eq!(ts.orders.len(), 1);
assert_eq!(ts.orders[0].gateway_name, "gwB");
assert!(ts.positions.is_empty() && ts.accounts.is_empty());
assert_eq!(s.clear_gateway_rows("gwZ"), 0, "没这个名字就什么都不动");
}
#[test]
fn u4d_trade_book_caps_at_trade_cap_and_alerts_once() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
for i in 0..(ek_ui::state::TRADE_CAP + 5) {
s.note_trade(&trade_row(&format!("T{i}")));
}
let ts = s.trade_snap();
assert_eq!(ts.trades.len(), ek_ui::state::TRADE_CAP, "触顶就丢最旧");
assert_eq!(
ts.trades[0].tradeid,
format!("T{}", ek_ui::state::TRADE_CAP + 4),
"最新的一笔仍在最前"
);
assert_eq!(s.trade_dropped(), 5);
assert_eq!(s.pending_alert_count(), 1, "触顶只弹一次,不每笔弹");
}
#[test]
fn u4e_position_volume_is_mutable_and_the_row_survives_flattening() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
s.note_position(&position_row("ctp", "rb2605", 5.0));
s.note_position(&position_row("ctp", "rb2605", 0.0));
let ts = s.trade_snap();
assert_eq!(ts.positions.len(), 1, "同 en_positionid 归并成一条");
assert_eq!(ts.positions[0].volume, 0.0, "volume 可变:平仓写 0,不删行");
}
#[test]
fn u4f_bad_payload_is_counted_not_silently_dropped() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
s.note_bad_payload("eOrder.");
s.note_bad_payload("eOrder.");
assert_eq!(s.bad_payload_count("eOrder."), 2);
assert_eq!(s.bad_payload_count("eTrade."), 0, "逐线各记各的");
assert!(
s.logs_snapshot()
.0
.iter()
.any(|l| l.contains("eOrder. 载荷类型异常")),
"第一次必须显形(静默丢 = 界面少了一路数据而没人知道)"
);
}
#[test]
fn u4g_trade_dedups_by_en_tradeid_across_push_and_seed() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
s.note_trade(&trade_row("T1"));
s.note_trade(&trade_row("T1"));
assert_eq!(s.trade_snap().trades.len(), 1, "同一笔成交只算一次");
s.note_trade(&trade_row("T1"));
assert_eq!(
s.trade_snap().trades.len(),
1,
"播种与推送同规则,实测 {:?}",
s.trade_snap()
.trades
.iter()
.map(|t| t.tradeid.clone())
.collect::<Vec<_>>()
);
s.note_trade(&trade_row("T2"));
let ids: Vec<_> = s
.trade_snap()
.trades
.iter()
.map(|t| t.tradeid.clone())
.collect();
assert_eq!(ids, vec!["T2", "T1"], "新成交在前");
let mut other = trade_row("T1");
other.gateway_name = "ctp_2".to_string();
s.note_trade(&other);
assert_eq!(s.trade_snap().trades.len(), 3, "跨网关同号成交是两笔");
s.clear_gateway_rows("ctp");
assert_eq!(s.trade_snap().trades.len(), 1, "只清 ctp 名下");
s.note_trade(&trade_row("T1"));
assert_eq!(s.trade_snap().trades.len(), 2, "清掉的笔要能重新入账");
}
#[test]
fn u4h_trade_cap_evicts_the_dedup_key_too() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
for i in 0..=ek_ui::state::TRADE_CAP {
s.note_trade(&trade_row(&format!("T{i}")));
}
assert_eq!(
s.trade_snap().trades.len(),
ek_ui::state::TRADE_CAP,
"账本封顶"
);
assert_eq!(s.trade_dropped(), 1, "最旧那笔被挤掉");
s.note_trade(&trade_row("T0"));
let snap = s.trade_snap();
assert_eq!(snap.trades.len(), ek_ui::state::TRADE_CAP, "封顶不变");
assert_eq!(snap.trades[0].tradeid, "T0", "挤掉的笔再推还能上屏");
}
#[test]
fn u4i_chain_note_lands_only_from_chain_lines() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
let form = ek_ui::state::CtpForm {
broker_id: "9999".to_string(),
td_user: "278153".to_string(),
td_addr: "tcp://127.0.0.1:30001".to_string(),
..ek_ui::state::CtpForm::default()
};
let acct = ek_ui::state::AccountRow::derive(&form).expect("夹具账户必须可派生");
let gw = acct.gateway.clone();
s.add_account(acct).expect("夹具账户必须入列");
assert_eq!(
s.account(&gw).unwrap().chain_note,
None,
"没推过状态行 = None(显示「—」)"
);
s.note_chain("gw_absent", "前置已连接");
assert!(s.account(&gw).unwrap().chain_note.is_none());
s.note_chain(&gw, "前置已连接");
assert_eq!(
s.account(&gw).unwrap().chain_note.as_deref(),
Some("前置已连接")
);
s.note_chain(&gw, "合约未到手,本轮查询已停止");
assert_eq!(
s.account(&gw).unwrap().chain_note.as_deref(),
Some("合约未到手,本轮查询已停止"),
"「阶段」栏跟着最近一条状态行走"
);
s.note_connect_sent(&gw);
assert!(
s.account(&gw).unwrap().chain_note.is_none(),
"『连接』入列必须清掉上一轮的链状态"
);
s.note_chain(&gw, "资金/持仓轮询中");
s.sync_accounts_connected(&[]);
assert!(
s.account(&gw).unwrap().chain_note.is_none(),
"实例不在场还留着「资金/持仓轮询中」= 替一个不存在的会话说话"
);
s.sync_accounts_connected(&[spill_row(&gw, 0)]);
s.note_chain(&gw, "结算单已确认");
assert_eq!(
s.account(&gw).unwrap().chain_note.as_deref(),
Some("结算单已确认"),
"重新装配后链状态照落"
);
assert_eq!(ek_ui::state::CHAIN_STATE_PREFIX, "[链] ");
let state_src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/state.rs"))
.expect("读 src/state.rs")
.replace("\r\n", "\n");
let i = state_src
.find("pub fn apply_status(")
.expect("apply_status 必须在 state.rs");
let j = state_src[i..]
.find("\n }\n")
.expect("apply_status 函数体收尾");
let apply = &state_src[i..i + j];
assert!(
!apply.contains("chain_note"),
"链状态只是显示,不得参与在线/断供判据,实测 {}",
apply.chars().take(120).collect::<String>()
);
let plugin_src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/plugin.rs"))
.expect("读 src/plugin.rs")
.replace("\r\n", "\n");
assert!(
plugin_src.contains("strip_prefix(crate::state::CHAIN_STATE_PREFIX)"),
"eLog 分支筛状态行必须走常量,不是散在分支里的字面量"
);
assert!(plugin_src.contains("note_chain(&d.gateway_name"));
let render_src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/render.rs"))
.expect("读 src/render.rs")
.replace("\r\n", "\n");
assert!(
render_src.contains("const ACCOUNT_HEADERS: [&str; 9]"),
"账户表 9 列(0.9.10 加「td 前置」列,用户裁定:配了什么就要显示)"
);
assert!(
render_src.contains(".num_columns(9)"),
"account_grid 的列数必须跟着表头一起改(7 列表头 + 8 列 Grid 就是错位)"
);
}
#[test]
fn u5_trade_snap_is_rebuilt_only_when_the_books_are_dirty() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
s.note_order(&order_row(
"ctp",
"A1",
ek_data::constant::Status::SUBMITTING,
1.0,
1.0,
));
let a = s.trade_snap();
let b = s.trade_snap();
assert!(
Arc::ptr_eq(&a, &b),
"没新事件就不重建(同一份 Arc,不是每次重新组帧)"
);
s.note_order(&order_row(
"ctp",
"A2",
ek_data::constant::Status::SUBMITTING,
2.0,
2.0,
));
s.note_order(&order_row(
"ctp",
"A3",
ek_data::constant::Status::SUBMITTING,
3.0,
3.0,
));
let c = s.trade_snap();
assert!(!Arc::ptr_eq(&a, &c), "脏了必须重建");
assert_eq!(c.orders.len(), 3);
let d = s.trade_snap();
assert!(Arc::ptr_eq(&c, &d), "三条事件只重建一次");
}
#[test]
fn u6_water_and_soft_cap_alerts_dedupe_per_level() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let s = UiShared::new();
let (w1, c1) = ek_ui::state::pending_thresholds(262_144);
assert_eq!((w1, c1), (16_384, 65_536), "缺省软顶行的阈 = 1/16 与 1/4");
let (w2, c2) = ek_ui::state::pending_thresholds(100_000);
assert_eq!((w2, c2), (6_250, 32_768), "ui 行(10 万软顶)的阈各算各的");
let (w3, c3) = ek_ui::state::pending_thresholds(1_000);
assert!(
w3 >= 1 && c3 >= w3 && c3 <= 1_000,
"小软顶收口不炸:({w3},{c3})"
);
assert_eq!(
ek_ui::state::pending_thresholds(0),
(1, 1),
"软顶 0 的异常快照:阈值收口到 1,不炸也不静默"
);
s.note_plugin_spill(&[water_row("dm", 0, 0, 0, 262_144)]);
assert_eq!(s.pending_alert_count(), 0, "零水位零丢弃不该响");
s.note_plugin_spill(&[water_row("dm", 0, 16_384, 16_384, 262_144)]);
assert_eq!(s.pending_alert_count(), 1, "越黄阈弹一条");
s.note_plugin_spill(&[water_row("dm", 0, 20_000, 20_000, 262_144)]);
assert_eq!(s.pending_alert_count(), 1, "同一级不重复弹");
s.note_plugin_spill(&[water_row("dm", 0, 65_536, 65_536, 262_144)]);
assert_eq!(s.pending_alert_count(), 2, "越红阈是另一件事,必须再弹");
s.note_plugin_spill(&[water_row("dm", 1, 65_536, 65_536, 262_144)]);
assert_eq!(s.pending_alert_count(), 3, "已触顶丢新帧必须显形");
s.note_plugin_spill(&[water_row("dm", 99, 65_536, 65_536, 262_144)]);
assert_eq!(s.pending_alert_count(), 3, "同一级只弹一次(常显交给运维窗)");
let t = UiShared::new();
t.note_plugin_spill(&[water_row("ui", 0, 40_000, 40_000, 100_000)]);
assert_eq!(
t.pending_alert_count(),
2,
"ui 行 4 万积压:黄阈(6250)与红阈(32768)都越过"
);
}
#[test]
fn u7_ops_window_stat_color_paints_the_worst_item() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
const DARK: ek_ui::tokens::Theme = ek_ui::tokens::Theme::Dark;
let def = ek_core::plugin::MAILBOX_SOFT_CAP; assert_eq!(
ek_ui::render::stat_color(DARK, 0, 0, 0, def),
ek_ui::tokens::semantic::MUTED,
"0/0/0 的意思是「没事」,不是数据"
);
assert_eq!(
ek_ui::render::stat_color(DARK, 0, 0, 16_384, def),
ek_ui::tokens::semantic::WARN,
"积压越黄阈 → 黄"
);
assert_eq!(
ek_ui::render::stat_color(DARK, 0, 0, 65_536, def),
ek_ui::tokens::semantic::ERROR,
"积压越红阈(软顶 1/4)→ 红:再涨就丢新帧"
);
assert_eq!(
ek_ui::render::stat_color(DARK, 1, 0, 0, def),
ek_ui::tokens::semantic::ERROR,
"新口径里 spilled>0 = 已触顶丢新帧 = 红色大事"
);
assert_eq!(
ek_ui::render::stat_color(DARK, 0, 1, 0, def),
ek_ui::tokens::semantic::ERROR,
"panic 照旧红"
);
assert_eq!(
ek_ui::render::stat_color(DARK, 0, 0, 40_000, def),
ek_ui::tokens::semantic::WARN,
"缺省软顶行:4 万没到它的 1/4"
);
assert_eq!(
ek_ui::render::stat_color(DARK, 0, 0, 40_000, ek_ui::plugin::UI_MAILBOX_SOFT_CAP),
ek_ui::tokens::semantic::ERROR,
"ui 行(10 万软顶):4 万越过它的 1/4"
);
}
#[test]
fn t4_order_rows_arrive_by_push_not_by_polling() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
boot(
&engine,
&ek_ui::ui_kinds(&shared, true),
&StationConfig {
plugins: vec!["cfg".to_string(), "dm".to_string(), "oms".to_string(), "ui".to_string()],
},
)
.unwrap();
engine.start().unwrap();
wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
engine
.event_engine()
.put(Event::with_data(
"eOrder.",
Arc::new(order_row(
"gwA",
"A1",
ek_data::constant::Status::SUBMITTING,
3.0,
7.0,
)),
))
.unwrap();
wait_until("pushed order lands", 4, || {
shared.trade_snap().orders.iter().any(|o| o.orderid == "A1")
});
shared.push_cmd(UiCmd::Remove {
name: "oms".to_string(),
});
wait_until("oms removed", 4, || {
shared.snapshot().rows.iter().all(|r| r.name != "oms")
});
assert!(
shared.trade_snap().orders.iter().any(|o| o.orderid == "A1"),
"摘掉 oms 不清交易表:那些行属于网关 gwA,不属于 oms(0.4 的「oms 不在场就清空」随四问一起退役)"
);
engine
.event_engine()
.put(Event::with_data(
ek_core::EVENT_PLUG_DOWN,
Arc::new(PlugNotice {
name: "gwA".to_string(),
version: "0".to_string(),
}),
))
.unwrap();
wait_until("gateway rows voided", 4, || {
shared.trade_snap().orders.is_empty()
});
engine.close_all(Duration::from_secs(2));
}
#[test]
fn u8_refresh_asks_no_trading_data_and_the_push_lines_are_routed() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-ui/target/cfg-test-root"));
let src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/plugin.rs"))
.expect("读 src/plugin.rs")
.replace("\r\n", "\n");
let body = |needle: &str, stop: &str| -> String {
let i = src
.find(needle)
.unwrap_or_else(|| panic!("plugin.rs 里找不到 {needle:?}"));
let j = src[i..]
.find(stop)
.unwrap_or_else(|| panic!("{needle:?} 之后找不到 {stop:?}"));
src[i..i + j + stop.len()].to_string()
};
let refresh = body(" fn refresh(&self) {", "\n }\n");
assert!(
!refresh.contains("ask_service"),
"refresh 那一拍不许直接 ask(0.5 M3),实测 {refresh:?}"
);
assert!(
!refresh.contains("run_cmds"),
"指令辞行不在 eTimer 拍上(0.5 M5),实测 {refresh:?}"
);
let rec = body(
" fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {",
"\n }\n",
);
for svc in [
"\"oms.get_all_orders\"",
"\"oms.get_all_trades\"",
"\"oms.get_all_positions\"",
"\"oms.get_all_accounts\"",
] {
assert_eq!(
src.matches(svc).count(),
1,
"{svc} 只许出现一次(启动一次四问),不许回到周期轮询"
);
assert!(rec.contains(svc), "{svc} 必须在 reconcile_trade 里");
}
let start = body(" fn on_start(&self) {", "\n }\n");
assert!(
start.contains("spawn_cmd_thread"),
"启动 = 起命令线程(会等的东西都归它),实测 {start:?}"
);
let spawn = body(" fn spawn_cmd_thread(&self) {", "\n }\n");
assert!(
spawn.contains("reconcile_trade"),
"启动一次四问挂在命令线程里:`on_start` 是宿主启动链的一环,四口最坏 8 秒不许把后面插件的启动一起拖住"
);
assert!(
start.contains("request_trade_reseed"),
"启动播种走标记(消费方 = 命令线程),不在 on_start 里直接问,实测 {start:?}"
);
let mail = body(
" fn on_mail(&self, event: &ek_event::Event) {",
"\n }\n",
);
assert!(
mail.contains("request_trade_reseed") && mail.contains("request_pipeline_refresh"),
"网关 / oms 上线必须重播种一次并重问装配:摘除时作废的行要能回来,实测 {:?}",
mail.chars().take(160).collect::<String>()
);
assert!(
src.contains("pub const UI_MAILBOX_SOFT_CAP: u64 = 100_000"),
"邮箱软顶必须是命名常量(带理由),不是散在 mailbox_soft_cap 里的字面量(0.7.0 口径)"
);
let mb = body(" fn mailbox_soft_cap(&self) -> u64 {", "\n }\n");
assert!(
mb.contains("UI_MAILBOX_SOFT_CAP") && !mb.contains("1024"),
"mailbox_soft_cap 吃常量,实测 {mb:?}"
);
assert!(
!src.contains("fn mailbox_cap("),
"0.4 的容量覆写 `mailbox_cap` 必须从链上消失(邮箱已无界,它不再是容量)"
);
let routes = body(" fn routes(&self)", "\n }\n");
for line in ["W_ORDER", "W_TRADE", "W_POSITION", "W_ACCOUNT"] {
assert!(routes.contains(line), "routes 里缺 {line},实测 {routes:?}");
}
assert!(
!routes.contains("Cow::Borrowed(\"eTradeSync\")") && !routes.contains("EVENT_TRADE_SYNC"),
"eTradeSync. 是 oms 处理完成交后的重推,订两条就是同一笔成交入账两次(注释里提到它可以,订阅不行)"
);
assert!(
!src.contains("OMS_ASK_TIMEOUT") && !src.contains("merge_frame"),
"0.4.5 的超时分级与读数裁决随四问一起退役,不许留成死代码"
);
}