diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index 8f15dc51..2e283c01 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -3158,6 +3158,230 @@ mod tests { assert_eq!(parsed["fields"], serde_json::json!([])); } + #[cfg(feature = "libsql")] + #[tokio::test] + async fn test_extensions_weixin_login_poll_broadcasts_auth_completed_and_activates() { + use axum::body::Body; + use tokio::time::{Duration, timeout}; + use tower::ServiceExt; + + let secrets = test_secrets_store(); + let (ext_mgr, _wasm_tools_dir, wasm_channels_dir, db, _db_tmp) = + test_ext_mgr_with_db(secrets.clone()).await; + + std::fs::write(wasm_channels_dir.path().join("weixin.wasm"), b"\0asm fake") + .expect("write fake weixin wasm"); + let caps = serde_json::json!({ + "type": "channel", + "name": "weixin", + "setup": { + "required_secrets": [ + {"name": "weixin_bot_token", "prompt": "Connect Weixin"} + ] + }, + "capabilities": { + "channel": { + "allowed_paths": ["/webhook/weixin"] + } + }, + "config": { + "base_url": "https://ilinkai.weixin.qq.com", + "bot_type": "3" + } + }); + std::fs::write( + wasm_channels_dir.path().join("weixin.capabilities.json"), + serde_json::to_string(&caps).expect("serialize weixin caps"), + ) + .expect("write weixin capabilities"); + + let channel_manager = Arc::new(crate::channels::ChannelManager::new()); + let runtime = Arc::new( + crate::channels::wasm::WasmChannelRuntime::new( + crate::channels::wasm::WasmChannelRuntimeConfig::for_testing(), + ) + .expect("runtime"), + ); + let pairing_store = Arc::new(crate::pairing::PairingStore::new()); + let router = Arc::new(crate::channels::wasm::WasmChannelRouter::new()); + ext_mgr + .set_channel_runtime( + Arc::clone(&channel_manager), + Arc::clone(&runtime), + Arc::clone(&pairing_store), + Arc::clone(&router), + std::collections::HashMap::new(), + ) + .await; + ext_mgr + .set_test_wasm_channel_loader(Arc::new({ + let runtime = Arc::clone(&runtime); + let pairing_store = Arc::clone(&pairing_store); + move |name| { + Ok(make_test_loaded_channel( + Arc::clone(&runtime), + name, + Arc::clone(&pairing_store), + )) + } + })) + .await; + ext_mgr + .set_test_weixin_login_starter(Arc::new(|user_id, base_url, bot_type| { + Ok(( + crate::extensions::weixin_login::PendingWeixinLogin { + user_id: user_id.to_string(), + session_id: "weixin-session-42".to_string(), + qrcode: "qr-42".to_string(), + qr_code_url: "https://qr.example/42".to_string(), + started_at: std::time::Instant::now(), + base_url: base_url.to_string(), + bot_type: bot_type.to_string(), + refresh_count: 0, + }, + crate::extensions::InteractiveLoginStartResult { + session_id: "weixin-session-42".to_string(), + status: "pending".to_string(), + message: "Scan the QR code in Weixin to finish connecting.".to_string(), + qr_code_url: Some("https://qr.example/42".to_string()), + instructions: Some( + "Keep this window open while you scan and confirm on your phone." + .to_string(), + ), + }, + )) + })) + .await; + ext_mgr + .set_test_weixin_login_poller(Arc::new(|session| { + if session.session_id != "weixin-session-42" { + return Err(crate::extensions::ExtensionError::Other(format!( + "unexpected session id: {}", + session.session_id + ))); + } + + Ok( + crate::extensions::weixin_login::WeixinLoginPollOutcome::Confirmed( + crate::extensions::weixin_login::ConfirmedWeixinLogin { + bot_token: "weixin-token-42".to_string(), + base_url: Some("https://weixin.example".to_string()), + ilink_bot_id: "wx-bot-42".to_string(), + }, + ), + ) + })) + .await; + + let state = test_gateway_state(Some(ext_mgr.clone())); + let mut receiver = state.sse.sender().subscribe(); + let app = Router::new() + .route( + "/api/extensions/{name}/login/start", + post(extensions_login_start_handler), + ) + .route( + "/api/extensions/{name}/login/poll", + post(extensions_login_poll_handler), + ) + .with_state(state); + + let mut start_req = axum::http::Request::builder() + .method("POST") + .uri("/api/extensions/weixin/login/start") + .header("content-type", "application/json") + .body(Body::from(r#"{"force":true}"#)) + .expect("start request"); + start_req.extensions_mut().insert(UserIdentity { + user_id: "test".to_string(), + workspace_read_scopes: Vec::new(), + }); + + let start_resp = ServiceExt::>::oneshot(app.clone(), start_req) + .await + .expect("start response"); + assert_eq!(start_resp.status(), StatusCode::OK); + let start_body = axum::body::to_bytes(start_resp.into_body(), 1024 * 64) + .await + .expect("start body"); + let start_json: serde_json::Value = + serde_json::from_slice(&start_body).expect("start json response"); + assert_eq!(start_json["success"], serde_json::Value::Bool(true)); + assert_eq!(start_json["status"], "pending"); + assert_eq!(start_json["session_id"], "weixin-session-42"); + assert_eq!(start_json["qr_code_url"], "https://qr.example/42"); + + let mut poll_req = axum::http::Request::builder() + .method("POST") + .uri("/api/extensions/weixin/login/poll") + .header("content-type", "application/json") + .body(Body::from(r#"{"session_id":"weixin-session-42"}"#)) + .expect("poll request"); + poll_req.extensions_mut().insert(UserIdentity { + user_id: "test".to_string(), + workspace_read_scopes: Vec::new(), + }); + + let poll_resp = ServiceExt::>::oneshot(app, poll_req) + .await + .expect("poll response"); + assert_eq!(poll_resp.status(), StatusCode::OK); + let poll_body = axum::body::to_bytes(poll_resp.into_body(), 1024 * 64) + .await + .expect("poll body"); + let poll_json: serde_json::Value = + serde_json::from_slice(&poll_body).expect("poll json response"); + assert_eq!(poll_json["success"], serde_json::Value::Bool(true)); + assert_eq!(poll_json["status"], "succeeded"); + assert_eq!(poll_json["activated"], serde_json::Value::Bool(true)); + assert!( + poll_json["message"] + .as_str() + .unwrap_or_default() + .contains("Weixin connected as wx-bot-42"), + "unexpected poll message: {poll_json:?}" + ); + + let auth_completed = timeout(Duration::from_secs(1), async { + loop { + match receiver.recv().await { + Ok(scoped) => match scoped.event { + crate::channels::web::types::SseEvent::AuthCompleted { + extension_name, + success, + message, + } => break (extension_name, success, message), + _ => continue, + }, + Err(error) => panic!("expected auth_completed event, got recv error: {error}"), + } + } + }) + .await + .expect("timed out waiting for auth_completed"); + assert_eq!(auth_completed.0, "weixin"); + assert!(auth_completed.1); + assert!(auth_completed.2.contains("Weixin connected as wx-bot-42")); + + assert!( + secrets + .exists("test", "weixin_bot_token") + .await + .expect("check weixin secret"), + "weixin token should be stored after successful poll" + ); + assert!( + channel_manager.get_channel("weixin").await.is_some(), + "weixin should be hot-added after successful poll" + ); + assert_eq!( + db.get_setting("test", "extensions.weixin.base_url") + .await + .expect("get weixin base_url setting"), + Some(serde_json::json!("https://weixin.example")) + ); + } + #[tokio::test] async fn test_extensions_setup_submit_returns_failure_when_not_activated() { use axum::body::Body; @@ -3860,6 +4084,66 @@ mod tests { (ext_mgr, wasm_tools_dir, wasm_channels_dir) } + #[cfg(feature = "libsql")] + async fn test_ext_mgr_with_db( + secrets: Arc, + ) -> ( + Arc, + tempfile::TempDir, + tempfile::TempDir, + Arc, + tempfile::TempDir, + ) { + let tool_registry = Arc::new(ToolRegistry::new()); + let mcp_sm = Arc::new(crate::tools::mcp::session::McpSessionManager::new()); + let mcp_pm = Arc::new(crate::tools::mcp::process::McpProcessManager::new()); + let wasm_tools_dir = tempfile::tempdir().expect("temp wasm tools dir"); + let wasm_channels_dir = tempfile::tempdir().expect("temp wasm channels dir"); + let (db, db_tmp) = crate::testing::test_db().await; + let ext_mgr = Arc::new(ExtensionManager::new( + mcp_sm, + mcp_pm, + secrets, + tool_registry, + None, + None, + wasm_tools_dir.path().to_path_buf(), + wasm_channels_dir.path().to_path_buf(), + None, + "test".to_string(), + Some(db.clone()), + vec![], + )); + (ext_mgr, wasm_tools_dir, wasm_channels_dir, db, db_tmp) + } + + #[cfg(feature = "libsql")] + fn make_test_loaded_channel( + runtime: Arc, + name: &str, + pairing_store: Arc, + ) -> crate::channels::wasm::LoadedChannel { + let prepared = Arc::new(crate::channels::wasm::PreparedChannelModule::for_testing( + name, + format!("Mock channel: {name}"), + )); + let capabilities = crate::channels::wasm::ChannelCapabilities::for_channel(name) + .with_path(format!("/webhook/{name}")); + + crate::channels::wasm::LoadedChannel { + channel: crate::channels::wasm::WasmChannel::new( + runtime, + prepared, + capabilities, + "default", + "{}".to_string(), + pairing_store, + None, + ), + capabilities_file: None, + } + } + #[tokio::test] async fn test_relay_oauth_callback_missing_state_param() { use axum::body::Body; diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index fede74eb..878f22ae 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -120,6 +120,20 @@ type TestWasmChannelLoader = #[cfg(test)] type TestTelegramBindingResolver = Arc) -> Result + Send + Sync>; +#[cfg(test)] +type TestWeixinLoginStarter = Arc< + dyn Fn( + &str, + &str, + &str, + ) -> Result<(PendingWeixinLogin, InteractiveLoginStartResult), ExtensionError> + + Send + + Sync, +>; +#[cfg(test)] +type TestWeixinLoginPoller = Arc< + dyn Fn(&mut PendingWeixinLogin) -> Result + Send + Sync, +>; const TELEGRAM_OWNER_BIND_TIMEOUT_SECS: u64 = 120; const TELEGRAM_OWNER_BIND_CHALLENGE_TTL_SECS: u64 = 300; @@ -442,6 +456,10 @@ pub struct ExtensionManager { test_wasm_channel_loader: RwLock>, #[cfg(test)] test_telegram_binding_resolver: RwLock>, + #[cfg(test)] + test_weixin_login_starter: RwLock>, + #[cfg(test)] + test_weixin_login_poller: RwLock>, } /// Sanitize a URL for logging by removing query parameters and credentials. @@ -556,11 +574,15 @@ impl ExtensionManager { test_wasm_channel_loader: RwLock::new(None), #[cfg(test)] test_telegram_binding_resolver: RwLock::new(None), + #[cfg(test)] + test_weixin_login_starter: RwLock::new(None), + #[cfg(test)] + test_weixin_login_poller: RwLock::new(None), } } #[cfg(test)] - async fn set_test_wasm_channel_loader(&self, loader: TestWasmChannelLoader) { + pub(crate) async fn set_test_wasm_channel_loader(&self, loader: TestWasmChannelLoader) { *self.test_wasm_channel_loader.write().await = Some(loader); } @@ -569,6 +591,16 @@ impl ExtensionManager { *self.test_telegram_binding_resolver.write().await = Some(resolver); } + #[cfg(test)] + pub(crate) async fn set_test_weixin_login_starter(&self, starter: TestWeixinLoginStarter) { + *self.test_weixin_login_starter.write().await = Some(starter); + } + + #[cfg(test)] + pub(crate) async fn set_test_weixin_login_poller(&self, poller: TestWeixinLoginPoller) { + *self.test_weixin_login_poller.write().await = Some(poller); + } + #[cfg(test)] pub(crate) async fn set_test_telegram_pending_verification( &self, @@ -4743,7 +4775,17 @@ impl ExtensionManager { let base_url = self.resolve_weixin_base_url(user_id).await; let bot_type = self.resolve_weixin_bot_type().await; - let (session, result) = start_weixin_login(user_id, &base_url, &bot_type).await?; + #[cfg(test)] + let login_result = + if let Some(starter) = self.test_weixin_login_starter.read().await.as_ref() { + starter(user_id, &base_url, &bot_type) + } else { + start_weixin_login(user_id, &base_url, &bot_type).await + }; + #[cfg(not(test))] + let login_result = start_weixin_login(user_id, &base_url, &bot_type).await; + + let (session, result) = login_result?; self.pending_weixin_logins .write() @@ -4789,7 +4831,15 @@ impl ExtensionManager { )); } + #[cfg(test)] + let outcome = if let Some(poller) = self.test_weixin_login_poller.read().await.as_ref() { + poller(session) + } else { + poll_weixin_login(session).await + }?; + #[cfg(not(test))] let outcome = poll_weixin_login(session).await?; + match outcome { WeixinLoginPollOutcome::Pending(result) => { if matches!(result.status.as_str(), "failed") { @@ -5944,8 +5994,13 @@ mod tests { normalize_hosted_callback_url, send_telegram_text_message, telegram_message_matches_verification_code, }; + use crate::extensions::weixin_login::{ + ConfirmedWeixinLogin, PendingWeixinLogin, WEIXIN_BASE_URL_SETTING_PATH, + WeixinLoginPollOutcome, + }; use crate::extensions::{ - ExtensionError, ExtensionKind, ExtensionSource, InstallResult, VerificationChallenge, + ExtensionError, ExtensionKind, ExtensionSource, InstallResult, InteractiveLoginStartResult, + VerificationChallenge, }; use crate::pairing::PairingStore; @@ -6893,6 +6948,203 @@ mod tests { ) } + #[cfg(feature = "libsql")] + #[tokio::test] + async fn test_weixin_interactive_login_poll_persists_state_and_activates() -> Result<(), String> + { + let dir = tempfile::tempdir().map_err(|err| format!("temp dir: {err}"))?; + let channels_dir = dir.path().join("channels"); + std::fs::create_dir_all(&channels_dir).map_err(|err| format!("channels dir: {err}"))?; + std::fs::write(channels_dir.join("weixin.wasm"), b"mock") + .map_err(|err| format!("write wasm: {err}"))?; + std::fs::write( + channels_dir.join("weixin.capabilities.json"), + serde_json::to_vec(&serde_json::json!({ + "type": "channel", + "name": "weixin", + "setup": { + "required_secrets": [ + { + "name": "weixin_bot_token", + "prompt": "Connect Weixin", + "optional": false + } + ] + }, + "capabilities": { + "channel": { + "allowed_paths": ["/webhook/weixin"] + } + }, + "config": { + "base_url": "https://ilinkai.weixin.qq.com", + "bot_type": "3" + } + })) + .map_err(|err| format!("serialize capabilities: {err}"))?, + ) + .map_err(|err| format!("write capabilities: {err}"))?; + + let (db, _db_tmp) = crate::testing::test_db().await; + let manager = { + use crate::secrets::{InMemorySecretsStore, SecretsCrypto}; + use crate::testing::credentials::TEST_CRYPTO_KEY; + use crate::tools::ToolRegistry; + use crate::tools::mcp::process::McpProcessManager; + use crate::tools::mcp::session::McpSessionManager; + + let master_key = secrecy::SecretString::from(TEST_CRYPTO_KEY.to_string()); + let crypto = Arc::new( + SecretsCrypto::new(master_key) + .unwrap_or_else(|err| panic!("failed to construct test crypto: {err}")), + ); + + Arc::new(ExtensionManager::new( + Arc::new(McpSessionManager::new()), + Arc::new(McpProcessManager::new()), + Arc::new(InMemorySecretsStore::new(crypto)), + Arc::new(ToolRegistry::new()), + None, + None, + dir.path().join("tools"), + channels_dir.clone(), + None, + "test".to_string(), + Some(db.clone()), + Vec::new(), + )) + }; + + let channel_manager = Arc::new(ChannelManager::new()); + let runtime = Arc::new( + WasmChannelRuntime::new(WasmChannelRuntimeConfig::for_testing()) + .map_err(|err| format!("runtime: {err}"))?, + ); + let pairing_store = Arc::new(PairingStore::with_base_dir( + dir.path().join("pairing-state"), + )); + let router = Arc::new(WasmChannelRouter::new()); + manager + .set_channel_runtime( + Arc::clone(&channel_manager), + Arc::clone(&runtime), + Arc::clone(&pairing_store), + Arc::clone(&router), + std::collections::HashMap::new(), + ) + .await; + manager + .set_test_wasm_channel_loader(Arc::new({ + let runtime = Arc::clone(&runtime); + let pairing_store = Arc::clone(&pairing_store); + move |name| { + Ok(make_test_loaded_channel( + Arc::clone(&runtime), + name, + Arc::clone(&pairing_store), + )) + } + })) + .await; + manager + .set_test_weixin_login_starter(Arc::new(|user_id, base_url, bot_type| { + Ok(( + PendingWeixinLogin { + user_id: user_id.to_string(), + session_id: "weixin-session-1".to_string(), + qrcode: "qr-123".to_string(), + qr_code_url: "https://qr.example/one".to_string(), + started_at: std::time::Instant::now(), + base_url: base_url.to_string(), + bot_type: bot_type.to_string(), + refresh_count: 0, + }, + InteractiveLoginStartResult { + session_id: "weixin-session-1".to_string(), + status: "pending".to_string(), + message: "Scan the QR code in Weixin to finish connecting.".to_string(), + qr_code_url: Some("https://qr.example/one".to_string()), + instructions: Some( + "Keep this window open while you scan and confirm on your phone." + .to_string(), + ), + }, + )) + })) + .await; + manager + .set_test_weixin_login_poller(Arc::new(|session| { + if session.session_id != "weixin-session-1" { + return Err(ExtensionError::Other(format!( + "unexpected session id: {}", + session.session_id + ))); + } + Ok(WeixinLoginPollOutcome::Confirmed(ConfirmedWeixinLogin { + bot_token: "weixin-token-123".to_string(), + base_url: Some("https://weixin.example".to_string()), + ilink_bot_id: "wx-bot-1".to_string(), + })) + })) + .await; + + let start = manager + .start_interactive_login("weixin", "test") + .await + .map_err(|err| format!("start interactive login: {err}"))?; + require_eq( + start.session_id.clone(), + "weixin-session-1".to_string(), + "start session id", + )?; + require_eq(start.status, "pending".to_string(), "start status")?; + + let poll = manager + .poll_interactive_login("weixin", &start.session_id, "test") + .await + .map_err(|err| format!("poll interactive login: {err}"))?; + + require_eq(poll.status, "succeeded".to_string(), "poll status")?; + require_eq(poll.activated, Some(true), "poll activated")?; + require( + poll.message.contains("Weixin connected as wx-bot-1"), + format!("unexpected poll message: {}", poll.message), + )?; + require( + manager.active_channel_names.read().await.contains("weixin"), + "weixin should be marked active after successful login", + )?; + require( + channel_manager.get_channel("weixin").await.is_some(), + "weixin should be hot-added to the running channel manager", + )?; + require_eq( + manager.load_persisted_active_channels("test").await, + vec!["weixin".to_string()], + "persisted active channels", + )?; + require( + manager + .secrets + .exists("test", "weixin_bot_token") + .await + .map_err(|err| format!("check stored weixin token: {err}"))?, + "weixin bot token should be stored after successful login", + )?; + let persisted_base_url = manager + .store + .as_ref() + .ok_or_else(|| "db-backed manager missing".to_string())? + .get_setting("test", WEIXIN_BASE_URL_SETTING_PATH) + .await + .map_err(|err| format!("weixin base_url setting query: {err}"))?; + require_eq( + persisted_base_url, + Some(serde_json::json!("https://weixin.example")), + "weixin base_url setting", + ) + } + #[tokio::test] async fn test_telegram_hot_activation_returns_verification_challenge_before_binding() -> Result<(), String> {