A lot

transport/Cargo.toml

@@ -21,6 +21,9 @@ thiserror = "2.0.18"
 rustls-native-certs = "0.8.4"
 log = "0.4"
 
+[dev-dependencies]
+rcgen = "0.14"
+
 [features]
 default = []
 # Enables hosting a MTP server

transport/src/connection.rs

@@ -56,7 +56,6 @@ impl Default for Policy {
 enum ReceivedFrame {
     Message(CommunicationValue),
     ClosedByPeer,
-    #[allow(dead_code)]
     Idle,
 }
 
@@ -517,3 +516,48 @@ impl Receiver {
         self.handle.close_reason()
     }
 }
+
+/* ================================ TESTS ================================ */
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_send_mode_derive() {
+        assert_eq!(SendMode::PersistentStream, SendMode::PersistentStream);
+        assert_ne!(SendMode::PersistentStream, SendMode::SingleStreamPerMessage);
+    }
+
+    #[test]
+    fn test_policy_default_values() {
+        let p = Policy::default();
+        assert_eq!(p.send_mode, SendMode::PersistentStream);
+        assert_eq!(p.max_message_size, 1_000_000_000);
+        assert_eq!(p.close_frame_len, u32::MAX);
+        assert_eq!(p.application_close_code, 0);
+        assert_eq!(p.open_stream_timeout, Duration::from_millis(2_000));
+        assert_eq!(p.write_timeout, Duration::from_millis(2_000));
+        assert_eq!(p.accept_stream_timeout, Duration::from_millis(10_000));
+        assert_eq!(p.read_timeout, Duration::from_millis(30_000));
+        assert_eq!(p.keep_alive_interval, Some(Duration::from_secs(3)));
+        assert_eq!(p.max_idle_timeout, Some(Duration::from_secs(30)));
+        assert_eq!(p.force_close_delay, Duration::from_millis(300));
+        assert_eq!(p.max_transient_recv_errors, 20);
+        assert_eq!(p.transient_recv_backoff, Duration::from_millis(100));
+        assert_eq!(p.receiver_queue_capacity, 1000);
+    }
+
+    #[test]
+    fn test_policy_clone() {
+        let p = Policy::default();
+        let cloned = p.clone();
+        assert_eq!(p.send_mode, cloned.send_mode);
+    }
+
+    #[test]
+    fn test_policy_debug() {
+        let p = Policy::default();
+        let debug_str = format!("{:?}", p);
+        assert!(debug_str.contains("Policy"));
+    }
+}

transport/src/connection_handle.rs

@@ -63,3 +63,70 @@ impl Default for ConnectionHandle {
         Self::new()
     }
 }
+
+/* ================================ TESTS ================================ */
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_new_is_open() {
+        let h = ConnectionHandle::new();
+        assert!(h.is_open());
+        assert!(!h.is_closed());
+    }
+
+    #[test]
+    fn test_close_transitions_state() {
+        let h = ConnectionHandle::new();
+        h.close(Some(CommunicationError::StreamClosed));
+        assert!(!h.is_open());
+        assert!(h.is_closed());
+    }
+
+    #[test]
+    fn test_close_reason_some() {
+        let h = ConnectionHandle::new();
+        h.close(Some(CommunicationError::UseAfterClosed));
+        assert!(h.close_reason().is_some());
+    }
+
+    #[test]
+    fn test_close_reason_none() {
+        let h = ConnectionHandle::new();
+        h.close(None);
+        assert!(h.close_reason().is_none());
+    }
+
+    #[test]
+    fn test_default_is_new() {
+        let h = ConnectionHandle::default();
+        assert!(h.is_open());
+    }
+
+    #[test]
+    fn test_close_with_error() {
+        let h = ConnectionHandle::new();
+        h.close_with_error(CommunicationError::MessageTooLarge);
+        assert!(h.is_closed());
+        assert!(h.close_reason().is_some());
+    }
+
+    #[test]
+    fn test_multiple_close_first_wins() {
+        let h = ConnectionHandle::new();
+        h.close(Some(CommunicationError::StreamClosed));
+        h.close(Some(CommunicationError::UseAfterClosed));
+        // First close reason is preserved
+        assert!(h.close_reason().is_some());
+    }
+
+    #[test]
+    fn test_close_sends_reason() {
+        let h = ConnectionHandle::new();
+        let mut rx = h.subscribe_close();
+        h.close(Some(CommunicationError::ClosedLocally));
+        // After close, the watch channel is updated
+        assert!(rx.borrow_and_update().is_some());
+    }
+}

transport/src/host.rs

@@ -1,4 +1,5 @@
 use crate::{ConnectionHandle, Policy, Receiver, Sender};
+use log;
 use mtp_common::CommunicationError;
 use rustls::pki_types::{PrivateKeyDer, pem::PemObject};
 use std::net::{IpAddr, SocketAddr};
