ek-ui 0.2.21

//! S6 总线级闭环:ui 插件 cmd 队列 → ask `ctp.configure` / `ctp.subscribe`
//! (ek-ctp 0.3.0 插件自带配置口;纯逻辑传输,native 关)。
//!
//! 判据:configure 受理日志 + 空地址守卫日志消失路径 + 订阅受理日志 +
//! 服务目录含 ctp.configure(ui 快照面)。

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};

use ek_ui::runtime::ui_kinds;
use ek_ui::state::{CtpForm, UiCmd, UiShared};

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))
}

/// configure 全链:空配置守卫日志在 → 手工表单入列 → 辞行 → ctp.configure 受理
/// → 传输层未接(纯逻辑)也不坏链;服务目录含 configure 口。
#[test]
fn c0_ui_ask_ctp_configure() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        // ui 先上:eLog 订阅先于 ctp on_start 的守卫日志(装配序即观察序)。
        plugins: vec![
            "ui".to_string(),
            "dm".to_string(),
            "oms".to_string(),
            "ctp".to_string(),
        ],
    };
    boot(&engine, &ui_kinds(&shared, false), &cfg).unwrap();
    engine.start().unwrap();

    // on_start 空配置 → connect 空地址守卫(插件自持连接;不瞎连)。
    wait_until("空地址守卫日志", 6, || {
        log_has(&shared, "未配置前置地址,跳过连接")
    });

    // 服务目录面:ui 快照应看到 ctp 的 configure 口(0.3.0 服务表 5 口)。
    wait_until("服务目录含 configure", 8, || {
        shared
            .snapshot()
            .services
            .iter()
            .any(|(name, ids)| name == "ctp" && ids.iter().any(|id| id == "ctp.configure"))
    });

    // 界面表单(模拟手工输入)→ cmd → ask。
    // 多账号面:cmd 随令携带网关名(本测试用 boot 预装的 `ctp` 实例)。
    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 被拒")
    });
}

/// subscribe 全链:订阅入列 → ctp.subscribe 受理(登录前登记面,纯逻辑无登录)。
#[test]
fn c1_ui_ask_ctp_subscribe() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec!["ctp".to_string(), "ui".to_string()],
    };
    boot(&engine, &ui_kinds(&shared, false), &cfg).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 受理")
    });
}

/// 坏表单不进总线面:空订阅在渲染层就被拦(此层只测 cmd 面 ——
/// 渲染拦截由 kittest s13 钉;这里钉 ask 失败面:ctp 不在场 → 失败日志不炸链)。
#[test]
fn c2_ask_missing_ctp_logs_failure() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec!["oms".to_string(), "ui".to_string()],
    };
    boot(&engine, &ui_kinds(&shared, false), &cfg).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 失败")
    });
    // 链不死:后续 eTimer 拍照常组快照。
    put_timer(&engine);
    wait_until("快照恢复", 6, || !shared.snapshot().rows.is_empty());
}

/// 收录勾选全链(0.2.12,随 ek-ctp 0.6.0 D-PC-FILTER):表单勾选必须进网关,
/// 受理日志回显生效值(`收录产品类:1` / `收录产品类:1 2`)。
///
/// 这条同时是「进程里两份 ek-ctp」的哨兵:本 crate 直接 ask 网关(载荷类型来自
/// ek-ctp),网关却由 station 工厂装配 —— 版本不同源时 `ask` 在网关 arm 处 downcast
/// 失败、不回复,日志里出现的是 `ctp.configure 失败` 而不是「受理」,**编译期不报错**。
/// 实测过的那次断链(ek-station 抬 ek-ctp 而 ek-oms 留旧版)就是这个形状。
#[test]
fn c3_ctp_form_product_classes_reach_gateway() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec!["ctp".to_string(), "ui".to_string()],
    };
    boot(&engine, &ui_kinds(&shared, false), &cfg).unwrap();
    engine.start().unwrap();

    // 缺省只勾期货(与 0.2.11 口径一致:只期货上抛)。
    let mut 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(),
        ..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")
    });

    // 勾上期权 → 生效值必须变成 `1 2`(网关侧归一化排序去重后空格连接)。
    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", 8, || {
        log_has(&shared, "收录产品类:1 2")
    });
}

/// c4 状态口 + 断开口总线级闭环(0.2.19 规划 P3 的跨 crate 钉)。
///
/// 判据分三层,缺一层都不算过:
/// 1. **类型同源**:`{gw}.status` 的 ask 必须真拿到 `GatewayStatus`。两份 ek-ctp 时
///    `type_name` 字符串守卫放行、网关 arm 的 TypeId downcast 失败 → 不回复 → 超时 →
///    界面走「状态口问不到」分支留下痕迹。所以"读数里没有『问不到』"就是 downcast 成功的证据。
/// 2. **诚实口径**:纯逻辑传输永远登不上 → 状态口读到了也必须显示「未知」,
///    不得因为 `td_handle_up` 为真就说「在线」(ek-ctp D-STATUS ①)。
/// 3. **断开 ≠ 移除**:`{gw}.disconnect` 后实例仍在场(服务目录还在),徽标变「已断开」,
///    且后续轮询不得把它翻回「未知」(用户主动停的会话比计数器更贴近意图)。
#[test]
fn c4_ui_polls_ctp_status_and_disconnect_keeps_instance() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec![
            "ui".to_string(),
            "dm".to_string(),
            "oms".to_string(),
            "ctp".to_string(),
        ],
    };
    boot(&engine, &ui_kinds(&shared, false), &cfg).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,
        "添加后仍是未装配"
    );

    // 『连接』:一次做完 svctl.start + {gw}.configure。
    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 受理"))
    });

    // 服务目录面:新实例的 7 口里必须含 status / disconnect(ui 快照 = 界面能看见的事实)。
    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"))
        })
    });

    // 在场位回填(svctl.list 快照)。
    wait_until("账户行标成已装配且在场", 10, || {
        shared
            .accounts()
            .iter()
            .any(|a| a.gateway == gw && a.connected)
    });

    // 状态轮询(每 2 拍一次):多喂几拍确保至少问到一次。
    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, "断开后实例仍在场");

    // 后续轮询不得把主动断开翻回「未知」(裁定 1 的界面侧后果)。
    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"))),
        "断开后实例仍在场(否则『断开』就等于移除)"
    );
}

/// c5 轮询守卫:未装配/不在场的账户**不发** `{gw}.status` 问令。
///
/// 为什么值得钉:实例被摘除后清册行还在,若照问不误,每次 ask 都要等满 2 秒超时,
/// 按账户串行就是把整个刷新循环按账户数拖住 —— 而那种账户本来也没有会话可问。
#[test]
fn c5_status_poll_skips_rows_without_instance() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec!["ui".to_string(), "ctp".to_string()],
    };
    boot(&engine, &ui_kinds(&shared, false), &cfg).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);
}