use std::sync::Arc;
use std::time::Duration;
use ek_core::main_engine::MainEngine;
use ek_event::{EVENT_TIMER, Event};
use ek_station::{StationConfig, boot_with_cfg};
use ek_ui::runtime::ui_kinds;
use ek_ui::state::{CtpForm, UiCmd, UiShared};
fn pin_cfg_root() {
ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("target/ui-ctp-seam"));
}
fn put_timer(engine: &Arc<MainEngine>) {
engine
.event_engine()
.put(Event::new(EVENT_TIMER.to_string()))
.expect("timer put");
}
fn wait_until(win: &str, secs: u64, mut f: impl FnMut() -> bool) {
for _ in 0..(secs * 20) {
if f() {
return;
}
std::thread::sleep(Duration::from_millis(50));
}
panic!("timeout: {win}");
}
fn log_has(shared: &UiShared, needle: &str) -> bool {
shared.logs_snapshot().0.iter().any(|l| l.contains(needle))
}
#[test]
fn c0_ui_ask_ctp_configure() {
pin_cfg_root();
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
let cfg = StationConfig {
plugins: vec![
"cfg".to_string(),
"ui".to_string(),
"dm".to_string(),
"oms".to_string(),
"ctp".to_string(),
],
};
let rig0 = ek_cfg::Store::with_root(std::path::PathBuf::from("target/ui-ctp-rig0"));
boot_with_cfg(&engine, &ui_kinds(&shared, false), &cfg, Some(&rig0)).unwrap();
engine.start().unwrap();
wait_until("空地址守卫日志", 6, || {
log_has(&shared, "未配置前置地址,跳过连接")
});
wait_until("服务目录含 configure", 8, || {
shared
.snapshot()
.services
.iter()
.any(|(name, ids)| name == "ctp" && ids.iter().any(|id| id == "ctp.configure"))
});
shared.push_cmd(UiCmd::CtpConfigure {
gateway: "ctp".to_string(),
form: CtpForm {
broker_id: "9999".to_string(),
td_addr: "tcp://127.0.0.1:30001".to_string(),
td_user: "777001".to_string(),
td_password: "pw".to_string(),
td_auth_code: "ac".to_string(),
td_app_id: "app".to_string(),
md_addr: "tcp://127.0.0.1:30011".to_string(),
..CtpForm::default()
},
});
put_timer(&engine);
wait_until("configure 受理日志", 8, || {
log_has(&shared, "ctp.configure 受理")
});
shared.push_cmd(UiCmd::CtpConfigure {
gateway: "ctp".to_string(),
form: CtpForm::default(),
});
put_timer(&engine);
wait_until("configure 被拒日志", 8, || {
log_has(&shared, "ctp.configure 被拒")
});
engine.close_all(std::time::Duration::from_secs(3));
}
#[test]
fn c1_ui_ask_ctp_subscribe() {
pin_cfg_root();
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
let cfg = StationConfig {
plugins: vec!["cfg".to_string(), "ctp".to_string(), "ui".to_string()],
};
let rig1 = ek_cfg::Store::with_root(std::path::PathBuf::from("target/ui-ctp-rig1"));
boot_with_cfg(&engine, &ui_kinds(&shared, false), &cfg, Some(&rig1)).unwrap();
engine.start().unwrap();
shared.push_cmd(UiCmd::CtpSubscribe {
gateway: "ctp".to_string(),
symbols: vec!["rb2605".to_string(), "sn2709".to_string()],
});
put_timer(&engine);
wait_until("subscribe 受理日志", 8, || {
log_has(&shared, "ctp.subscribe rb2605,sn2709 受理")
});
}
#[test]
fn c2_ask_missing_ctp_logs_failure() {
pin_cfg_root();
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
let cfg = StationConfig {
plugins: vec!["cfg".to_string(), "oms".to_string(), "ui".to_string()],
};
let rig2 = ek_cfg::Store::with_root(std::path::PathBuf::from("target/ui-ctp-rig2"));
boot_with_cfg(&engine, &ui_kinds(&shared, false), &cfg, Some(&rig2)).unwrap();
engine.start().unwrap();
shared.push_cmd(UiCmd::CtpConfigure {
gateway: "ctp".to_string(),
form: CtpForm {
td_addr: "tcp://127.0.0.1:30001".to_string(),
broker_id: "9999".to_string(),
td_user: "u".to_string(),
td_password: "p".to_string(),
..CtpForm::default()
},
});
put_timer(&engine);
wait_until("configure 失败日志(ctp 不在场)", 8, || {
log_has(&shared, "ctp.configure 失败")
});
put_timer(&engine);
wait_until("快照恢复", 6, || !shared.snapshot().rows.is_empty());
}
#[test]
fn c3_ctp_form_product_classes_reach_gateway() {
pin_cfg_root();
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
let cfg = StationConfig {
plugins: vec!["cfg".to_string(), "ctp".to_string(), "ui".to_string()],
};
let rig3 = ek_cfg::Store::with_root(std::path::PathBuf::from("target/ui-ctp-rig3"));
boot_with_cfg(&engine, &ui_kinds(&shared, false), &cfg, Some(&rig3)).unwrap();
engine.start().unwrap();
let mut form = CtpForm {
broker_id: "9999".to_string(),
td_addr: "tcp://182.254.243.31:40001".to_string(),
td_user: "777001".to_string(),
td_password: "pw".to_string(),
..CtpForm::default()
};
assert_eq!(form.product_classes(), vec!["1".to_string()]);
shared.push_cmd(UiCmd::CtpConfigure {
gateway: "ctp".to_string(),
form: form.clone(),
});
put_timer(&engine);
wait_until("缺省勾选回显 收录产品类:1", 8, || {
log_has(&shared, "ctp.configure 受理") && log_has(&shared, "收录产品类:1")
});
form.accept_options = true;
assert_eq!(
form.product_classes(),
vec!["1".to_string(), "2".to_string()],
"勾选 → CTP 码表:期货 '1' + 期权 '2'"
);
shared.push_cmd(UiCmd::CtpConfigure {
gateway: "ctp".to_string(),
form,
});
put_timer(&engine);
wait_until("勾期权后回显 收录产品类:1 2", 12, || {
log_has(&shared, "收录产品类:1 2")
});
engine.close_all(std::time::Duration::from_secs(3));
}
#[test]
fn c4_ui_polls_ctp_status_and_disconnect_keeps_instance() {
pin_cfg_root();
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
let cfg = StationConfig {
plugins: vec![
"cfg".to_string(),
"ui".to_string(),
"dm".to_string(),
"oms".to_string(),
"ctp".to_string(),
],
};
let rig4 = ek_cfg::Store::with_root(std::path::PathBuf::from("target/ui-ctp-rig4"));
boot_with_cfg(&engine, &ui_kinds(&shared, false), &cfg, Some(&rig4)).unwrap();
engine.start().unwrap();
let form = CtpForm {
broker_id: "9999".to_string(),
td_addr: "tcp://127.0.0.1:30001".to_string(),
td_user: "777001".to_string(),
td_password: "fake-pw-1234".to_string(),
td_auth_code: "fake-auth".to_string(),
td_app_id: "fake-app".to_string(),
md_addr: "tcp://127.0.0.1:30011".to_string(),
..CtpForm::default()
};
let gw = ek_ui::state::AccountRow::derive(&form).expect("派生注册名");
let gw = gw.gateway.clone();
assert_eq!(gw, "ctp_9999_777001", "派生名口径(寻址唯一用这个)");
shared
.add_account(ek_ui::state::AccountRow::derive(&form).expect("派生"))
.expect("添加账户");
assert_eq!(
shared.accounts()[0].stage,
ek_ui::state::AccountStage::NotAssembled,
"添加后仍是未装配"
);
shared.push_cmd(UiCmd::Connect {
gateway: gw.clone(),
form,
});
put_timer(&engine);
wait_until("svctl.start 上线 + configure 受理", 10, || {
log_has(&shared, &format!("svctl.start ctp → {gw} 上线"))
&& log_has(&shared, &format!("{gw}.configure 受理"))
});
wait_until("服务目录含 status 与 disconnect", 10, || {
shared.snapshot().services.iter().any(|(name, ids)| {
name == &gw
&& ids.iter().any(|id| id.ends_with(".status"))
&& ids.iter().any(|id| id.ends_with(".disconnect"))
})
});
wait_until("账户行标成已装配且在场", 10, || {
shared
.accounts()
.iter()
.any(|a| a.gateway == gw && a.connected)
});
for _ in 0..6 {
put_timer(&engine);
std::thread::sleep(Duration::from_millis(120));
}
let rows = shared.accounts();
let row = rows.iter().find(|a| a.gateway == gw).expect("行还在");
assert_eq!(
row.login,
ek_ui::state::LoginState::Unknown,
"纯逻辑传输登不上:读到状态口也必须是「未知」,不是「在线」"
);
let note = row.status_note.clone().unwrap_or_default();
assert!(
!note.contains("状态口问不到"),
"读到「问不到」= ask 没成功 = 两份 ek-ctp 的静默断(类型不同源):{note}"
);
shared.push_cmd(UiCmd::Disconnect {
gateway: gw.clone(),
});
put_timer(&engine);
wait_until("断开回执日志", 10, || {
log_has(&shared, &format!("断开 {gw}:")) && log_has(&shared, "会话已停")
});
let rows = shared.accounts();
let row = rows.iter().find(|a| a.gateway == gw).expect("行还在");
assert_eq!(
row.login,
ek_ui::state::LoginState::Stopped,
"徽标须变「已断开」"
);
assert_eq!(
row.stage,
ek_ui::state::AccountStage::Assembled,
"断开不改装配态:实例还在(移除才是停实例)"
);
assert!(row.connected, "断开后实例仍在场");
for _ in 0..6 {
put_timer(&engine);
std::thread::sleep(Duration::from_millis(120));
}
let rows = shared.accounts();
let row = rows.iter().find(|a| a.gateway == gw).expect("行还在");
assert_eq!(
row.login,
ek_ui::state::LoginState::Stopped,
"轮询后仍是「已断开」(主动停会话优先于计数器)"
);
assert!(
shared
.snapshot()
.services
.iter()
.any(|(name, ids)| name == &gw && ids.iter().any(|id| id.ends_with(".status"))),
"断开后实例仍在场(否则『断开』就等于移除)"
);
engine.close_all(std::time::Duration::from_secs(3));
}
#[test]
fn c5_status_poll_skips_rows_without_instance() {
pin_cfg_root();
let engine = Arc::new(MainEngine::new(None));
let shared = Arc::new(UiShared::new());
let cfg = StationConfig {
plugins: vec!["cfg".to_string(), "ui".to_string(), "ctp".to_string()],
};
let rig5 = ek_cfg::Store::with_root(std::path::PathBuf::from("target/ui-ctp-rig5"));
boot_with_cfg(&engine, &ui_kinds(&shared, false), &cfg, Some(&rig5)).unwrap();
engine.start().unwrap();
let form = CtpForm {
broker_id: "9999".to_string(),
td_addr: "tcp://127.0.0.1:30001".to_string(),
td_user: "777002".to_string(),
td_password: "fake-pw-1234".to_string(),
td_auth_code: "fake-auth".to_string(),
td_app_id: "fake-app".to_string(),
..CtpForm::default()
};
let row = ek_ui::state::AccountRow::derive(&form).expect("派生注册名");
let gw = row.gateway.clone();
shared.add_account(row).expect("添加(不连接)");
for _ in 0..6 {
put_timer(&engine);
std::thread::sleep(Duration::from_millis(120));
}
let rows = shared.accounts();
let row = rows.iter().find(|a| a.gateway == gw).expect("行还在");
assert!(!row.connected, "没有实例 → 不在场");
assert_eq!(row.stage, ek_ui::state::AccountStage::NotAssembled);
assert!(
row.status_note.is_none(),
"未装配的行不得被轮询(问不到会留下「状态口问不到」痕迹,且每次都要等超时)"
);
assert_eq!(row.login, ek_ui::state::LoginState::Unknown);
}