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/crates/acp-nats-ws/src/connection.rs

@@ -180,3 +180,66 @@ 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 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:?}"),
+            }
+        }
+    }
+}

rsworkspace/crates/acp-nats-ws/src/main.rs

@@ -252,4 +252,67 @@ mod tests {
 
         conn_thread.join().unwrap();
     }
+
+    #[tokio::test]
+    async fn test_shutdown_while_connection_active() {
+        let nats_mock = AdvancedMockNatsClient::new();
+        let config = Config::new(
+            acp_nats::AcpPrefix::new("acp").unwrap(),
+            acp_nats::NatsConfig {
+                servers: vec!["localhost:4222".to_string()],
+                auth: trogon_nats::NatsAuth::None,
+            },
+        );
+
+        let _injector = nats_mock.inject_messages();
+
+        nats_mock.hang_next_request();
+
+        let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
+        let (conn_tx, conn_rx) = mpsc::unbounded_channel::<ConnectionRequest>();
+
+        let nats_mock_clone = nats_mock.clone();
+        let conn_thread = std::thread::Builder::new()
+            .name(THREAD_NAME.into())
+            .spawn(move || run_connection_thread(conn_rx, nats_mock_clone, config))
+            .expect("failed to spawn connection thread");
+
+        let state = UpgradeState {
+            conn_tx,
+            shutdown_tx: shutdown_tx.clone(),
+        };
+
+        let app = axum::Router::new()
+            .route("/ws", axum::routing::get(upgrade::handle))
+            .with_state(state);
+
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        let server_task = tokio::spawn(async move {
+            axum::serve(listener, app)
+                .with_graceful_shutdown(async move {
+                    let _ = shutdown_rx.changed().await;
+                })
+                .await
+                .unwrap();
+        });
+
+        let ws_url = format!("ws://{}/ws", addr);
+        let (mut ws_stream, _) = connect_async(&ws_url).await.unwrap();
+
+        let req =
+            r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion": 0}}"#;
+        ws_stream.send(Message::Text(req.into())).await.unwrap();
+
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        shutdown_tx.send(true).unwrap();
+
+        let _ = tokio::time::timeout(Duration::from_secs(5), server_task)
+            .await
+            .expect("server task did not shut down");
+
+        conn_thread.join().unwrap();
+    }
 }

rsworkspace/crates/acp-nats-ws/src/upgrade.rs

@@ -30,3 +30,72 @@ pub async fn handle(ws: WebSocketUpgrade, State(state): State<UpgradeState>) ->
         }
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures_util::SinkExt;
+    use std::time::Duration;
+    use tokio::net::TcpListener;
+    use tokio_tungstenite::connect_async;
+    use tokio_tungstenite::tungstenite::Message;
+
+    #[tokio::test]
+    async fn handle_sends_connection_request_through_channel() {
+        let (shutdown_tx, _shutdown_rx) = watch::channel(false);
+        let (conn_tx, mut conn_rx) = mpsc::unbounded_channel::<ConnectionRequest>();
+
+        let state = UpgradeState {
+            conn_tx,
+            shutdown_tx: shutdown_tx.clone(),
+        };
+
+        let app = axum::Router::new()
+            .route("/ws", axum::routing::get(handle))
+            .with_state(state);
+
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        let url = format!("ws://{}/ws", addr);
+        let (_ws, _) = connect_async(&url).await.unwrap();
+
+        let req = tokio::time::timeout(Duration::from_secs(2), conn_rx.recv())
+            .await
+            .expect("timeout waiting for ConnectionRequest")
+            .expect("channel closed");
+
+        assert!(!*req.shutdown_rx.borrow());
+    }
+
+    #[tokio::test]
+    async fn handle_logs_error_when_conn_rx_dropped() {
+        let (shutdown_tx, _shutdown_rx) = watch::channel(false);
+        let (conn_tx, conn_rx) = mpsc::unbounded_channel::<ConnectionRequest>();
+
+        let state = UpgradeState {
+            conn_tx,
+            shutdown_tx: shutdown_tx.clone(),
+        };
+
+        let app = axum::Router::new()
+            .route("/ws", axum::routing::get(handle))
+            .with_state(state);
+
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        drop(conn_rx);
+
+        let url = format!("ws://{}/ws", addr);
+        let (_ws, _) = connect_async(&url).await.unwrap();
+
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}

