use std::fs;
use std::marker::Sync;
use std::net::{Ipv4Addr, SocketAddr, TcpListener};
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use backon::{ExponentialBuilder, Retryable};
use tracing::{debug, error};
use crate::ApiClient;
mod error;
pub use self::error::*;
mod test_server;
pub use self::test_server::*;
#[derive(Debug, derive_more::Deref, derive_more::DerefMut)]
pub struct TestClient<T> {
#[allow(dead_code)]
local_addr: SocketAddr,
#[deref]
#[deref_mut]
client: ApiClient,
handle: Option<ServerTaskHandle>,
#[allow(dead_code)]
test_server: Arc<T>,
}
struct ServerTaskHandle(tokio::task::JoinHandle<()>);
impl ServerTaskHandle {
fn new(handle: tokio::task::JoinHandle<()>) -> Self {
Self(handle)
}
fn abort(&self) {
self.0.abort();
}
#[cfg(test)]
fn is_finished(&self) -> bool {
self.0.is_finished()
}
}
impl std::fmt::Debug for ServerTaskHandle {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_tuple("ServerTaskHandle")
.field(&self.0.id())
.finish()
}
}
impl<T> TestClient<T>
where
T: TestServer + Send + Sync + 'static,
{
pub async fn start(test_server: T) -> Result<Self, TestAppError> {
let addr = SocketAddr::from((Ipv4Addr::LOCALHOST, 0));
let listener = TcpListener::bind(addr)?;
let local_addr = listener.local_addr()?;
let test_server = Arc::new(test_server);
let handle = tokio::spawn({
let server = Arc::clone(&test_server);
async move {
if let Err(error) = server.launch(listener).await {
error!(?error, "Server launch failed");
}
}
});
let TestServerConfig {
api_client,
min_backoff_delay,
max_backoff_delay,
backoff_jitter,
max_retry_attempts,
} = test_server.config();
let client = api_client.unwrap_or_else(ApiClient::builder);
let client = client.with_port(local_addr.port()).build()?;
let healthy = Self::wait_for_health(
&test_server,
&client,
local_addr,
min_backoff_delay,
max_backoff_delay,
backoff_jitter,
max_retry_attempts,
)
.await;
if !healthy {
return Err(TestAppError::UnhealthyServer {
timeout: max_backoff_delay,
});
}
let result = Self {
local_addr,
client,
handle: Some(ServerTaskHandle::new(handle)),
test_server,
};
Ok(result)
}
async fn wait_for_health(
test_server: &Arc<T>,
client: &ApiClient,
local_addr: SocketAddr,
min_backoff_delay: Duration,
max_backoff_delay: Duration,
backoff_jitter: bool,
max_retry_attempts: usize,
) -> bool {
let mut backoff_builder = ExponentialBuilder::default()
.with_min_delay(min_backoff_delay)
.with_max_delay(max_backoff_delay)
.with_max_times(max_retry_attempts);
if backoff_jitter {
backoff_builder = backoff_builder.with_jitter();
}
let backoff = backoff_builder;
let health_check = || {
let mut client = client.clone();
let server = Arc::clone(test_server);
async move {
let result = server.is_healthy(&mut client).await;
match result {
Ok(HealthStatus::Healthy) => {
debug!("🟢 server healthy");
Ok(true)
}
Ok(HealthStatus::Unhealthy) => {
debug!("🟠 server not yet healthy, retrying with exponential backoff");
Err(std::io::Error::new(
std::io::ErrorKind::ConnectionRefused,
"Server not healthy yet",
))
}
Ok(HealthStatus::Uncheckable) => {
debug!("❓wait until a connection can be establish with the server");
let connection = tokio::net::TcpStream::connect(local_addr).await;
if let Err(err) = &connection {
error!(?err, %local_addr, "Oops, fail to establish connection");
}
Ok(connection.is_ok())
}
Err(error) => {
error!(?error, "Health check error");
Ok(false)
}
}
}
};
health_check.retry(&backoff).await.unwrap_or(false)
}
pub async fn write_openapi(mut self, path: impl AsRef<Path>) -> Result<(), TestAppError> {
let path = path.as_ref();
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let openapi = self.client.collected_openapi().await;
let ext = path.extension().unwrap_or_default();
let contents = if ext == "yml" || ext == "yaml" {
openapi.to_yaml().map_err(|err| TestAppError::YamlError {
error: format!("{err:#?}"),
})?
} else {
serde_json::to_string_pretty(&openapi)?
};
fs::write(path, contents)?;
Ok(())
}
}
impl<T> Drop for TestClient<T> {
fn drop(&mut self) {
if let Some(handle) = self.handle.take() {
handle.abort();
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ApiClient;
use std::net::{Ipv4Addr, TcpListener};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::net::TcpListener as TokioTcpListener;
#[derive(Debug)]
struct MockTestServer {
should_be_healthy: Arc<AtomicBool>,
startup_delay: Duration,
custom_config: Option<TestServerConfig>,
}
impl MockTestServer {
fn new() -> Self {
Self {
should_be_healthy: Arc::new(AtomicBool::new(true)),
startup_delay: Duration::from_millis(10),
custom_config: None,
}
}
fn with_health_status(self, healthy: bool) -> Self {
self.should_be_healthy.store(healthy, Ordering::Relaxed);
self
}
fn with_startup_delay(mut self, delay: Duration) -> Self {
self.startup_delay = delay;
self
}
fn with_config(mut self, config: TestServerConfig) -> Self {
self.custom_config = Some(config);
self
}
}
impl TestServer for MockTestServer {
type Error = std::io::Error;
async fn launch(&self, listener: TcpListener) -> Result<(), Self::Error> {
if !self.startup_delay.is_zero() {
tokio::time::sleep(self.startup_delay).await;
}
listener.set_nonblocking(true)?;
let tokio_listener = TokioTcpListener::from_std(listener)?;
loop {
if let Ok((mut stream, _)) = tokio_listener.accept().await {
tokio::spawn(async move {
let response = "HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n";
let _ =
tokio::io::AsyncWriteExt::write_all(&mut stream, response.as_bytes())
.await;
let _ = tokio::io::AsyncWriteExt::shutdown(&mut stream).await;
});
}
}
}
async fn is_healthy(&self, _client: &mut ApiClient) -> Result<HealthStatus, Self::Error> {
Ok(if self.should_be_healthy.load(Ordering::Relaxed) {
HealthStatus::Healthy
} else {
HealthStatus::Unhealthy
})
}
fn config(&self) -> TestServerConfig {
self.custom_config.clone().unwrap_or_default()
}
}
#[tokio::test]
async fn test_test_client_start_success() {
let server = MockTestServer::new();
let result = TestClient::start(server).await;
assert!(result.is_ok());
let test_client = result.unwrap();
assert!(test_client.handle.is_some());
let addr = test_client.local_addr;
assert_eq!(addr.ip(), Ipv4Addr::LOCALHOST);
assert_ne!(addr.port(), 0); }
#[tokio::test]
async fn test_test_client_start_with_custom_config() {
let min_delay = Duration::from_millis(5);
let max_delay = Duration::from_millis(100);
let client_builder = ApiClient::builder()
.with_host("test.example.com")
.with_port(8080);
let config = TestServerConfig {
api_client: Some(client_builder),
min_backoff_delay: min_delay,
max_backoff_delay: max_delay,
backoff_jitter: false,
max_retry_attempts: 5,
};
let server = MockTestServer::new().with_config(config);
let result = TestClient::start(server).await;
assert!(result.is_ok());
let test_client = result.unwrap();
assert!(test_client.handle.is_some());
}
#[tokio::test]
async fn test_test_client_start_unhealthy_server() {
let expected_max_delay = Duration::from_millis(50);
let config = TestServerConfig {
api_client: None,
min_backoff_delay: Duration::from_millis(5),
max_backoff_delay: expected_max_delay,
backoff_jitter: false,
max_retry_attempts: 3,
};
let server = MockTestServer::new()
.with_health_status(false)
.with_config(config);
let result = TestClient::start(server).await;
assert!(result.is_err());
match result.unwrap_err() {
TestAppError::UnhealthyServer {
timeout: actual_timeout,
} => {
assert_eq!(actual_timeout, expected_max_delay);
}
other => panic!("Expected UnhealthyServer error, got: {other:?}"),
}
}
#[tokio::test]
async fn test_test_client_start_slow_server() {
let server = MockTestServer::new().with_startup_delay(Duration::from_millis(50));
let result = TestClient::start(server).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_test_client_deref_to_api_client() {
let server = MockTestServer::new();
let mut test_client = TestClient::start(server)
.await
.expect("client should start");
let openapi = test_client.collected_openapi().await;
assert_eq!(openapi.info.title, ""); }
#[tokio::test]
async fn test_test_client_deref_mut_to_api_client() {
let server = MockTestServer::new();
let test_client = TestClient::start(server)
.await
.expect("client should start");
let result = test_client.get("/test");
assert!(result.is_ok());
}
#[tokio::test]
async fn test_test_client_write_openapi_json() {
let server = MockTestServer::new();
let test_client = TestClient::start(server)
.await
.expect("client should start");
let temp_file = "/tmp/test_openapi.json";
let result = test_client.write_openapi(temp_file).await;
assert!(result.is_ok());
let content = std::fs::read_to_string(temp_file).expect("file should exist");
let json: serde_json::Value = serde_json::from_str(&content).expect("should be valid JSON");
assert!(json.get("openapi").is_some());
assert!(json.get("info").is_some());
let _ = std::fs::remove_file(temp_file);
}
#[tokio::test]
async fn test_test_client_write_openapi_yaml() {
let server = MockTestServer::new();
let test_client = TestClient::start(server)
.await
.expect("client should start");
let temp_file = "/tmp/test_openapi.yml";
let result = test_client.write_openapi(temp_file).await;
assert!(result.is_ok());
let content = std::fs::read_to_string(temp_file).expect("file should exist");
let yaml: serde_json::Value =
serde_saphyr::from_str(&content).expect("should be valid YAML");
assert!(yaml.get("openapi").is_some());
assert!(yaml.get("info").is_some());
let _ = std::fs::remove_file(temp_file);
}
#[tokio::test]
async fn test_test_client_write_openapi_creates_parent_dirs() {
let server = MockTestServer::new();
let test_client = TestClient::start(server)
.await
.expect("client should start");
let temp_dir = "/tmp/test_clawspec_dir/subdir";
let temp_file = format!("{temp_dir}/openapi.json");
let result = test_client.write_openapi(&temp_file).await;
assert!(result.is_ok());
assert!(std::fs::metadata(&temp_file).is_ok());
let _ = std::fs::remove_dir_all("/tmp/test_clawspec_dir");
}
#[tokio::test]
async fn test_test_client_drop_aborts_handle() {
let server = MockTestServer::new();
let test_client = TestClient::start(server)
.await
.expect("client should start");
let handle = test_client.handle.as_ref().unwrap();
assert!(!handle.is_finished());
drop(test_client);
tokio::time::sleep(Duration::from_millis(10)).await;
}
#[test]
fn test_test_client_debug_trait() {
let server = MockTestServer::new();
fn assert_debug<T: std::fmt::Debug>(_: &T) {}
assert_debug(&server);
}
#[test]
fn test_test_client_trait_bounds() {
#[allow(dead_code)]
fn assert_bounds<T>(_: TestClient<T>)
where
T: TestServer + Send + Sync + 'static,
{
}
}
#[derive(Debug)]
struct ErrorTestServer {
error_type: ErrorType,
}
#[derive(Debug)]
enum ErrorType {
#[allow(dead_code)]
BindFailure,
HealthTimeout,
}
impl ErrorTestServer {
#[allow(dead_code)]
fn bind_failure() -> Self {
Self {
error_type: ErrorType::BindFailure,
}
}
fn health_timeout() -> Self {
Self {
error_type: ErrorType::HealthTimeout,
}
}
}
impl TestServer for ErrorTestServer {
type Error = std::io::Error;
async fn launch(&self, listener: TcpListener) -> Result<(), Self::Error> {
match self.error_type {
ErrorType::BindFailure => {
Err(std::io::Error::new(
std::io::ErrorKind::AddrInUse,
"Simulated bind failure",
))
}
ErrorType::HealthTimeout => {
listener.set_nonblocking(true)?;
let _tokio_listener = TokioTcpListener::from_std(listener)?;
loop {
tokio::time::sleep(Duration::from_secs(1)).await;
}
}
}
}
async fn is_healthy(&self, _client: &mut ApiClient) -> Result<HealthStatus, Self::Error> {
match self.error_type {
ErrorType::BindFailure => Ok(HealthStatus::Unhealthy),
ErrorType::HealthTimeout => {
Ok(HealthStatus::Unhealthy)
}
}
}
fn config(&self) -> TestServerConfig {
TestServerConfig {
api_client: None,
min_backoff_delay: Duration::from_millis(1), max_backoff_delay: Duration::from_millis(10), backoff_jitter: false, max_retry_attempts: 3, }
}
}
#[tokio::test]
async fn test_test_client_start_health_timeout() {
let server = ErrorTestServer::health_timeout();
let result = TestClient::start(server).await;
assert!(result.is_err());
match result.unwrap_err() {
TestAppError::UnhealthyServer { timeout } => {
assert_eq!(timeout, Duration::from_millis(10));
}
other => panic!("Expected UnhealthyServer error, got: {other:?}"),
}
}
}