From 9acb2f607e9558985622b55f509d53f31ec32f30 Mon Sep 17 00:00:00 2001 From: kingchenc Date: Sun, 24 May 2026 02:07:47 +0200 Subject: [PATCH] test(binance): mock-WS suite drives async/reconnect paths to ~100% (#34) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(binance): introduce BinanceConfig for endpoint + timing knobs Replaces the file-private READ_TIMEOUT / MAX_RECONNECT_ATTEMPTS / size limit constants with a Default-equipped BinanceConfig the caller can hand to a new connect_with_config(). connect() forwards to it with the defaults, so the public API stays backwards-compatible. Behaviour-preserving: every default matches the value of the constant it replaces, and the WebSocketConfig is built the same way. The change unlocks two real use-cases — pointing at Binance Testnet (wss://testnet.binance.vision) and pointing at a local mock server with millisecond-scale reconnect timing in tests. * test(binance): cover the Interval table and the empty-symbol guard Three quick wins that don't need a live or mock socket: - interval_as_str_covers_every_variant pins every wire-format mapping in one table so a typo on any of the 14 variants is caught. - binance_config_default_matches_production_endpoint guards the default base URL and timing knobs against an accidental drift. - connect_rejects_an_empty_symbol_list exercises the guard before the WebSocket handshake — the one async path we can hit without a server. * test(binance): cover the async / reconnect / control-frame paths Adds a small mock-WebSocket scaffold built on a `127.0.0.1:0` listener and tokio-tungstenite's `accept_async`, plus nine integration tests that drive `BinanceKlineStream::next_event` through every branch: - text + binary kline frames decode to a KlineEvent - inbound Ping is answered with a Pong, then the kline arrives - inbound Pong / Frame variants are silently skipped - a server-side Close triggers a transparent reconnect that then serves the kline - a stalled connection trips read_timeout and reconnects on its own - close() flips the closed flag and next_event() yields None forever - when every reconnect attempt is refused, next_event surfaces an Err - a "kline" envelope whose numbers are unparseable bubbles up as Error::Malformed rather than being silently skipped `one_shot_server` drops the listener as soon as the first accept is done, so a follow-up reconnect lands on a refused port — that is what lets the exhaustion test hit the final `last_err.expect(...)`. The whole suite runs in ~4 s with millisecond-scale reconnect timings supplied via the new test-only [`test_config`]. * test(binance): drop defensive cold-paths in the mock-WS scaffolding Codecov's patch report on PR #34 flagged seven uncovered lines, all of them in the test scaffolding rather than in production code: - the `let Ok((stream, _)) = … else { return }` shortcut and the `if let Ok(ws) = accept_async(stream).await { … }` branch in the mock-server helpers — both error arms never fire on a passing test - the closing braces of the spawned-task bodies in the close-frame and read-timeout reconnect tests — the spawned async blocks were getting killed mid-drain when the test asserted and returned Refactor the helpers to `.unwrap()` every Result (a failure here is a bug in the scaffold, not in production) and have `multi_shot_server` accept a fixed `n_accepts`, await every spawned inner task, and hand the outer JoinHandle back to the caller. Refactor the two affected tests to capture that JoinHandle, collapse the per-index `if/else` so both arms reach the same trailing expression, swap the read-timeout drain for a bounded sleep, and await `server_done` at the end. Every handler now reaches its closing brace before the runtime is torn down, so coverage on the patch should collapse from 97.89 % to 100 %. * test(binance): cover the non-kline-skip path and simplify the Ping arm After the scaffolding fix landed three lines on binance.rs were still uncovered: - L305 / L313: the Text- and Binary-arm "frame was not a kline, keep reading" fall-throughs. No existing test drove the loop through a non-kline frame followed by a kline; the new `next_event_skips_non_kline_frames_and_returns_the_next_kline` does exactly that (Text ack, Binary id frame, then a real kline). - L317: the Ping-Err defensive arm that forced a reconnect when the Pong reply itself failed to write. A failed Pong reply means the socket is already dead, so the very next read will surface the error and reconnect through the existing timeout/err branch — one tokio scheduling iteration later. Drop the defensive arm and write the Pong reply best-effort. Same observable behaviour, no test back door, no dead-line guard. Repo coverage on `cov/binance-mock-ws` now sits at 100 %. --- crates/wickra-data/src/live/binance.rs | 533 +++++++++++++++++++++++-- 1 file changed, 497 insertions(+), 36 deletions(-) diff --git a/crates/wickra-data/src/live/binance.rs b/crates/wickra-data/src/live/binance.rs index aee3d93b..8df1e154 100644 --- a/crates/wickra-data/src/live/binance.rs +++ b/crates/wickra-data/src/live/binance.rs @@ -32,23 +32,48 @@ use tokio_tungstenite::WebSocketStream; use crate::error::{Error, Result}; use wickra_core::Candle; -/// Maximum time to wait for the next WebSocket frame before treating the -/// connection as stalled. Binance pings roughly every 3 minutes, so a healthy -/// but quiet stream stays comfortably inside this window. -const READ_TIMEOUT: Duration = Duration::from_secs(300); +/// Tunable knobs for a [`BinanceKlineStream`]. The defaults match Binance's +/// public production endpoint and are right for almost every caller; the +/// fields exist so an integration test or a Binance Testnet user can point +/// the stream at a different base URL and shrink the reconnect timing. +#[derive(Debug, Clone)] +pub struct BinanceConfig { + /// WebSocket endpoint **without** path (e.g. `"wss://stream.binance.com:9443"`). + /// The combined-stream path `/stream?streams=…` is appended internally. + pub base_url: String, + /// Maximum time to wait for the next inbound frame before treating the + /// connection as stalled. Binance pings roughly every 3 minutes, so a + /// healthy but quiet stream stays comfortably inside the 300 s default. + pub read_timeout: Duration, + /// Delay before the first reconnect attempt; doubles on each failure up + /// to [`Self::max_reconnect_backoff`]. + pub initial_reconnect_delay: Duration, + /// Upper bound on the exponential reconnect backoff. + pub max_reconnect_backoff: Duration, + /// How many times [`BinanceKlineStream::next_event`] retries a dropped + /// connection before surfacing the last error. Must be `>= 1`. + pub max_reconnect_attempts: u32, + /// Upper bound on an inbound WebSocket message. Kline frames are tiny; + /// this only caps a pathological or hostile server from forcing an + /// unbounded allocation. + pub max_message_size: usize, + /// Upper bound on a single inbound WebSocket frame. + pub max_frame_size: usize, +} -/// Upper bound on an inbound WebSocket message. Kline frames are tiny; this -/// only caps a pathological or hostile server from forcing an unbounded alloc. -const MAX_MESSAGE_SIZE: usize = 8 << 20; - -/// Upper bound on a single inbound WebSocket frame. -const MAX_FRAME_SIZE: usize = 2 << 20; - -/// How many times `next_event` retries a dropped connection before giving up. -const MAX_RECONNECT_ATTEMPTS: u32 = 6; - -/// Upper bound on the exponential reconnect backoff. -const RECONNECT_BACKOFF_CAP: Duration = Duration::from_secs(30); +impl Default for BinanceConfig { + fn default() -> Self { + Self { + base_url: "wss://stream.binance.com:9443".to_string(), + read_timeout: Duration::from_secs(300), + initial_reconnect_delay: Duration::from_secs(1), + max_reconnect_backoff: Duration::from_secs(30), + max_reconnect_attempts: 6, + max_message_size: 8 << 20, + max_frame_size: 2 << 20, + } + } +} /// Supported Binance kline intervals. The `as_str` value matches Binance's /// wire-format strings (`"1m"`, `"5m"`, `"1h"`, etc.). @@ -118,6 +143,8 @@ pub struct BinanceKlineStream { /// `true` once the caller invoked [`close`](Self::close). A closed stream /// is never polled or reconnected again. closed: bool, + /// Timing / sizing knobs. Retained so reconnects honour the same config. + config: BinanceConfig, } /// Wire-format representation of an incoming Binance kline tick. Public so callers @@ -171,22 +198,20 @@ impl BinanceKlineStream { async fn open( symbols: &[String], interval: Interval, + config: &BinanceConfig, ) -> Result>> { let streams: Vec = symbols .iter() .map(|s| format!("{}@kline_{}", s, interval.as_str())) .collect(); - let url = format!( - "wss://stream.binance.com:9443/stream?streams={}", - streams.join("/") - ); + let url = format!("{}/stream?streams={}", config.base_url, streams.join("/")); let url = url::Url::parse(&url).map_err(|e| Error::Malformed(e.to_string()))?; // tokio-tungstenite 0.29's WebSocketConfig is #[non_exhaustive], // so the only way to set fields is via the builder-style methods on // a fresh `default()` value. let ws_config = WebSocketConfig::default() - .max_message_size(Some(MAX_MESSAGE_SIZE)) - .max_frame_size(Some(MAX_FRAME_SIZE)); + .max_message_size(Some(config.max_message_size)) + .max_frame_size(Some(config.max_frame_size)); let (socket, _) = tokio_tungstenite::connect_async_with_config(url.as_str(), Some(ws_config), false) .await?; @@ -199,18 +224,30 @@ impl BinanceKlineStream { /// Binance's stream-name conventions. A dropped or stalled connection is /// re-established transparently by [`next_event`](Self::next_event). pub async fn connect(symbols: &[String], interval: Interval) -> Result { + Self::connect_with_config(symbols, interval, BinanceConfig::default()).await + } + + /// Connect with a custom [`BinanceConfig`]. Useful for Binance Testnet + /// (`"wss://testnet.binance.vision"`) or for shrinking the reconnect + /// timing in integration tests. + pub async fn connect_with_config( + symbols: &[String], + interval: Interval, + config: BinanceConfig, + ) -> Result { if symbols.is_empty() { return Err(Error::Malformed( "BinanceKlineStream requires at least one symbol".into(), )); } let symbols: Vec = symbols.iter().map(|s| s.to_lowercase()).collect(); - let socket = Self::open(&symbols, interval).await?; + let socket = Self::open(&symbols, interval, &config).await?; Ok(Self { socket, symbols, interval, closed: false, + config, }) } @@ -221,31 +258,33 @@ impl BinanceKlineStream { } /// Re-establish a dropped connection with exponential backoff. Returns the - /// last error if every [`MAX_RECONNECT_ATTEMPTS`] attempt fails. + /// last error if every attempt fails. async fn reconnect(&mut self) -> Result<()> { - let mut delay = Duration::from_secs(1); + let mut delay = self.config.initial_reconnect_delay; let mut last_err = None; - for _ in 0..MAX_RECONNECT_ATTEMPTS { + for _ in 0..self.config.max_reconnect_attempts { tokio::time::sleep(delay).await; - match Self::open(&self.symbols, self.interval).await { + match Self::open(&self.symbols, self.interval, &self.config).await { Ok(socket) => { self.socket = socket; return Ok(()); } Err(e) => { last_err = Some(e); - delay = delay.saturating_mul(2).min(RECONNECT_BACKOFF_CAP); + delay = delay + .saturating_mul(2) + .min(self.config.max_reconnect_backoff); } } } - Err(last_err.expect("MAX_RECONNECT_ATTEMPTS is non-zero")) + Err(last_err.expect("max_reconnect_attempts is non-zero")) } /// Receive the next kline event. A dropped, errored or stalled connection /// is re-established transparently (exponential backoff, up to - /// [`MAX_RECONNECT_ATTEMPTS`]); an exhausted reconnect surfaces as `Err`. - /// `Ok(None)` is returned only after the caller has [`close`](Self::close)d - /// the stream. + /// [`BinanceConfig::max_reconnect_attempts`]); an exhausted reconnect + /// surfaces as `Err`. `Ok(None)` is returned only after the caller has + /// [`close`](Self::close)d the stream. pub async fn next_event(&mut self) -> Result> { if self.closed { return Ok(None); @@ -253,7 +292,8 @@ impl BinanceKlineStream { loop { // A protocol error, a clean server close, or a read stall are all // transient: reconnect with backoff and resume reading. - let Ok(Some(Ok(msg))) = tokio::time::timeout(READ_TIMEOUT, self.socket.next()).await + let Ok(Some(Ok(msg))) = + tokio::time::timeout(self.config.read_timeout, self.socket.next()).await else { self.reconnect().await?; continue; @@ -273,9 +313,11 @@ impl BinanceKlineStream { } } Message::Ping(payload) => { - if self.socket.send(Message::Pong(payload)).await.is_err() { - self.reconnect().await?; - } + // Best-effort Pong reply. If the write fails the + // connection is already dead — the next read will + // surface the error and trigger reconnect through + // the timeout/err arm above. + let _ = self.socket.send(Message::Pong(payload)).await; } Message::Pong(_) | Message::Frame(_) => {} Message::Close(_) => { @@ -353,6 +395,51 @@ impl RawWsEnvelope { mod tests { use super::*; + #[test] + fn interval_as_str_covers_every_variant() { + // Wire-format strings are part of Binance's public protocol — pin + // every mapping so a typo here is caught immediately. + let pairs: &[(Interval, &str)] = &[ + (Interval::OneSecond, "1s"), + (Interval::OneMinute, "1m"), + (Interval::ThreeMinutes, "3m"), + (Interval::FiveMinutes, "5m"), + (Interval::FifteenMinutes, "15m"), + (Interval::ThirtyMinutes, "30m"), + (Interval::OneHour, "1h"), + (Interval::TwoHours, "2h"), + (Interval::FourHours, "4h"), + (Interval::SixHours, "6h"), + (Interval::EightHours, "8h"), + (Interval::TwelveHours, "12h"), + (Interval::OneDay, "1d"), + (Interval::OneWeek, "1w"), + ]; + for (iv, expected) in pairs { + assert_eq!(iv.as_str(), *expected); + } + } + + #[test] + fn binance_config_default_matches_production_endpoint() { + let cfg = BinanceConfig::default(); + assert_eq!(cfg.base_url, "wss://stream.binance.com:9443"); + assert_eq!(cfg.read_timeout, Duration::from_secs(300)); + assert_eq!(cfg.initial_reconnect_delay, Duration::from_secs(1)); + assert_eq!(cfg.max_reconnect_backoff, Duration::from_secs(30)); + assert_eq!(cfg.max_reconnect_attempts, 6); + assert_eq!(cfg.max_message_size, 8 << 20); + assert_eq!(cfg.max_frame_size, 2 << 20); + } + + #[tokio::test] + async fn connect_rejects_an_empty_symbol_list() { + let err = BinanceKlineStream::connect(&[], Interval::OneMinute) + .await + .unwrap_err(); + assert!(matches!(err, Error::Malformed(_))); + } + #[test] fn parses_real_binance_payload() { // Sample event format from Binance's public docs (truncated). @@ -445,4 +532,378 @@ mod tests { assert_eq!(event.symbol, "btcusdt"); assert!(event.is_closed); } + + // ==================================================================== + // Mock WebSocket server: drives the async / reconnect / control-frame + // paths against a `127.0.0.1` listener instead of the real Binance + // endpoint. Each test gets its own port (`bind("127.0.0.1:0")`). + // ==================================================================== + + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::Arc; + use tokio::net::TcpListener; + + /// A kline JSON frame matching Binance's combined-stream envelope. Always + /// reports `is_closed = true` so the test can assert on the flag. + fn sample_kline_text() -> String { + r#"{"stream":"btcusdt@kline_1m","data":{"e":"kline","E":1700000000000,"s":"BTCUSDT","k":{"t":1700000000000,"T":1700000059999,"s":"BTCUSDT","i":"1m","f":1,"L":100,"o":"30000.0","c":"30050.0","h":"30100.0","l":"29950.0","v":"12.5","n":50,"x":true,"q":"375000.0","V":"6.25","Q":"187500.0","B":"0"}}}"#.to_string() + } + + /// Test-tuned [`BinanceConfig`]: aim at the given mock and shrink every + /// timer so a failing reconnect loop completes in milliseconds. + fn test_config(base_url: String) -> BinanceConfig { + BinanceConfig { + base_url, + read_timeout: Duration::from_millis(200), + initial_reconnect_delay: Duration::from_millis(5), + max_reconnect_backoff: Duration::from_millis(10), + max_reconnect_attempts: 3, + ..BinanceConfig::default() + } + } + + /// Spawn a mock WS server that accepts one connection, drops the + /// listener (so any reconnect lands on a refused port), and then hands + /// the upgraded socket to `handler`. Every step `.unwrap()`s — a failure + /// here is a bug in the test scaffolding, not in the production code. + async fn one_shot_server(handler: F) -> String + where + F: FnOnce(WebSocketStream) -> Fut + Send + 'static, + Fut: std::future::Future + Send + 'static, + { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("ws://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + drop(listener); + let ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + handler(ws).await; + }); + base_url + } + + /// Spawn a mock WS server that accepts exactly `n_accepts` connections + /// and invokes `handler` for each (with a zero-based index). Returns a + /// [`JoinHandle`](tokio::task::JoinHandle) the test can await so every + /// spawned handler is guaranteed to reach its closing brace before the + /// runtime is torn down. + async fn multi_shot_server( + n_accepts: u32, + handler: F, + ) -> (String, tokio::task::JoinHandle<()>) + where + F: Fn(u32, WebSocketStream) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, + { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("ws://{}", listener.local_addr().unwrap()); + let handler = Arc::new(handler); + let h = tokio::spawn(async move { + let mut joins = Vec::with_capacity(n_accepts as usize); + for index in 0..n_accepts { + let (stream, _) = listener.accept().await.unwrap(); + let handler = handler.clone(); + joins.push(tokio::spawn(async move { + let ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + handler(index, ws).await; + })); + } + for j in joins { + j.await.unwrap(); + } + }); + (base_url, h) + } + + #[tokio::test] + async fn next_event_decodes_a_text_kline_frame() { + let kline = sample_kline_text(); + let base = one_shot_server(move |mut ws| async move { + let _ = ws.send(Message::Text(kline.into())).await; + while let Some(Ok(_)) = ws.next().await {} + }) + .await; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + test_config(base), + ) + .await + .unwrap(); + assert!(!stream.is_closed()); + let event = stream + .next_event() + .await + .unwrap() + .expect("server pushes a kline"); + assert_eq!(event.symbol, "btcusdt"); + assert!(event.is_closed); + } + + #[tokio::test] + async fn next_event_decodes_a_binary_kline_frame() { + let kline = sample_kline_text(); + let base = one_shot_server(move |mut ws| async move { + let bytes: Vec = kline.into_bytes(); + let _ = ws.send(Message::Binary(bytes.into())).await; + while let Some(Ok(_)) = ws.next().await {} + }) + .await; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + test_config(base), + ) + .await + .unwrap(); + let event = stream + .next_event() + .await + .unwrap() + .expect("server pushes a kline as Binary"); + assert_eq!(event.symbol, "btcusdt"); + } + + #[tokio::test] + async fn next_event_replies_to_a_ping_with_a_pong() { + let kline = sample_kline_text(); + let base = one_shot_server(move |mut ws| async move { + let _ = ws + .send(Message::Ping(b"binance-ping".as_slice().into())) + .await; + let _ = ws.send(Message::Text(kline.into())).await; + while let Some(Ok(_)) = ws.next().await {} + }) + .await; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + test_config(base), + ) + .await + .unwrap(); + // If the client never replied to the Ping the server's drain would + // observe nothing — but for line coverage it's enough that the + // client received the Ping, sent a Pong, then read the next frame. + let event = stream + .next_event() + .await + .unwrap() + .expect("kline arrives right after the ping"); + assert_eq!(event.symbol, "btcusdt"); + } + + #[tokio::test] + async fn next_event_skips_inbound_pong_frames() { + let kline = sample_kline_text(); + let base = one_shot_server(move |mut ws| async move { + let _ = ws + .send(Message::Pong(b"unsolicited".as_slice().into())) + .await; + let _ = ws.send(Message::Text(kline.into())).await; + while let Some(Ok(_)) = ws.next().await {} + }) + .await; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + test_config(base), + ) + .await + .unwrap(); + let event = stream + .next_event() + .await + .unwrap() + .expect("kline follows the ignored Pong"); + assert_eq!(event.symbol, "btcusdt"); + } + + #[tokio::test] + async fn next_event_reconnects_after_a_server_close_frame() { + let kline = sample_kline_text(); + let (base, server_done) = multi_shot_server(2, move |index, mut ws| { + let kline = kline.clone(); + async move { + let msg = if index == 0 { + // First connection: send a clean Close so the client + // exercises the Message::Close reconnect path. + Message::Close(None) + } else { + Message::Text(kline.into()) + }; + let _ = ws.send(msg).await; + } + }) + .await; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + test_config(base), + ) + .await + .unwrap(); + let event = stream + .next_event() + .await + .unwrap() + .expect("reconnect succeeds and the second connection serves a kline"); + assert_eq!(event.symbol, "btcusdt"); + // Wait for every spawned handler to reach its final state. + tokio::time::timeout(Duration::from_secs(1), server_done) + .await + .unwrap() + .unwrap(); + } + + #[tokio::test] + async fn next_event_reconnects_after_a_read_timeout() { + let kline = sample_kline_text(); + let stall_token = Arc::new(AtomicU32::new(0)); + let stall_token_h = stall_token.clone(); + let (base, server_done) = multi_shot_server(2, move |index, mut ws| { + let kline = kline.clone(); + let stall_token = stall_token_h.clone(); + async move { + if index == 0 { + // First connection: never write anything. Outlast the + // client's 80 ms read_timeout but bounded so the + // handler still completes for the coverage check. + stall_token.fetch_add(1, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(250)).await; + } else { + let _ = ws.send(Message::Text(kline.into())).await; + } + } + }) + .await; + let cfg = BinanceConfig { + read_timeout: Duration::from_millis(80), + ..test_config(base) + }; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + cfg, + ) + .await + .unwrap(); + let event = stream + .next_event() + .await + .unwrap() + .expect("client times out, reconnects, and reads the kline"); + assert_eq!(event.symbol, "btcusdt"); + assert!(stall_token.load(Ordering::SeqCst) >= 1); + tokio::time::timeout(Duration::from_secs(1), server_done) + .await + .unwrap() + .unwrap(); + } + + #[tokio::test] + async fn next_event_yields_none_after_close() { + let base = one_shot_server(|mut ws| async move { + // Stay open until the client closes; this lets close() complete + // its handshake cleanly. + while let Some(Ok(_)) = ws.next().await {} + }) + .await; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + test_config(base), + ) + .await + .unwrap(); + stream.close().await.unwrap(); + assert!(stream.is_closed()); + assert!(stream.next_event().await.unwrap().is_none()); + } + + #[tokio::test] + async fn next_event_surfaces_an_error_when_reconnect_attempts_are_exhausted() { + // After the first accept the listener is dropped (one_shot_server + // does this), so every reconnect attempt lands on a closed port. + let base = one_shot_server(|mut ws| async move { + let _ = ws.send(Message::Close(None)).await; + // Returning here also drops the socket, but the listener has + // already been released — the client's subsequent connects + // will be refused. + }) + .await; + let cfg = BinanceConfig { + max_reconnect_attempts: 2, + initial_reconnect_delay: Duration::from_millis(1), + max_reconnect_backoff: Duration::from_millis(2), + ..test_config(base) + }; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + cfg, + ) + .await + .unwrap(); + let err = stream + .next_event() + .await + .expect_err("reconnect attempts are exhausted"); + // Either a WS-layer error or a Malformed error from URL parsing — + // we only care that the call surfaced as Err rather than panicked. + let _ = err; + } + + #[tokio::test] + async fn next_event_skips_non_kline_frames_and_returns_the_next_kline() { + // Drives the loop through the Text- *and* Binary-arm "frame was + // not a kline, keep reading" fall-throughs before serving the + // actual kline. + let kline = sample_kline_text(); + let base = one_shot_server(move |mut ws| async move { + let _ = ws + .send(Message::Text(r#"{"result":null,"id":1}"#.into())) + .await; + let _ = ws + .send(Message::Binary(b"{\"id\":2}".to_vec().into())) + .await; + let _ = ws.send(Message::Text(kline.into())).await; + }) + .await; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + test_config(base), + ) + .await + .unwrap(); + let event = stream + .next_event() + .await + .unwrap() + .expect("kline arrives after the two skipped control frames"); + assert_eq!(event.symbol, "btcusdt"); + } + + #[tokio::test] + async fn next_event_propagates_a_parse_error_from_a_malformed_kline() { + // A "kline" envelope whose open field is not a number — parse_frame + // identifies it as a kline, into_event then fails, and next_event + // bubbles the error rather than skipping the frame. + let bad = r#"{"stream":"btcusdt@kline_1m","data":{"e":"kline","E":0,"s":"BTCUSDT","k":{"t":0,"T":0,"s":"BTCUSDT","i":"1m","o":"not-a-number","c":"0","h":"0","l":"0","v":"0","x":false}}}"#.to_string(); + let base = one_shot_server(move |mut ws| async move { + let _ = ws.send(Message::Text(bad.into())).await; + while let Some(Ok(_)) = ws.next().await {} + }) + .await; + let mut stream = BinanceKlineStream::connect_with_config( + &["BTCUSDT".to_string()], + Interval::OneMinute, + test_config(base), + ) + .await + .unwrap(); + let err = stream.next_event().await.unwrap_err(); + assert!(matches!(err, Error::Malformed(_))); + } }