v0.2.7-pre.009

crates/ksp-onchain-transport-lib/unit_tests/ws_transactions.rs

@@ -0,0 +1,122 @@
+// file: crates/ksp-onchain-transport-lib/unit_tests/ws_transactions.rs
+// version: 1
+
+use futures_util::SinkExt; // rust-rules: trait-import
+use futures_util::StreamExt; // rust-rules: trait-import
+
+fn local_endpoint(url: &str) -> crate::WsEndpointSettings {
+    return crate::WsEndpointSettings::new(
+        "local_ws_transactions",
+        true,
+        crate::WsProviderName::new("local-fixture"),
+        crate::WsClusterName::new("local"),
+        crate::WsProtocolKind::SolanaStandard,
+        crate::WsEndpointUrl::parse(url).expect("local test WebSocket URL must parse"),
+        crate::WsSessionSettings::default(),
+    );
+}
+
+async fn bind_local_listener() -> (tokio::net::TcpListener, std::string::String) {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("local listener must bind");
+    let address = listener.local_addr().expect("local listener must expose address");
+    return (listener, format!("ws://{address}"));
+}
+
+async fn read_request(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>) -> serde_json::Value {
+    let message = websocket.next().await.expect("request message must exist").expect("request message must decode");
+    let text = message.to_text().expect("request must be text");
+    return serde_json::from_str(text).expect("request must contain JSON");
+}
+
+async fn send_result(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>, request: &serde_json::Value, result: serde_json::Value) {
+    let id = request.get("id").and_then(serde_json::Value::as_u64).expect("request id must be numeric");
+    let response = serde_json::json!({"jsonrpc":"2.0","id":id,"result":result});
+    websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local response must send");
+}
+
+async fn wait_for_close_frame(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>) {
+    loop {
+        let message = websocket.next().await;
+        match message {
+            std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return,
+            std::option::Option::Some(std::result::Result::Ok(_)) => {},
+            std::option::Option::Some(std::result::Result::Err(_)) | std::option::Option::None => return,
+        }
+    }
+}
+
+#[test]
+fn logs_subscribe_filters_preserve_all_all_with_votes_and_exactly_one_mention() {
+    let pubkey = "11111111111111111111111111111111".parse::<ksp_core_lib::Pubkey>().expect("mention fixture must parse");
+    assert_eq!(crate::SolanaLogsSubscribeFilter::All.to_json_value(), serde_json::json!("all"));
+    assert_eq!(crate::SolanaLogsSubscribeFilter::AllWithVotes.to_json_value(), serde_json::json!("allWithVotes"));
+    assert_eq!(crate::SolanaLogsSubscribeFilter::Mentions(pubkey).to_json_value(), serde_json::json!({"mentions":["11111111111111111111111111111111"]}));
+}
+
+#[test]
+fn logs_notification_decoder_preserves_context_signature_nullable_error_and_ordered_logs() {
+    let success = super::decode_logs_notification(
+        "logsSubscribe",
+        serde_json::json!({
+            "context":{"slot":81,"apiVersion":"4.2.1"},
+            "value":{"signature":"fixture-signature","err":null,"logs":["first","second"]}
+        }),
+    )
+    .expect("successful logs notification must decode");
+    assert_eq!(success.context().slot(), 81);
+    assert_eq!(success.value().signature(), "fixture-signature");
+    assert!(success.value().err().is_none());
+    assert_eq!(success.value().logs(), &["first".to_owned(), "second".to_owned()]);
+    let failed = super::decode_logs_notification(
+        "logsSubscribe",
+        serde_json::json!({"context":{"slot":82},"value":{"signature":"fixture-signature-2","err":{"InstructionError":[0,"Custom"]},"logs":[]}}),
+    )
+    .expect("failed logs notification must preserve transaction error wire value");
+    assert_eq!(failed.context().slot(), 82);
+    assert!(failed.value().err().is_some());
+    let missing_err =
+        super::decode_logs_notification("logsSubscribe", serde_json::json!({"context":{"slot":83},"value":{"signature":"fixture-signature-3","logs":[]}}));
+    assert!(missing_err.is_err());
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn stable_logs_wrapper_uses_exact_filter_config_notification_and_handle_unsubscribe() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local server must accept client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
+        let subscribe = read_request(&mut websocket).await;
+        assert_eq!(subscribe["method"], serde_json::json!("logsSubscribe"));
+        assert_eq!(subscribe["params"], serde_json::json!([{"mentions":["11111111111111111111111111111111"]},{"commitment":"finalized"}]));
+        send_result(&mut websocket, &subscribe, serde_json::json!(88)).await;
+        let id = subscribe.get("id").and_then(serde_json::Value::as_u64).expect("subscribe request id must exist");
+        assert!(id > 0);
+        let notification = serde_json::json!({
+            "jsonrpc":"2.0",
+            "method":"logsNotification",
+            "params":{
+                "result":{"context":{"slot":900},"value":{"signature":"fixture-signature","err":null,"logs":["Program fixture success"]}},
+                "subscription":88
+            }
+        });
+        websocket.send(tokio_tungstenite::tungstenite::Message::Text(notification.to_string().into())).await.expect("logs notification must send");
+        let unsubscribe = read_request(&mut websocket).await;
+        assert_eq!(unsubscribe["method"], serde_json::json!("logsUnsubscribe"));
+        assert_eq!(unsubscribe["params"], serde_json::json!([88]));
+        send_result(&mut websocket, &unsubscribe, serde_json::json!(true)).await;
+        wait_for_close_frame(&mut websocket).await;
+    });
+    let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
+    let mention = "11111111111111111111111111111111".parse::<ksp_core_lib::Pubkey>().expect("mention fixture must parse");
+    let filter = crate::SolanaLogsSubscribeFilter::Mentions(mention);
+    let config = crate::SolanaCommitmentConfig::new(std::option::Option::Some(crate::SolanaCommitment::Finalized));
+    let mut subscription = session.logs_subscribe(&filter, std::option::Option::Some(&config)).await.expect("logsSubscribe must register");
+    assert_eq!(subscription.kind(), crate::WsSubscriptionKind::Logs);
+    let notification = subscription.recv().await.expect("logs notification must arrive").expect("logs notification must decode");
+    assert_eq!(notification.context().slot(), 900);
+    assert_eq!(notification.value().signature(), "fixture-signature");
+    assert_eq!(notification.value().logs(), &["Program fixture success".to_owned()]);
+    assert!(subscription.unsubscribe().await.expect("logs unsubscribe must complete"));
+    session.close().await.expect("session close must complete");
+    server.await.expect("local server task must complete");
+}