#[cfg(feature = "websocket")]
mod websocket_tests {
use sdforge::websocket::{
parse_websocket_message, AppState, ConnectionManager, WebSocketConfig, WebSocketConnection,
WebSocketMessage,
};
use std::sync::{Arc, Mutex};
#[tokio::test]
async fn test_websocket_connection_establishment() {
let manager = Arc::new(ConnectionManager::new());
for i in 0..5 {
let conn_id = format!("conn-{}", i);
let (conn, _rx) = WebSocketConnection::new(conn_id.clone());
manager.add_connection(conn_id, conn).await;
}
assert_eq!(manager.connection_count().await, 5);
let conn = manager.get_connection("conn-2").await;
assert!(conn.is_some());
assert_eq!(conn.unwrap().id(), "conn-2");
}
#[cfg(feature = "security")]
#[tokio::test]
async fn test_websocket_with_auth_headers() {
let auth =
sdforge::security::BearerAuth::try_new("ValidSecret123!ABCDEFGHIJKLMNOPQRSTUVWXYZ")
.expect("valid secret");
let config = WebSocketConfig {
auth: Some(auth),
..Default::default()
};
assert!(config.auth.is_some());
}
#[tokio::test]
async fn test_websocket_multiple_connections() {
let manager = Arc::new(ConnectionManager::new());
let num_connections = 50;
for i in 0..num_connections {
let conn_id = format!("multi-conn-{}", i);
let (conn, _rx) = WebSocketConnection::new(conn_id);
manager.add_connection(format!("conn-{}", i), conn).await;
}
assert_eq!(manager.connection_count().await, num_connections);
for i in 0..num_connections {
manager.remove_connection(&format!("conn-{}", i)).await;
}
assert_eq!(manager.connection_count().await, 0);
}
#[tokio::test]
async fn test_websocket_send_request_message() {
let (conn, mut receiver) = WebSocketConnection::new("send-req".to_string());
let request = WebSocketMessage::Request {
id: "req-123".to_string(),
method: "get_data".to_string(),
params: serde_json::json!({"key": "value"}),
};
let result = conn.send(request.clone()).await;
assert!(result.is_ok());
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(msg) = received {
match msg {
WebSocketMessage::Request { id, method, params } => {
assert_eq!(id, "req-123");
assert_eq!(method, "get_data");
assert_eq!(params["key"], "value");
}
_ => panic!("Expected Request message"),
}
}
}
#[tokio::test]
async fn test_websocket_receive_response() {
let (conn, mut receiver) = WebSocketConnection::new("recv-resp".to_string());
let response = WebSocketMessage::Response {
id: "resp-456".to_string(),
result: serde_json::json!({"status": "ok", "data": [1, 2, 3]}),
};
conn.send(response.clone()).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(msg) = received {
match msg {
WebSocketMessage::Response { id, result } => {
assert_eq!(id, "resp-456");
assert_eq!(result["status"], "ok");
assert_eq!(result["data"], serde_json::json!([1, 2, 3]));
}
_ => panic!("Expected Response message"),
}
}
}
#[tokio::test]
async fn test_websocket_json_message_format() {
let (conn, mut receiver) = WebSocketConnection::new("json-format".to_string());
let request = WebSocketMessage::Request {
id: "json-test".to_string(),
method: "complex_query".to_string(),
params: serde_json::json!({
"string": "hello",
"number": 42,
"float": 1.5,
"boolean": true,
"null": null,
"array": [{"a": 1}, {"b": 2}],
"nested": {"level1": {"level2": "deep"}}
}),
};
conn.send(request.clone()).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Request { params, .. }) = received {
assert_eq!(params["string"], "hello");
assert_eq!(params["number"], 42);
assert_eq!(params["float"], 1.5);
assert_eq!(params["boolean"], true);
assert!(params["null"].is_null());
assert_eq!(params["array"].as_array().unwrap().len(), 2);
assert_eq!(params["nested"]["level1"]["level2"], "deep");
}
}
#[tokio::test]
async fn test_websocket_binary_message_support() {
let (conn, mut receiver) = WebSocketConnection::new("binary-support".to_string());
let binary_data = "SGVsbG8gV29ybGQhIFRoaXMgaXMgYmluYXJ5IGRhdGEh";
let request = WebSocketMessage::Request {
id: "binary-001".to_string(),
method: "process_binary".to_string(),
params: serde_json::json!({
"encoding": "base64",
"data": binary_data,
"mime_type": "application/octet-stream"
}),
};
conn.send(request.clone()).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Request { params, .. }) = received {
assert_eq!(params["encoding"], "base64");
assert_eq!(params["data"], binary_data);
assert_eq!(params["mime_type"], "application/octet-stream");
}
}
#[tokio::test]
async fn test_websocket_message_ordering() {
let (conn, mut receiver) = WebSocketConnection::new("ordering".to_string());
for i in 0..10 {
let msg = WebSocketMessage::Notification {
event: "ordered_event".to_string(),
data: serde_json::json!({"index": i, "timestamp": i * 100}),
};
conn.send(msg).await.unwrap();
}
for i in 0..10 {
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Notification { data, .. }) = received {
assert_eq!(data["index"], i);
assert_eq!(data["timestamp"], i * 100);
}
}
}
#[tokio::test]
async fn test_websocket_heartbeat_mechanism() {
let manager = Arc::new(ConnectionManager::new());
let (conn, _rx) = WebSocketConnection::new("heartbeat-conn".to_string());
manager
.add_connection("heartbeat-conn".to_string(), conn)
.await;
assert_eq!(manager.connection_count().await, 1);
assert!(manager.get_connection("heartbeat-conn").await.is_some());
for _ in 0..3 {
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
assert!(manager.get_connection("heartbeat-conn").await.is_some());
}
manager.remove_connection("heartbeat-conn").await;
assert_eq!(manager.connection_count().await, 0);
}
#[tokio::test]
async fn test_websocket_ping_pong_handling() {
let (conn, mut receiver) = WebSocketConnection::new("ping-pong".to_string());
let ping = WebSocketMessage::Notification {
event: "ping".to_string(),
data: serde_json::json!({"timestamp": 1234567890}),
};
conn.send(ping).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Notification { event, .. }) = received {
assert_eq!(event, "ping");
}
let pong = WebSocketMessage::Notification {
event: "pong".to_string(),
data: serde_json::json!({"timestamp": 1234567891}),
};
conn.send(pong).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Notification { event, .. }) = received {
assert_eq!(event, "pong");
}
}
#[tokio::test]
async fn test_websocket_graceful_disconnect() {
let manager = Arc::new(ConnectionManager::new());
let (conn, mut receiver) = WebSocketConnection::new("graceful-disconn".to_string());
manager
.add_connection("graceful-disconn".to_string(), conn.clone())
.await;
let last_msg = WebSocketMessage::Notification {
event: "goodbye".to_string(),
data: serde_json::json!({"reason": "normal_close"}),
};
conn.send(last_msg).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
manager.remove_connection("graceful-disconn").await;
assert_eq!(manager.connection_count().await, 0);
assert!(manager.get_connection("graceful-disconn").await.is_none());
}
#[tokio::test]
async fn test_websocket_abnormal_disconnect() {
let manager = Arc::new(ConnectionManager::new());
for i in 0..5 {
let (conn, _) = WebSocketConnection::new(format!("abnormal-{}", i));
manager
.add_connection(format!("abnormal-{}", i), conn)
.await;
}
assert_eq!(manager.connection_count().await, 5);
manager.remove_connection("abnormal-0").await;
manager.remove_connection("abnormal-2").await;
manager.remove_connection("abnormal-4").await;
assert_eq!(manager.connection_count().await, 2);
assert!(manager.get_connection("abnormal-1").await.is_some());
assert!(manager.get_connection("abnormal-3").await.is_some());
assert!(manager.get_connection("abnormal-0").await.is_none());
}
#[tokio::test]
async fn test_websocket_reconnection_handling() {
let manager = Arc::new(ConnectionManager::new());
let (conn1, _) = WebSocketConnection::new("reconnect-conn".to_string());
manager
.add_connection("reconnect-conn".to_string(), conn1)
.await;
assert_eq!(manager.connection_count().await, 1);
manager.remove_connection("reconnect-conn").await;
assert_eq!(manager.connection_count().await, 0);
let (conn2, _) = WebSocketConnection::new("reconnect-conn".to_string());
manager
.add_connection("reconnect-conn".to_string(), conn2)
.await;
assert_eq!(manager.connection_count().await, 1);
let retrieved = manager.get_connection("reconnect-conn").await;
assert!(retrieved.is_some());
assert_eq!(retrieved.unwrap().id(), "reconnect-conn");
manager.remove_connection("reconnect-conn").await;
assert_eq!(manager.connection_count().await, 0);
}
#[tokio::test]
async fn test_websocket_broadcast_message() {
let manager = Arc::new(ConnectionManager::new());
let mut receivers = Vec::new();
for i in 0..3 {
let (conn, rx) = WebSocketConnection::new(format!("broadcast-{}", i));
manager
.add_connection(format!("broadcast-{}", i), conn)
.await;
receivers.push(rx);
}
let broadcast_msg = Arc::new(WebSocketMessage::Notification {
event: "broadcast".to_string(),
data: serde_json::json!({"message": "Hello all!"}),
});
manager.broadcast(&broadcast_msg).await;
for (i, rx) in receivers.iter_mut().enumerate() {
let received =
tokio::time::timeout(tokio::time::Duration::from_secs(1), rx.recv()).await;
assert!(received.is_ok(), "Receiver {} should receive message", i);
let msg = received.unwrap().unwrap();
match msg {
WebSocketMessage::Notification { event, data } => {
assert_eq!(event, "broadcast");
assert_eq!(data["message"], "Hello all!");
}
_ => panic!("Expected Notification"),
}
}
}
#[test]
fn test_websocket_message_request_serialization() {
let message = WebSocketMessage::Request {
id: "123".to_string(),
method: "get_data".to_string(),
params: serde_json::json!({"key": "value"}),
};
let serialized = serde_json::to_string(&message).unwrap();
assert!(serialized.contains("\"type\":\"request\""));
assert!(serialized.contains("123"));
assert!(serialized.contains("get_data"));
}
#[test]
fn test_websocket_message_response_serialization() {
let message = WebSocketMessage::Response {
id: "456".to_string(),
result: serde_json::json!({"status": "ok"}),
};
let serialized = serde_json::to_string(&message).unwrap();
assert!(serialized.contains("\"type\":\"response\""));
assert!(serialized.contains("456"));
}
#[test]
fn test_websocket_message_error_serialization() {
let message = WebSocketMessage::Error {
id: "789".to_string(),
error: "Something went wrong".to_string(),
};
let serialized = serde_json::to_string(&message).unwrap();
assert!(serialized.contains("\"type\":\"error\""));
assert!(serialized.contains("Something went wrong"));
}
#[test]
fn test_websocket_config_default() {
let config = WebSocketConfig::default();
let _ = config;
}
#[test]
fn test_connection_manager_new() {
let manager = ConnectionManager::new();
let _ = manager;
}
#[test]
fn test_app_state_new() {
let manager = Arc::new(ConnectionManager::new());
let state = AppState::new(manager);
let _ = state;
}
#[test]
fn test_websocket_connection_new() {
let (conn, _rx) = WebSocketConnection::new("test-conn-1".to_string());
assert_eq!(conn.id(), "test-conn-1");
}
#[tokio::test]
async fn test_connection_lifecycle() {
let manager = Arc::new(ConnectionManager::new());
let conn_id = "lifecycle-test".to_string();
let (conn, mut receiver) = WebSocketConnection::new(conn_id.clone());
manager.add_connection(conn_id.clone(), conn.clone()).await;
assert_eq!(manager.connection_count().await, 1);
assert!(manager.get_connection(&conn_id).await.is_some());
let msg = WebSocketMessage::Notification {
event: "lifecycle_event".to_string(),
data: serde_json::json!({"phase": "active"}),
};
conn.send(msg).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
let msg2 = WebSocketMessage::Notification {
event: "lifecycle_check".to_string(),
data: serde_json::json!({"status": "ok"}),
};
conn.send(msg2).await.unwrap();
manager.remove_connection(&conn_id).await;
assert_eq!(manager.connection_count().await, 0);
assert!(manager.get_connection(&conn_id).await.is_none());
}
#[tokio::test]
async fn test_connection_id_uniqueness() {
let manager = Arc::new(ConnectionManager::new());
let mut connection_ids = Vec::new();
for i in 0..100 {
let conn_id = format!("unique-conn-{}", i);
let (conn, _rx) = WebSocketConnection::new(conn_id.clone());
manager.add_connection(conn_id.clone(), conn).await;
connection_ids.push(conn_id);
}
for id in &connection_ids {
let conn = manager.get_connection(id).await;
assert!(conn.is_some(), "Connection {} should exist", id);
assert_eq!(conn.unwrap().id(), id);
}
for id in &connection_ids {
manager.remove_connection(id).await;
}
assert_eq!(manager.connection_count().await, 0);
}
#[tokio::test]
async fn test_connection_clone_usage() {
let manager = Arc::new(ConnectionManager::new());
let (conn1, mut receiver) = WebSocketConnection::new("clone-original".to_string());
manager
.add_connection("clone-original".to_string(), conn1.clone())
.await;
let msg = WebSocketMessage::Notification {
event: "clone_test".to_string(),
data: serde_json::json!({"source": "cloned_connection"}),
};
conn1.send(msg).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Notification { event, data }) = received {
assert_eq!(event, "clone_test");
assert_eq!(data["source"], "cloned_connection");
}
manager.remove_connection("clone-original").await;
}
#[tokio::test]
async fn test_concurrent_message_sending() {
let (conn, mut receiver) = WebSocketConnection::new("concurrent-send".to_string());
let num_messages = 50;
let conn_clone = conn.clone();
tokio::spawn(async move {
for i in 0..num_messages {
let msg = WebSocketMessage::Notification {
event: format!("event_{}", i),
data: serde_json::json!({"index": i}),
};
let _ = conn_clone.send(msg).await;
}
})
.await
.unwrap();
let mut received_count = 0;
while let Ok(Some(_msg)) =
tokio::time::timeout(tokio::time::Duration::from_secs(2), receiver.recv()).await
{
received_count += 1;
}
assert_eq!(received_count, num_messages);
}
#[tokio::test]
async fn test_concurrent_connection_creation() {
let manager = Arc::new(ConnectionManager::new());
let num_connections = 100;
let mut handles = Vec::new();
for i in 0..num_connections {
let manager_clone = manager.clone();
let conn_id = format!("concurrent-conn-{}", i);
handles.push(tokio::spawn(async move {
let (conn, _rx) = WebSocketConnection::new(conn_id.clone());
manager_clone.add_connection(conn_id, conn).await;
}));
}
for handle in handles {
handle.await.unwrap();
}
assert_eq!(manager.connection_count().await, num_connections);
for i in 0..num_connections {
manager
.remove_connection(&format!("concurrent-conn-{}", i))
.await;
}
assert_eq!(manager.connection_count().await, 0);
}
#[tokio::test]
async fn test_message_queue_processing() {
let (conn, mut receiver) = WebSocketConnection::new("queue-test".to_string());
for i in 0..20 {
let msg = WebSocketMessage::Request {
id: format!("queue-req-{}", i),
method: "process".to_string(),
params: serde_json::json!({"queue_index": i}),
};
conn.send(msg).await.unwrap();
}
let mut indices = Vec::new();
for _ in 0..20 {
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Request { params, .. }) = received {
indices.push(params["queue_index"].as_i64().unwrap());
}
}
indices.sort();
for (i, idx) in indices.iter().enumerate() {
assert_eq!(*idx, i as i64);
}
}
#[tokio::test]
async fn test_mixed_message_types() {
let (conn, mut receiver) = WebSocketConnection::new("mixed-types".to_string());
let request = WebSocketMessage::Request {
id: "req-1".to_string(),
method: "test".to_string(),
params: serde_json::json!({"type": "request"}),
};
let response = WebSocketMessage::Response {
id: "resp-1".to_string(),
result: serde_json::json!({"type": "response"}),
};
let error = WebSocketMessage::Error {
id: "err-1".to_string(),
error: "test error".to_string(),
};
let notification = WebSocketMessage::Notification {
event: "notify".to_string(),
data: serde_json::json!({"type": "notification"}),
};
conn.send(request).await.unwrap();
conn.send(response).await.unwrap();
conn.send(error).await.unwrap();
conn.send(notification).await.unwrap();
let mut types = Vec::new();
for _ in 0..4 {
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(msg) = received {
match msg {
WebSocketMessage::Request { .. } => types.push("request"),
WebSocketMessage::Response { .. } => types.push("response"),
WebSocketMessage::Error { .. } => types.push("error"),
WebSocketMessage::Notification { .. } => types.push("notification"),
}
}
}
assert_eq!(types.len(), 4);
assert!(types.contains(&"request"));
assert!(types.contains(&"response"));
assert!(types.contains(&"error"));
assert!(types.contains(&"notification"));
}
#[tokio::test]
async fn test_large_scale_broadcast() {
let manager = Arc::new(ConnectionManager::new());
let num_connections = 50;
let mut receivers = Vec::new();
for i in 0..num_connections {
let (conn, rx) = WebSocketConnection::new(format!("broadcast-{}", i));
manager
.add_connection(format!("broadcast-{}", i), conn)
.await;
receivers.push(rx);
}
let broadcast_msg = Arc::new(WebSocketMessage::Notification {
event: "large_broadcast".to_string(),
data: serde_json::json!({
"message": "This is a broadcast message",
"timestamp": chrono::Utc::now().timestamp()
}),
});
manager.broadcast(&broadcast_msg).await;
for (i, rx) in receivers.iter_mut().enumerate() {
let received =
tokio::time::timeout(tokio::time::Duration::from_secs(2), rx.recv()).await;
assert!(
received.is_ok(),
"Receiver {} should receive message within timeout",
i
);
let msg = received.unwrap().unwrap();
match msg {
WebSocketMessage::Notification { event, .. } => {
assert_eq!(event, "large_broadcast");
}
_ => panic!("Expected Notification message"),
}
}
for i in 0..num_connections {
manager.remove_connection(&format!("broadcast-{}", i)).await;
}
}
#[tokio::test]
async fn test_broadcast_cleans_failed_connections() {
let manager = Arc::new(ConnectionManager::new());
let (conn1, _rx) = WebSocketConnection::new("will-fail".to_string());
manager.add_connection("will-fail".to_string(), conn1).await;
let (conn2, mut rx2) = WebSocketConnection::new("will-succeed".to_string());
manager
.add_connection("will-succeed".to_string(), conn2)
.await;
assert_eq!(manager.connection_count().await, 2);
let msg = Arc::new(WebSocketMessage::Notification {
event: "test".to_string(),
data: serde_json::json!({}),
});
manager.broadcast(&msg).await;
let received =
tokio::time::timeout(tokio::time::Duration::from_millis(100), rx2.recv()).await;
assert!(received.is_ok());
manager.remove_connection("will-fail").await;
manager.remove_connection("will-succeed").await;
}
#[tokio::test]
async fn test_unicode_message_content() {
let (conn, mut receiver) = WebSocketConnection::new("unicode-test".to_string());
let request = WebSocketMessage::Request {
id: "unicode-测试-🎉".to_string(),
method: "方法-метод-🔧".to_string(),
params: serde_json::json!({
"chinese": "你好世界",
"japanese": "こんにちは世界",
"korean": "안녕하세요",
"russian": "Привет мир",
"arabic": "مرحبا بالعالم",
"emoji": "🎉🎊🎁🎄🎅",
"special": "⟨test⟩«test»「test」"
}),
};
conn.send(request.clone()).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Request { id, method, params }) = received {
assert_eq!(id, "unicode-测试-🎉");
assert_eq!(method, "方法-метод-🔧");
assert_eq!(params["chinese"], "你好世界");
assert_eq!(params["emoji"], "🎉🎊🎁🎄🎅");
}
}
#[tokio::test]
async fn test_large_params_message() {
let (conn, mut receiver) = WebSocketConnection::new("large-params".to_string());
let large_array: Vec<i32> = (0..1000).collect();
let request = WebSocketMessage::Request {
id: "large-params".to_string(),
method: "batch_process".to_string(),
params: serde_json::json!({
"items": large_array,
"metadata": {
"count": 1000,
"type": "batch"
}
}),
};
conn.send(request.clone()).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Request { params, .. }) = received {
let items = params["items"].as_array().unwrap();
assert_eq!(items.len(), 1000);
assert_eq!(items[0], 0);
assert_eq!(items[999], 999);
}
}
#[tokio::test]
async fn test_deeply_nested_json() {
let mut nested = serde_json::json!({"value": 42});
for i in 0..10 {
nested = serde_json::json!({
format!("level_{}", i): nested,
"sibling": i
});
}
let (conn, mut receiver) = WebSocketConnection::new("nested".to_string());
let request = WebSocketMessage::Request {
id: "nested-deep".to_string(),
method: "process_nested".to_string(),
params: nested,
};
conn.send(request).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
}
#[tokio::test]
async fn test_error_message_handling() {
let (conn, mut receiver) = WebSocketConnection::new("error-handling".to_string());
let error = WebSocketMessage::Error {
id: "error-123".to_string(),
error: "Invalid operation: resource not found".to_string(),
};
conn.send(error.clone()).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
if let Some(WebSocketMessage::Error { id, error }) = received {
assert_eq!(id, "error-123");
assert!(error.contains("resource not found"));
}
}
#[tokio::test]
async fn test_empty_message_content() {
let (conn, mut receiver) = WebSocketConnection::new("empty-content".to_string());
let request = WebSocketMessage::Request {
id: String::new(),
method: String::new(),
params: serde_json::json!({}),
};
conn.send(request).await.unwrap();
let received = receiver.recv().await;
assert!(received.is_some());
}
#[tokio::test]
async fn test_app_state_connection_management() {
let manager = Arc::new(ConnectionManager::new());
let state = AppState::new(manager.clone());
let (conn, _) = WebSocketConnection::new("app-state-conn".to_string());
state
.manager
.add_connection("app-state-conn".to_string(), conn)
.await;
assert_eq!(state.manager.connection_count().await, 1);
assert!(state
.manager
.get_connection("app-state-conn")
.await
.is_some());
assert_eq!(state.config.max_message_size, 1_048_576);
state.manager.remove_connection("app-state-conn").await;
}
#[cfg(feature = "security")]
#[tokio::test]
async fn test_app_state_custom_config() {
let auth =
sdforge::security::BearerAuth::try_new("ValidSecret123!ABCDEFGHIJKLMNOPQRSTUVWXYZ")
.expect("valid secret");
let config = WebSocketConfig {
auth: Some(auth),
max_message_size: 2_000_000,
..Default::default()
};
let manager = Arc::new(ConnectionManager::new());
let state = AppState::with_config(config, manager.clone());
assert!(state.config.auth.is_some());
assert_eq!(state.config.max_message_size, 2_000_000);
}
#[tokio::test]
async fn test_concurrent_broadcast() {
let manager = Arc::new(ConnectionManager::new());
let num_connections = 10;
let num_broadcasts = 5;
let mut receivers = Vec::new();
for i in 0..num_connections {
let (conn, rx) = WebSocketConnection::new(format!("concurrent-broadcast-{}", i));
manager
.add_connection(format!("concurrent-broadcast-{}", i), conn)
.await;
receivers.push((rx, Mutex::new(0)));
}
let mut handles = Vec::new();
for i in 0..num_broadcasts {
let msg = Arc::new(WebSocketMessage::Notification {
event: format!("concurrent_broadcast_{}", i),
data: serde_json::json!({"broadcast_id": i}),
});
let manager_clone = manager.clone();
handles.push(tokio::spawn(async move {
manager_clone.broadcast(&msg).await;
}));
}
for handle in handles {
handle.await.unwrap();
}
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
for i in 0..num_connections {
manager
.remove_connection(&format!("concurrent-broadcast-{}", i))
.await;
}
}
#[tokio::test]
async fn test_rapid_add_remove_connections() {
let manager = Arc::new(ConnectionManager::new());
let iterations = 100;
for i in 0..iterations {
let conn_id = format!("rapid-{}", i);
let (conn, _) = WebSocketConnection::new(conn_id.clone());
manager.add_connection(conn_id.clone(), conn).await;
manager.remove_connection(&conn_id).await;
}
assert_eq!(manager.connection_count().await, 0);
}
#[tokio::test]
async fn test_connection_id_replacement() {
let manager = Arc::new(ConnectionManager::new());
let conn_id = "replaceable".to_string();
let (conn1, _) = WebSocketConnection::new(conn_id.clone());
manager.add_connection(conn_id.clone(), conn1).await;
assert_eq!(manager.connection_count().await, 1);
let (conn2, _) = WebSocketConnection::new(conn_id.clone());
manager.add_connection(conn_id.clone(), conn2).await;
let count = manager.connection_count().await;
assert!((1..=2).contains(&count));
let retrieved = manager.get_connection(&conn_id).await;
assert!(retrieved.is_some());
manager.remove_connection(&conn_id).await;
}
#[tokio::test]
async fn test_high_throughput_messaging() {
let (conn, mut receiver) = WebSocketConnection::new("throughput-test".to_string());
let num_messages = 500;
let start = tokio::time::Instant::now();
for i in 0..num_messages {
let msg = WebSocketMessage::Notification {
event: format!("event_{}", i),
data: serde_json::json!({"index": i}),
};
conn.send(msg).await.unwrap();
}
let mut received_count = 0;
while let Ok(Some(_)) =
tokio::time::timeout(tokio::time::Duration::from_secs(5), receiver.recv()).await
{
received_count += 1;
if received_count >= num_messages {
break;
}
}
let elapsed = start.elapsed();
assert_eq!(received_count, num_messages);
println!("Received {} messages in {:?}", received_count, elapsed);
}
#[tokio::test]
async fn test_connection_lookup_performance() {
let manager = Arc::new(ConnectionManager::new());
for i in 0..100 {
let (conn, _) = WebSocketConnection::new(format!("perf-{}", i));
manager.add_connection(format!("perf-{}", i), conn).await;
}
let start = tokio::time::Instant::now();
for i in 0..100 {
let conn = manager.get_connection(&format!("perf-{}", i)).await;
assert!(conn.is_some());
}
let elapsed = start.elapsed();
println!("100 connection lookups took {:?}", elapsed);
for i in 0..100 {
manager.remove_connection(&format!("perf-{}", i)).await;
}
}
#[test]
fn test_invalid_json_handling() {
let invalid_inputs = vec![
"", " ", "{", "}", "[", "{{}}", "{\"type\"}", "not json", "null", "123", ];
for input in invalid_inputs {
let result = parse_websocket_message(input);
assert!(result.is_err(), "Input {:?} should produce an error", input);
}
}
#[test]
fn test_message_type_boundaries() {
let req = WebSocketMessage::Request {
id: "".to_string(),
method: "".to_string(),
params: serde_json::json!(null),
};
let json = serde_json::to_string(&req).unwrap();
assert!(serde_json::from_str::<WebSocketMessage>(&json).is_ok());
let large_array: Vec<u32> = (0..1000).collect();
let resp = WebSocketMessage::Response {
id: "big-response".to_string(),
result: serde_json::json!({"values": large_array}),
};
let json = serde_json::to_string(&resp).unwrap();
assert!(serde_json::from_str::<WebSocketMessage>(&json).is_ok());
let long_error = "x".repeat(10000);
let err = WebSocketMessage::Error {
id: "long-error".to_string(),
error: long_error,
};
let json = serde_json::to_string(&err).unwrap();
assert!(serde_json::from_str::<WebSocketMessage>(&json).is_ok());
}
#[cfg(feature = "security")]
#[test]
fn test_config_deep_clone() {
let auth =
sdforge::security::BearerAuth::try_new("ValidSecret123!ABCDEFGHIJKLMNOPQRSTUVWXYZ")
.expect("valid secret");
let config1 = WebSocketConfig {
auth: Some(auth),
max_message_size: 2_000_000,
..Default::default()
};
let config2 = config1.clone();
assert!(config2.auth.is_some());
assert_eq!(config2.max_message_size, 2_000_000);
}
#[test]
fn test_default_config_consistency() {
let config = WebSocketConfig::default();
assert_eq!(config.max_message_size, 1_048_576);
}
}