@@ -13,7 +14,13 @@ pub struct Host {
 
 impl Host {
     pub async fn next(&mut self) -> Option<(Sender, Receiver)> {
-        self.incoming.recv().await
+        log::warn!("[transport Host::next] waiting on recv...");
+        let result = self.incoming.recv().await;
+        match &result {
+            Some(_) => log::warn!("[transport Host::next] received connection"),
+            None => log::warn!("[transport Host::next] incoming channel closed - sender dropped"),
+        }
+        result
     }
 
     pub fn local_addr(&self) -> std::net::SocketAddr {
@@ -22,6 +29,7 @@ impl Host {
 }
 
 pub async fn host(
+    ip: IpAddr,
     port: u16,
     cert_pem: Vec<u8>,
     key_pem: Vec<u8>,
@@ -29,7 +37,7 @@ pub async fn host(
 ) -> Result<Host, CommunicationError> {
     let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
 
-    let server_config = configure_server(port, cert_pem, key_pem, &policy).await?;
+    let server_config = configure_server(ip, port, cert_pem, key_pem, &policy).await?;
     let endpoint = Endpoint::server(server_config)
         .map_err(|e| CommunicationError::Other(format!("Endpoint creation failed: {}", e)))?;
 
@@ -42,25 +50,37 @@ pub async fn host(
     let policy = Arc::new(policy);
 
     let task = tokio::spawn(async move {
+        log::warn!("[transport bg task] started");
         loop {
+            log::warn!("[transport bg task] waiting for connection...");
             let incoming_session = endpoint.accept().await;
+            log::warn!("[transport bg task] got incoming session");
 
             let request = match incoming_session.await {
-                Ok(req) => req,
-                Err(_) => {
+                Ok(req) => {
+                    log::warn!("[transport bg task] got request");
+                    req
+                }
+                Err(e) => {
+                    log::warn!("[transport bg task] incoming session error: {e}");
                     continue;
                 }
             };
 
             let connection = match request.accept().await {
-                Ok(conn) => conn,
-                Err(_) => {
+                Ok(conn) => {
+                    log::warn!("[transport bg task] connection accepted");
+                    conn
+                }
+                Err(e) => {
+                    log::warn!("[transport bg task] accept error: {e}");
                     continue;
                 }
             };
 
             let incoming_tx = incoming_tx.clone();
             let policy = policy.clone();
+            eprintln!("[transport bg task] spawning handle_connection");
             tokio::spawn(handle_connection(connection, incoming_tx, policy));
         }
     });
@@ -85,6 +105,7 @@ async fn handle_connection(
 }
 
 async fn configure_server(
+    bind_ip: IpAddr,
     port: u16,
     cert_pem: Vec<u8>,
     key_pem: Vec<u8>,
@@ -104,12 +125,6 @@ async fn configure_server(
 
     tls_config.alpn_protocols = vec![b"h3".to_vec()];
 
-    let bind_ip = std::env::var("mtp_BIND")
-        .ok()
-        .and_then(|s| if s.is_empty() { None } else { Some(s) })
-        .unwrap_or_else(|| "::".to_string())
-        .parse::<IpAddr>()
-        .map_err(|e| CommunicationError::ParseError(e.to_string()))?;
     let bind_addr = SocketAddr::new(bind_ip, port);
 
     let server_config = ServerConfig::builder()

transport/tests/integration.rs

@@ -0,0 +1,174 @@
+use std::net::{IpAddr, Ipv4Addr};
+
+use mtp_transport::{Policy, host, connect};
+
+fn generate_self_signed_cert() -> (Vec<u8>, Vec<u8>) {
+    let key_pair = rcgen::KeyPair::generate().unwrap();
+    let params =
+        rcgen::CertificateParams::new(vec!["localhost".into(), "127.0.0.1".into()]).unwrap();
+    let cert = params.self_signed(&key_pair).unwrap();
+    let cert_pem = cert.pem();
+    let key_pem = key_pair.serialize_pem();
+    (cert_pem.into_bytes(), key_pem.into_bytes())
+}
+
+#[tokio::test]
+async fn test_host_start_and_stop() {
+    let (cert_pem, key_pem) = generate_self_signed_cert();
+    let h = host(
+        IpAddr::V4(Ipv4Addr::LOCALHOST),
+        0,
+        cert_pem,
+        key_pem,
+        Policy::default(),
+    )
+    .await
+    .unwrap();
+    let addr = h.local_addr();
+    // Port should be non-zero (OS-assigned)
+    assert!(addr.port() > 0);
+}
+
+#[tokio::test]
+async fn test_send_receive_roundtrip() {
+    let (cert_pem, key_pem) = generate_self_signed_cert();
+    let mut h = host(
+        IpAddr::V4(Ipv4Addr::LOCALHOST),
+        0,
+        cert_pem.clone(),
+        key_pem,
+        Policy::default(),
+    )
+    .await
+    .unwrap();
+    let addr = h.local_addr();
+
+    let url = format!("https://127.0.0.1:{}", addr.port());
+    let (client_tx, client_rx) =
+        connect(&url, Some(cert_pem), Policy::default()).await.unwrap();
+
+    // Accept on host side
+    let (host_tx, host_rx) = h.next().await.unwrap();
+
+    // Client sends a simple message
+    let msg = mtp_codec::CommunicationValue::new(mtp_codec::CommunicationType::Ping)
+        .add_data(
+            mtp_codec::DataTypeId(6),
+            mtp_codec::DataValue::UnsignedNumber(42),
+        );
+    client_tx.send(&msg).await.unwrap();
+
+    // Host receives it
+    let received = host_rx.receive().await.unwrap();
+    assert_eq!(received.get_type(), mtp_codec::CommunicationTypeId(19)); // Ping
+    let val = received.get_data(mtp_codec::DataTypeId(6)).clone();
+    assert_eq!(val, mtp_codec::DataValue::UnsignedNumber(42));
+
+    // Host sends a response
+    let resp = mtp_codec::CommunicationValue::new(mtp_codec::CommunicationType::Pong)
+        .add_data(
+            mtp_codec::DataTypeId(6),
+            mtp_codec::DataValue::UnsignedNumber(99),
+        );
+    host_tx.send(&resp).await.unwrap();
+
+    // Client receives it
+    let client_received = client_rx.receive().await.unwrap();
+    assert_eq!(client_received.get_type(), mtp_codec::CommunicationTypeId(20)); // Pong
+    let client_val = client_received.get_data(mtp_codec::DataTypeId(6)).clone();
+    assert_eq!(client_val, mtp_codec::DataValue::UnsignedNumber(99));
+
+    // Close both sides
+    client_tx.close();
+    host_tx.close();
+}
+
+#[tokio::test]
+async fn test_concurrent_messages() {
+    let (cert_pem, key_pem) = generate_self_signed_cert();
+    let mut h = host(
+        IpAddr::V4(Ipv4Addr::LOCALHOST),
+        0,
+        cert_pem.clone(),
+        key_pem,
+        Policy::default(),
+    )
+    .await
+    .unwrap();
+    let addr = h.local_addr();
+
+    let url = format!("https://127.0.0.1:{}", addr.port());
+    let (client_tx, _client_rx) =
+        connect(&url, Some(cert_pem), Policy::default()).await.unwrap();
+
+    let (_host_tx, host_rx) = h.next().await.unwrap();
+
+    // Send 5 messages in sequence
+    for i in 0..5u128 {
+        let msg = mtp_codec::CommunicationValue::new(mtp_codec::CommunicationType::Ping)
+            .add_data(
+                mtp_codec::DataTypeId(6),
+                mtp_codec::DataValue::UnsignedNumber(i),
+            );
+        client_tx.send(&msg).await.unwrap();
+    }
+
+    // Receive all 5 in order
+    for i in 0..5u128 {
+        let received = host_rx.receive().await.unwrap();
+        let val = received.get_data(mtp_codec::DataTypeId(6)).clone();
+        assert_eq!(val, mtp_codec::DataValue::UnsignedNumber(i));
+    }
+
+    // Send 3 responses back
+    for i in 0..3u128 {
+        let msg = mtp_codec::CommunicationValue::new(mtp_codec::CommunicationType::Pong)
+            .add_data(
+                mtp_codec::DataTypeId(6),
+                mtp_codec::DataValue::UnsignedNumber(i * 10),
+            );
+        client_tx.send(&msg).await.unwrap();
+    }
+
+    for i in 0..3u128 {
+        let received = host_rx.receive().await.unwrap();
+        let val = received.get_data(mtp_codec::DataTypeId(6)).clone();
+        assert_eq!(val, mtp_codec::DataValue::UnsignedNumber(i * 10));
+    }
+
+    client_tx.close();
+}
+
+#[tokio::test]
+async fn test_close_detection() {
+    let (cert_pem, key_pem) = generate_self_signed_cert();
+    let mut h = host(
+        IpAddr::V4(Ipv4Addr::LOCALHOST),
+        0,
+        cert_pem.clone(),
+        key_pem,
+        Policy::default(),
+    )
+    .await
+    .unwrap();
+    let addr = h.local_addr();
+
+    let url = format!("https://127.0.0.1:{}", addr.port());
+    let (client_tx, _client_rx) =
+        connect(&url, Some(cert_pem), Policy::default()).await.unwrap();
+
+    let (_host_tx, host_rx) = h.next().await.unwrap();
+
+    // Send a message then close
+    let msg = mtp_codec::CommunicationValue::new(mtp_codec::CommunicationType::Ping);
+    client_tx.send(&msg).await.unwrap();
+    client_tx.close();
+
+    // Host should still receive the message
+    let received = host_rx.receive().await.unwrap();
+    assert_eq!(received.get_type(), mtp_codec::CommunicationTypeId(19)); // Ping
+
+    // Host should get an error or closed signal on next receive
+    let result = host_rx.receive().await;
+    assert!(result.is_err());
+}