#[cfg(feature = "grpc")]
mod grpc_integration_tests {
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use sdforge::core::ApiMetadata;
use sdforge::grpc::sdforge_v1::{
sd_forge_service_client::SdForgeServiceClient, CallRequest, InfoRequest,
};
use sdforge::grpc::{GrpcServerConfig, SdForgeGrpcService};
use tonic::transport::Channel;
use tonic::Request;
const TEST_SERVER_ADDR: &str = "127.0.0.1:0"; const TEST_TIMEOUT_SECS: u64 = 30;
async fn setup_grpc_test_server() -> (SdForgeServiceClient<Channel>, SocketAddr) {
let (tx, rx) = std::sync::mpsc::channel();
let server_handle = tokio::spawn(async move {
let addr: SocketAddr = TEST_SERVER_ADDR.parse().unwrap();
let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
let bound_addr = listener.local_addr().unwrap();
tx.send(bound_addr).unwrap();
let service = SdForgeGrpcService::default();
let incoming = tokio_stream::wrappers::TcpListenerStream::new(listener);
tonic::transport::Server::builder()
.add_service(
sdforge::grpc::sdforge_v1::sd_forge_service_server::SdForgeServiceServer::new(
service,
)
.max_decoding_message_size(4 * 1024 * 1024),
)
.serve_with_incoming(incoming)
.await
.unwrap();
});
let server_addr = rx.recv().unwrap();
let channel = Channel::builder(format!("http://{}", server_addr).parse().unwrap())
.connect()
.await
.expect("Failed to connect to test server");
let client = SdForgeServiceClient::new(channel);
tokio::spawn(async move {
server_handle.await.ok();
});
(client, server_addr)
}
fn create_test_call_request(
method: &str,
params: HashMap<String, String>,
data: &str,
) -> CallRequest {
CallRequest {
method: method.to_string(),
parameters: params,
data: data.to_string(),
}
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_server_streaming() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let methods = vec!["stream_item_1", "stream_item_2", "stream_item_3"];
for method in methods {
let request = create_test_call_request(method, HashMap::new(), "");
let result = client.call(Request::new(request)).await;
assert!(
result.is_ok(),
"Server streaming request should succeed for method: {}",
method
);
let response = result.unwrap().into_inner();
assert!(response.success, "Response should indicate success");
}
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_client_streaming() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let items: Vec<(String, HashMap<String, String>)> = vec![
(
"batch_create".to_string(),
HashMap::from([("item".to_string(), "1".to_string())]),
),
(
"batch_create".to_string(),
HashMap::from([("item".to_string(), "2".to_string())]),
),
(
"batch_create".to_string(),
HashMap::from([("item".to_string(), "3".to_string())]),
),
];
let mut success_count = 0;
for (method, params) in items {
let request = create_test_call_request(&method, params, "");
if client.call(Request::new(request)).await.is_ok() {
success_count += 1;
}
}
assert_eq!(
success_count, 3,
"All client streaming requests should succeed"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_bidirectional_streaming() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let operations: Vec<(&str, HashMap<String, String>)> = vec![
(
"process_start",
HashMap::from([("id".to_string(), "1".to_string())]),
),
(
"process_data",
HashMap::from([
("id".to_string(), "1".to_string()),
("chunk".to_string(), "a".to_string()),
]),
),
(
"process_data",
HashMap::from([
("id".to_string(), "1".to_string()),
("chunk".to_string(), "b".to_string()),
]),
),
(
"process_end",
HashMap::from([("id".to_string(), "1".to_string())]),
),
];
for (method, params) in operations {
let request = create_test_call_request(method, params, "");
let result = client.call(Request::new(request)).await;
assert!(
result.is_ok(),
"Bidirectional streaming operation should succeed: {}",
method
);
}
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_call_with_metadata() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("metadata_test", HashMap::new(), "");
let mut req = Request::new(request);
req.metadata_mut().insert(
"x-correlation-id",
tonic::metadata::MetadataValue::try_from("test-correlation-123").unwrap(),
);
req.metadata_mut().insert(
"x-request-timestamp",
tonic::metadata::MetadataValue::try_from("1700000000").unwrap(),
);
let result = client.call(req).await;
assert!(result.is_ok(), "Call with metadata should succeed");
let response = result.unwrap().into_inner();
assert!(response.success, "Response should indicate success");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_metadata_canonical_headers() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("header_test", HashMap::new(), "");
let mut req = Request::new(request);
req.metadata_mut().insert(
"x-custom-header",
tonic::metadata::MetadataValue::try_from("canonical-value").unwrap(),
);
let result = client.call(req).await;
assert!(result.is_ok(), "Call with canonical headers should succeed");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_deadline_propagation() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("deadline_test", HashMap::new(), "");
let mut req = Request::new(request);
req.set_timeout(Duration::from_secs(TEST_TIMEOUT_SECS));
let result = client.call(req).await;
assert!(
result.is_ok(),
"Call with deadline should succeed within timeout"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_deadline_short_timeout() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("fast_op", HashMap::new(), "");
let mut req = Request::new(request);
req.set_timeout(Duration::from_secs(1));
let result = client.call(req).await;
assert!(
result.is_ok(),
"Fast operation should complete within short timeout"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_invalid_request_format() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("", HashMap::new(), "");
let result = client.call(Request::new(request)).await;
assert!(
result.is_ok(),
"Server should handle empty method name gracefully"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_service_unavailable() {
let addr: SocketAddr = "127.0.0.1:19999".parse().unwrap();
let channel = Channel::builder(format!("http://{}", addr).parse().unwrap())
.connect_timeout(Duration::from_millis(100))
.connect()
.await;
assert!(
channel.is_err(),
"Connection to unavailable service should fail"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_deadline_exceeded() {
let addr: SocketAddr = "127.0.0.1:19998".parse().unwrap();
let channel = Channel::builder(format!("http://{}", addr).parse().unwrap())
.connect_timeout(Duration::from_millis(50))
.timeout(Duration::from_millis(50))
.connect()
.await;
match channel {
Err(e) => {
let error_str = e.to_string().to_lowercase();
assert!(
error_str.contains("timeout")
|| error_str.contains("connect")
|| error_str.contains("status"),
"Error should be timeout-related: {}",
error_str
);
}
Ok(_) => {
let mut client = SdForgeServiceClient::new(channel.unwrap());
let request = create_test_call_request("test", HashMap::new(), "");
let mut req = Request::new(request);
req.set_timeout(Duration::from_secs(0));
let result = client.call(req).await;
assert!(result.is_err(), "Call with zero timeout should fail");
}
}
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_connection_failure() {
let addr: SocketAddr = "127.0.0.1:19997".parse().unwrap();
let result = tonic::transport::Endpoint::new(format!("http://{}", addr))
.unwrap()
.connect()
.await;
assert!(result.is_err(), "Connection to closed port should fail");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_error_status_mapping() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("status_test", HashMap::new(), "");
let result = client.call(Request::new(request)).await;
assert!(result.is_ok(), "Valid request should return OK status");
let response = result.unwrap().into_inner();
assert_eq!(
response.status_code, 200,
"Successful response should have status 200"
);
}
#[test]
fn test_grpc_service_config_default() {
let config = GrpcServerConfig::default();
assert_eq!(
config.max_connections, 1000,
"Default max_connections should be 1000"
);
assert_eq!(
config.timeout_seconds, 30,
"Default timeout_seconds should be 30"
);
}
#[test]
fn test_grpc_service_config_custom() {
let config = GrpcServerConfig {
max_connections: 500,
timeout_seconds: 60,
#[cfg(feature = "security")]
auth: None,
};
assert_eq!(
config.max_connections, 500,
"Custom max_connections should be 500"
);
assert_eq!(
config.timeout_seconds, 60,
"Custom timeout_seconds should be 60"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_interceptors_execution() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("interceptor_test", HashMap::new(), "");
let result = client.call(Request::new(request)).await;
assert!(
result.is_ok(),
"Request through interceptor layer should succeed"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_load_balancing_strategy() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request_count = 10;
let mut success_count = 0;
for i in 0..request_count {
let request = create_test_call_request(
"balance_test",
HashMap::from([("request_id".to_string(), i.to_string())]),
"",
);
if client.call(Request::new(request)).await.is_ok() {
success_count += 1;
}
}
assert_eq!(
success_count, request_count,
"All load-balanced requests should succeed"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_concurrent_load_balancing() {
let (client, _server_addr) = setup_grpc_test_server().await;
let client = Arc::new(tokio::sync::Mutex::new(client));
let mut handles = vec![];
for i in 0..5 {
let client_clone = client.clone();
let handle = tokio::spawn(async move {
let mut client = client_clone.lock().await;
let request = create_test_call_request(
"concurrent_balance",
HashMap::from([("index".to_string(), i.to_string())]),
"",
);
client.call(Request::new(request)).await
});
handles.push(handle);
}
let mut success_count = 0;
for handle in handles {
if handle.await.unwrap().is_ok() {
success_count += 1;
}
}
assert_eq!(
success_count, 5,
"All concurrent load-balanced requests should succeed"
);
}
#[cfg(feature = "security")]
#[test]
fn test_grpc_route_creation_with_metadata() {
use sdforge::grpc::GrpcRoute;
let metadata = ApiMetadata::new(
"test_service".to_string(),
"v1".to_string(),
"Test service description".to_string(),
Some(300),
true,
);
let route = GrpcRoute::new("test_service".to_string(), metadata);
let debug_str = format!("{:?}", route);
assert!(debug_str.contains("test_service"));
}
#[cfg(feature = "security")]
#[test]
fn test_grpc_route_streaming_metadata() {
use sdforge::grpc::GrpcRoute;
let streaming_metadata = ApiMetadata::new(
"stream_service".to_string(),
"v2".to_string(),
"A streaming service".to_string(),
None,
true,
);
let stream_route = GrpcRoute::new("stream_service".to_string(), streaming_metadata);
let _stream_debug = format!("{:?}", stream_route);
let normal_metadata = ApiMetadata::new(
"normal_service".to_string(),
"v1".to_string(),
"A normal service".to_string(),
Some(60),
false,
);
let normal_route = GrpcRoute::new("normal_service".to_string(), normal_metadata);
let _normal_debug = format!("{:?}", normal_route);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_call_method_with_params() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let mut params = HashMap::new();
params.insert("key1".to_string(), "value1".to_string());
params.insert("key2".to_string(), "value2".to_string());
let request = create_test_call_request("parameterized_call", params, "");
let result = client.call(Request::new(request)).await;
assert!(result.is_ok(), "Call with parameters should succeed");
let response = result.unwrap().into_inner();
assert!(response.success, "Response should indicate success");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_call_method_with_data() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let data = r#"{"key": "value", "nested": {"foo": "bar"}}"#;
let request = create_test_call_request("data_call", HashMap::new(), data);
let result = client.call(Request::new(request)).await;
assert!(result.is_ok(), "Call with data payload should succeed");
let response = result.unwrap().into_inner();
assert!(response.success, "Response should indicate success");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_get_info() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = InfoRequest {
version: "".to_string(),
};
let result = client.get_info(Request::new(request)).await;
assert!(result.is_ok(), "GetInfo should succeed");
let response = result.unwrap().into_inner();
assert_eq!(
response.name, "SdForge Service",
"Service name should match"
);
assert_eq!(response.version, "0.1.0", "Service version should match");
assert!(
!response.methods.is_empty(),
"Service should have available methods"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_get_info_with_version() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = InfoRequest {
version: "1.0.0".to_string(),
};
let result = client.get_info(Request::new(request)).await;
assert!(result.is_ok(), "GetInfo with version should succeed");
}
#[test]
fn test_grpc_valid_ipv4_address() {
let addr: SocketAddr = "127.0.0.1:50051".parse().unwrap();
assert_eq!(addr.port(), 50051);
assert_eq!(addr.ip().to_string(), "127.0.0.1");
}
#[test]
fn test_grpc_localhost_address() {
let addr_str = "localhost:8080";
let parts: Vec<&str> = addr_str.split(':').collect();
assert_eq!(parts.len(), 2);
assert_eq!(parts[0], "localhost");
assert_eq!(parts[1], "8080");
}
#[test]
fn test_grpc_invalid_address_format() {
let result: Result<SocketAddr, _> = "invalid-address".parse();
assert!(result.is_err(), "Invalid address format should be rejected");
}
#[test]
fn test_grpc_missing_port() {
let result: Result<SocketAddr, _> = "127.0.0.1".parse();
assert!(result.is_err(), "Address without port should be rejected");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_response_success_flag() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("success_test", HashMap::new(), "");
let result = client.call(Request::new(request)).await;
assert!(result.is_ok(), "Call should succeed");
let response = result.unwrap().into_inner();
assert!(
response.success,
"Successful response should have success=true"
);
assert_eq!(
response.status_code, 200,
"Successful response should have status_code 200"
);
assert!(
response.error.is_empty(),
"Successful response should have empty error"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_response_data_format() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request("data_format_test", HashMap::new(), "");
let result = client.call(Request::new(request)).await;
assert!(result.is_ok(), "Call should succeed");
let response = result.unwrap().into_inner();
let data: serde_json::Value =
serde_json::from_str(&response.data).expect("Response data should be valid JSON");
assert!(
data.get("method").is_some(),
"Response data should contain method field"
);
assert_eq!(
data.get("result").and_then(|v| v.as_str()),
Some("processed"),
"Response data should contain processed result"
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_connection_reuse() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
for i in 0..5 {
let request = create_test_call_request(
"reuse_test",
HashMap::from([("iteration".to_string(), i.to_string())]),
"",
);
let result = client.call(Request::new(request)).await;
assert!(
result.is_ok(),
"Request {} on reused connection should succeed",
i
);
}
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_connection_keep_alive() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request1 = create_test_call_request("keepalive_1", HashMap::new(), "");
let result1 = client.call(Request::new(request1)).await;
assert!(result1.is_ok(), "First request should succeed");
tokio::time::sleep(Duration::from_millis(10)).await;
let request2 = create_test_call_request("keepalive_2", HashMap::new(), "");
let result2 = client.call(Request::new(request2)).await;
assert!(
result2.is_ok(),
"Second request should succeed (keep-alive)"
);
}
#[tokio::test]
#[cfg(feature = "security")]
async fn test_grpc_server_config_with_security() {
use sdforge::security::BearerAuth;
let auth =
BearerAuth::try_new("TestSecret123!ABCDEFGHIJKLMNOPQRSTUVWXYZ").expect("Valid secret");
let config = GrpcServerConfig {
max_connections: 100,
timeout_seconds: 60,
auth: Some(auth),
};
assert!(config.auth.is_some(), "Config should have auth when set");
}
#[tokio::test]
#[cfg(feature = "security")]
async fn test_grpc_server_config_without_security() {
let config = GrpcServerConfig::default();
assert!(config.auth.is_none(), "Default config should have no auth");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_high_concurrency() {
let (client, _server_addr) = setup_grpc_test_server().await;
let client = Arc::new(tokio::sync::Mutex::new(client));
let request_count = 50;
let mut handles = vec![];
for i in 0..request_count {
let client_clone = client.clone();
let handle = tokio::spawn(async move {
let mut client = client_clone.lock().await;
let request = create_test_call_request(
"stress_test",
HashMap::from([("request_num".to_string(), i.to_string())]),
"",
);
client.call(Request::new(request)).await
});
handles.push(handle);
}
let mut success_count = 0;
for handle in handles {
if handle.await.unwrap().is_ok() {
success_count += 1;
}
}
assert!(
success_count >= request_count - 5,
"Most high-concurrency requests should succeed, got {} of {}",
success_count,
request_count
);
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_large_method_name() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let long_method = format!("method_{}", "x".repeat(1000));
let request = create_test_call_request(&long_method, HashMap::new(), "");
let result = client.call(Request::new(request)).await;
assert!(result.is_ok(), "Call with large method name should succeed");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_unicode_support() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request(
"unicode_method",
HashMap::from([
("name".to_string(), "ζ΅θ―η¨ζ·".to_string()),
("emoji".to_string(), "πππ".to_string()),
]),
"Unicode data: δ½ ε₯½δΈη π",
);
let result = client.call(Request::new(request)).await;
assert!(result.is_ok(), "Call with unicode should succeed");
}
#[tokio::test]
#[ignore = "environmental issue: real network binding to 127.0.0.1:0, hangs in CI/sandboxed environments"]
async fn test_grpc_special_characters() {
let (mut client, _server_addr) = setup_grpc_test_server().await;
let request = create_test_call_request(
"special_chars",
HashMap::from([
("email".to_string(), "user@example.com".to_string()),
("path".to_string(), "/api/v1/users".to_string()),
("query".to_string(), "name=John&age=30".to_string()),
]),
r#"{"json": "with \"quotes\" and \n newlines"}"#,
);
let result = client.call(Request::new(request)).await;
assert!(
result.is_ok(),
"Call with special characters should succeed"
);
}
#[test]
fn test_grpc_build_server_valid_address() {
let addr = "127.0.0.1:0";
let socket_addr: SocketAddr = addr.parse().unwrap();
assert!(socket_addr.port() == 0 || socket_addr.port() > 0);
}
#[test]
fn test_grpc_build_server_invalid_address() {
let invalid_addr = "not-valid";
let result: Result<SocketAddr, _> = invalid_addr.parse();
assert!(
result.is_err(),
"Invalid address should be rejected by build_server"
);
}
#[test]
fn test_grpc_build_server_with_config() {
let config = GrpcServerConfig {
max_connections: 200,
timeout_seconds: 45,
#[cfg(feature = "security")]
auth: None,
};
assert_eq!(config.max_connections, 200);
assert_eq!(config.timeout_seconds, 45);
}
}