rsworkspace/crates/acp-telemetry/Cargo.toml

@@ -15,4 +15,5 @@ tracing-subscriber = { version = "0.3.22", features = ["env-filter", "fmt", "jso
 trogon-std = { path = "../trogon-std" }
 
 [dev-dependencies]
+tokio = { version = "1.49.0", features = ["rt", "macros", "signal"] }
 trogon-std = { path = "../trogon-std", features = ["test-support"] }

rsworkspace/crates/acp-telemetry/src/log.rs

@@ -81,6 +81,7 @@ fn platform_log_dir(service_name: ServiceName) -> Result<PathBuf, Box<dyn std::e
 #[cfg(test)]
 mod tests {
     use super::*;
+    use opentelemetry::KeyValue;
     use trogon_std::env::InMemoryEnv;
     use trogon_std::fs::MemFs;
 
@@ -109,4 +110,15 @@ mod tests {
         let dir = platform_log_dir(ServiceName::AcpNatsWs).unwrap();
         assert!(dir.ends_with(ServiceName::AcpNatsWs.as_str()));
     }
+
+    #[test]
+    fn init_provider_lifecycle() {
+        let resource = opentelemetry_sdk::Resource::builder()
+            .with_service_name("test-log")
+            .with_attributes(vec![KeyValue::new("test", "true")])
+            .build();
+
+        let provider = init_provider(&resource);
+        assert!(provider.is_ok());
+    }
 }

rsworkspace/crates/acp-telemetry/src/metric.rs

@@ -42,3 +42,21 @@ pub(crate) fn shutdown() {
         eprintln!("Failed to shutdown meter provider: {e}");
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use opentelemetry::KeyValue;
+    use opentelemetry_sdk::Resource;
+
+    #[test]
+    fn init_provider_returns_valid_provider() {
+        let resource = Resource::builder()
+            .with_service_name("test-metric")
+            .with_attributes(vec![KeyValue::new("test", "true")])
+            .build();
+
+        let provider = init_provider(&resource);
+        assert!(provider.is_ok());
+    }
+}

rsworkspace/crates/acp-telemetry/src/trace.rs

@@ -32,3 +32,21 @@ pub(crate) fn shutdown() {
         eprintln!("Failed to shutdown tracer provider: {e}");
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use opentelemetry::KeyValue;
+    use opentelemetry_sdk::Resource;
+
+    #[test]
+    fn init_provider_returns_valid_provider() {
+        let resource = Resource::builder()
+            .with_service_name("test-trace")
+            .with_attributes(vec![KeyValue::new("test", "true")])
+            .build();
+
+        let provider = init_provider(&resource);
+        assert!(provider.is_ok());
+    }
+}

rsworkspace/crates/acp-telemetry/tests/lifecycle.rs

@@ -0,0 +1,19 @@
+use acp_telemetry::ServiceName;
+use trogon_std::env::InMemoryEnv;
+use trogon_std::fs::MemFs;
+
+#[test]
+fn init_logger_creates_log_dir_and_shuts_down_cleanly() {
+    let env = InMemoryEnv::new();
+    env.set("ACP_LOG_DIR", "/tmp/test-coverage-lifecycle");
+    let fs = MemFs::new();
+
+    acp_telemetry::init_logger(ServiceName::AcpNatsStdio, "test", &env, &fs);
+
+    assert!(
+        fs.dir_exists(&std::path::PathBuf::from("/tmp/test-coverage-lifecycle")),
+        "init_logger should create the configured log directory"
+    );
+
+    acp_telemetry::shutdown_otel();
+}