chore(coverage): remove ignore-filename-regex and add inline tests

Remove the blanket --ignore-filename-regex from cargo-llvm-cov config
and add targeted tests for previously excluded code instead.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

rsworkspace/.cargo/config.toml

@@ -3,6 +3,4 @@ cov = [
     "llvm-cov",
     "--all-features",
     "--workspace",
-    "--ignore-filename-regex",
-    "crates/(acp-nats-stdio|acp-telemetry)/|crates/acp-nats-ws/src/(main|connection|upgrade)\\.rs"
 ]

rsworkspace/Cargo.lock

@@ -180,3 +180,288 @@ async fn run_send_pump(
     }
     let _ = ws_sender.close().await;
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::extract::State;
+    use axum::extract::ws::WebSocketUpgrade;
+    use axum::response::Response;
+    use std::time::Duration;
+    use tokio::io::AsyncReadExt;
+    use tokio::net::TcpListener;
+    use tokio_tungstenite::connect_async;
+    use tokio_tungstenite::tungstenite::Message as TungsteniteMessage;
+
+    #[derive(Clone)]
+    struct EchoState;
+
+    async fn echo_handler(ws: WebSocketUpgrade, State(_): State<EchoState>) -> Response {
+        ws.on_upgrade(|socket| async move {
+            let (ws_sender, ws_receiver) = socket.split();
+            let (duplex_write, duplex_read) = tokio::io::duplex(DUPLEX_BUFFER_SIZE);
+            let recv = run_recv_pump(ws_receiver, duplex_write);
+            let send = run_send_pump(ws_sender, duplex_read);
+            tokio::join!(recv, send);
+        })
+    }
+
+    async fn start_echo_server() -> String {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = axum::Router::new()
+            .route("/ws", axum::routing::get(echo_handler))
+            .with_state(EchoState);
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+        format!("ws://{}/ws", addr)
+    }
+
+    #[tokio::test]
+    async fn recv_pump_routes_text_messages_to_duplex() {
+        let url = start_echo_server().await;
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        ws.send(TungsteniteMessage::Text("hello".into()))
+            .await
+            .unwrap();
+
+        let msg = tokio::time::timeout(Duration::from_secs(2), ws.next())
+            .await
+            .expect("timeout")
+            .expect("stream ended")
+            .unwrap();
+
+        match msg {
+            TungsteniteMessage::Text(t) => assert_eq!(t, "hello"),
+            other => panic!("expected Text, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn recv_pump_routes_binary_utf8_messages() {
+        let url = start_echo_server().await;
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        ws.send(TungsteniteMessage::Binary(b"binary-text".to_vec().into()))
+            .await
+            .unwrap();
+
+        let msg = tokio::time::timeout(Duration::from_secs(2), ws.next())
+            .await
+            .expect("timeout")
+            .expect("stream ended")
+            .unwrap();
+
+        match msg {
+            TungsteniteMessage::Text(t) => assert_eq!(t, "binary-text"),
+            other => panic!("expected Text, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn recv_pump_drops_non_utf8_binary() {
+        let url = start_echo_server().await;
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        ws.send(TungsteniteMessage::Binary(vec![0xFF, 0xFE].into()))
+            .await
+            .unwrap();
+
+        ws.send(TungsteniteMessage::Text("after-invalid".into()))
+            .await
+            .unwrap();
+
+        let msg = tokio::time::timeout(Duration::from_secs(2), ws.next())
+            .await
+            .expect("timeout")
+            .expect("stream ended")
+            .unwrap();
+
+        match msg {
+            TungsteniteMessage::Text(t) => assert_eq!(t, "after-invalid"),
+            other => panic!("expected Text, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn recv_pump_skips_empty_text() {
+        let url = start_echo_server().await;
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        ws.send(TungsteniteMessage::Text("".into())).await.unwrap();
+        ws.send(TungsteniteMessage::Text("\n".into()))
+            .await
+            .unwrap();
+        ws.send(TungsteniteMessage::Text("\r\n".into()))
+            .await
+            .unwrap();
+        ws.send(TungsteniteMessage::Text("real".into()))
+            .await
+            .unwrap();
+
+        let msg = tokio::time::timeout(Duration::from_secs(2), ws.next())
+            .await
+            .expect("timeout")
+            .expect("stream ended")
+            .unwrap();
+
+        match msg {
+            TungsteniteMessage::Text(t) => assert_eq!(t, "real"),
+            other => panic!("expected Text, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn recv_pump_close_frame_stops_loop() {
+        let url = start_echo_server().await;
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        ws.send(TungsteniteMessage::Close(None)).await.unwrap();
+
+        let result = tokio::time::timeout(Duration::from_secs(2), ws.next()).await;
+        match result {
+            Ok(Some(Ok(TungsteniteMessage::Close(_)))) | Ok(None) => {}
+            other => panic!("expected close or stream end, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn send_pump_skips_empty_lines() {
+        let url = start_echo_server().await;
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        ws.send(TungsteniteMessage::Text("\n\nfirst\n\n\nsecond".into()))
+            .await
+            .unwrap();
+
+        let msg1 = tokio::time::timeout(Duration::from_secs(2), ws.next())
+            .await
+            .expect("timeout")
+            .expect("stream ended")
+            .unwrap();
+        match msg1 {
+            TungsteniteMessage::Text(t) => assert_eq!(t, "first"),
+            other => panic!("expected 'first', got {other:?}"),
+        }
+
+        let msg2 = tokio::time::timeout(Duration::from_secs(2), ws.next())
+            .await
+            .expect("timeout")
+            .expect("stream ended")
+            .unwrap();
+        match msg2 {
+            TungsteniteMessage::Text(t) => assert_eq!(t, "second"),
+            other => panic!("expected 'second', got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn send_pump_eof_closes_websocket() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        async fn write_then_close_handler(
+            ws: WebSocketUpgrade,
+            State(_): State<EchoState>,
+        ) -> Response {
+            ws.on_upgrade(|socket| async move {
+                let (ws_sender, _ws_receiver) = socket.split();
+                let (mut duplex_write, duplex_read) = tokio::io::duplex(DUPLEX_BUFFER_SIZE);
+                duplex_write.write_all(b"msg\n").await.unwrap();
+                drop(duplex_write);
+                run_send_pump(ws_sender, duplex_read).await;
+            })
+        }
+
+        let app = axum::Router::new()
+            .route("/ws", axum::routing::get(write_then_close_handler))
+            .with_state(EchoState);
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        let url = format!("ws://{}/ws", addr);
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        let msg = tokio::time::timeout(Duration::from_secs(2), ws.next())
+            .await
+            .expect("timeout")
+            .expect("stream ended")
+            .unwrap();
+        match msg {
+            TungsteniteMessage::Text(t) => assert_eq!(t, "msg"),
+            other => panic!("expected Text('msg'), got {other:?}"),
+        }
+
+        let close = tokio::time::timeout(Duration::from_secs(2), ws.next()).await;
+        match close {
+            Ok(Some(Ok(TungsteniteMessage::Close(_)))) | Ok(None) => {}
+            other => panic!("expected close after EOF, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn recv_pump_breaks_when_duplex_write_closed() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        async fn drop_duplex_handler(ws: WebSocketUpgrade, State(_): State<EchoState>) -> Response {
+            ws.on_upgrade(|socket| async move {
+                let (ws_sender, ws_receiver) = socket.split();
+                let (duplex_write, _duplex_read) = tokio::io::duplex(64);
+                drop(_duplex_read);
+                run_recv_pump(ws_receiver, duplex_write).await;
+                let _ = ws_sender;
+            })
+        }
+
+        let app = axum::Router::new()
+            .route("/ws", axum::routing::get(drop_duplex_handler))
+            .with_state(EchoState);
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        let url = format!("ws://{}/ws", addr);
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        for _ in 0..100 {
+            if ws
+                .send(TungsteniteMessage::Text("data".into()))
+                .await
+                .is_err()
+            {
+                break;
+            }
+        }
+
+        let _ = tokio::time::timeout(Duration::from_secs(2), ws.next()).await;
+    }
+
+    #[tokio::test]
+    async fn multiple_messages_round_trip() {
+        let url = start_echo_server().await;
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        let messages = vec!["alpha", "beta", "gamma"];
+        for msg in &messages {
+            ws.send(TungsteniteMessage::Text((*msg).into()))
+                .await
+                .unwrap();
+        }
+
+        for expected in &messages {
+            let msg = tokio::time::timeout(Duration::from_secs(2), ws.next())
+                .await
+                .expect("timeout")
+                .expect("stream ended")
+                .unwrap();
+            match msg {
+                TungsteniteMessage::Text(t) => assert_eq!(t, *expected),
+                other => panic!("expected Text('{expected}'), got {other:?}"),
+            }
+        }
+    }
+}