245 lines
9.1 KiB
Rust
245 lines
9.1 KiB
Rust
use herodb::{server::Server, options::DBOption};
|
|
use std::path::PathBuf;
|
|
use std::time::Duration;
|
|
use tokio::time::sleep;
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
|
use tokio::net::TcpStream;
|
|
|
|
// Helper function to start a test server with clean data directory
|
|
async fn start_test_server(test_name: &str) -> (Server, u16) {
|
|
use std::sync::atomic::{AtomicU16, Ordering};
|
|
static PORT_COUNTER: AtomicU16 = AtomicU16::new(17000);
|
|
|
|
// Get a unique port for this test
|
|
let port = PORT_COUNTER.fetch_add(1, Ordering::SeqCst);
|
|
|
|
let test_dir = format!("/tmp/herodb_test_{}", test_name);
|
|
|
|
// Clean up any existing test data
|
|
let _ = std::fs::remove_dir_all(&test_dir);
|
|
std::fs::create_dir_all(&test_dir).unwrap();
|
|
|
|
let option = DBOption {
|
|
dir: PathBuf::from(test_dir),
|
|
port,
|
|
debug: true,
|
|
encrypt: false,
|
|
encryption_key: None,
|
|
backend: herodb::options::BackendType::Redb,
|
|
admin_secret: "test-admin".to_string(),
|
|
};
|
|
|
|
let server = Server::new(option).await;
|
|
(server, port)
|
|
}
|
|
|
|
// Helper function to send Redis command and get response
|
|
async fn send_redis_command(port: u16, command: &str) -> String {
|
|
let mut stream = TcpStream::connect(format!("127.0.0.1:{}", port)).await.unwrap();
|
|
|
|
// Acquire ReadWrite permissions on this new connection
|
|
let handshake = "*4\r\n$6\r\nSELECT\r\n$1\r\n0\r\n$3\r\nKEY\r\n$10\r\ntest-admin\r\n";
|
|
stream.write_all(handshake.as_bytes()).await.unwrap();
|
|
let mut buffer = [0; 1024];
|
|
let _ = stream.read(&mut buffer).await.unwrap(); // Read and ignore the OK for handshake
|
|
|
|
// Now send the intended command
|
|
stream.write_all(command.as_bytes()).await.unwrap();
|
|
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
String::from_utf8_lossy(&buffer[..n]).to_string()
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_basic_redis_functionality() {
|
|
let (mut server, port) = start_test_server("basic").await;
|
|
|
|
// Start server in background with timeout
|
|
let server_handle = tokio::spawn(async move {
|
|
let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
|
|
.await
|
|
.unwrap();
|
|
|
|
// Accept only a few connections for testing
|
|
for _ in 0..10 {
|
|
if let Ok((stream, _)) = listener.accept().await {
|
|
let _ = server.handle(stream).await;
|
|
}
|
|
}
|
|
});
|
|
|
|
sleep(Duration::from_millis(100)).await;
|
|
|
|
// Test PING
|
|
let response = send_redis_command(port, "*1\r\n$4\r\nPING\r\n").await;
|
|
assert!(response.contains("PONG"));
|
|
|
|
// Test SET
|
|
let response = send_redis_command(port, "*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n").await;
|
|
assert!(response.contains("OK"));
|
|
|
|
// Test GET
|
|
let response = send_redis_command(port, "*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n").await;
|
|
assert!(response.contains("value"));
|
|
|
|
// Test HSET
|
|
let response = send_redis_command(port, "*4\r\n$4\r\nHSET\r\n$4\r\nhash\r\n$5\r\nfield\r\n$5\r\nvalue\r\n").await;
|
|
assert!(response.contains("1"));
|
|
|
|
// Test HGET
|
|
let response = send_redis_command(port, "*3\r\n$4\r\nHGET\r\n$4\r\nhash\r\n$5\r\nfield\r\n").await;
|
|
assert!(response.contains("value"));
|
|
|
|
// Test EXISTS
|
|
let response = send_redis_command(port, "*2\r\n$6\r\nEXISTS\r\n$3\r\nkey\r\n").await;
|
|
assert!(response.contains("1"));
|
|
|
|
// Test TTL
|
|
let response = send_redis_command(port, "*2\r\n$3\r\nTTL\r\n$3\r\nkey\r\n").await;
|
|
assert!(response.contains("-1")); // No expiration
|
|
|
|
// Test TYPE
|
|
let response = send_redis_command(port, "*2\r\n$4\r\nTYPE\r\n$3\r\nkey\r\n").await;
|
|
assert!(response.contains("string"));
|
|
|
|
// Test QUIT to close connection gracefully
|
|
let mut stream = TcpStream::connect(format!("127.0.0.1:{}", port)).await.unwrap();
|
|
stream.write_all("*1\r\n$4\r\nQUIT\r\n".as_bytes()).await.unwrap();
|
|
let mut buffer = [0; 1024];
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
let response = String::from_utf8_lossy(&buffer[..n]);
|
|
assert!(response.contains("OK"));
|
|
|
|
// Ensure the stream is closed
|
|
stream.shutdown().await.unwrap();
|
|
|
|
// Stop the server
|
|
server_handle.abort();
|
|
|
|
println!("✅ All basic Redis functionality tests passed!");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_hash_operations() {
|
|
let (mut server, port) = start_test_server("hash_ops").await;
|
|
|
|
// Start server in background with timeout
|
|
let server_handle = tokio::spawn(async move {
|
|
let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
|
|
.await
|
|
.unwrap();
|
|
|
|
// Accept only a few connections for testing
|
|
for _ in 0..5 {
|
|
if let Ok((stream, _)) = listener.accept().await {
|
|
let _ = server.handle(stream).await;
|
|
}
|
|
}
|
|
});
|
|
|
|
sleep(Duration::from_millis(100)).await;
|
|
|
|
// Test HSET multiple fields
|
|
let response = send_redis_command(port, "*6\r\n$4\r\nHSET\r\n$4\r\nhash\r\n$6\r\nfield1\r\n$6\r\nvalue1\r\n$6\r\nfield2\r\n$6\r\nvalue2\r\n").await;
|
|
assert!(response.contains("2")); // 2 new fields
|
|
|
|
// Test HGETALL
|
|
let response = send_redis_command(port, "*2\r\n$7\r\nHGETALL\r\n$4\r\nhash\r\n").await;
|
|
assert!(response.contains("field1"));
|
|
assert!(response.contains("value1"));
|
|
assert!(response.contains("field2"));
|
|
assert!(response.contains("value2"));
|
|
|
|
// Test HEXISTS
|
|
let response = send_redis_command(port, "*3\r\n$7\r\nHEXISTS\r\n$4\r\nhash\r\n$6\r\nfield1\r\n").await;
|
|
assert!(response.contains("1"));
|
|
|
|
// Test HLEN
|
|
let response = send_redis_command(port, "*2\r\n$4\r\nHLEN\r\n$4\r\nhash\r\n").await;
|
|
assert!(response.contains("2"));
|
|
|
|
// Test HSCAN
|
|
let response = send_redis_command(port, "*7\r\n$5\r\nHSCAN\r\n$4\r\nhash\r\n$1\r\n0\r\n$5\r\nMATCH\r\n$1\r\n*\r\n$5\r\nCOUNT\r\n$2\r\n10\r\n").await;
|
|
assert!(response.contains("field1"));
|
|
assert!(response.contains("value1"));
|
|
assert!(response.contains("field2"));
|
|
assert!(response.contains("value2"));
|
|
|
|
// Stop the server
|
|
// For hash operations, we don't have a persistent stream, so we'll just abort the server.
|
|
// The server should handle closing its connections.
|
|
server_handle.abort();
|
|
|
|
println!("✅ All hash operations tests passed!");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_transaction_operations() {
|
|
let (mut server, port) = start_test_server("transactions").await;
|
|
|
|
// Start server in background with timeout
|
|
let server_handle = tokio::spawn(async move {
|
|
let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
|
|
.await
|
|
.unwrap();
|
|
|
|
// Accept only a few connections for testing
|
|
for _ in 0..5 {
|
|
if let Ok((stream, _)) = listener.accept().await {
|
|
let _ = server.handle(stream).await;
|
|
}
|
|
}
|
|
});
|
|
|
|
sleep(Duration::from_millis(100)).await;
|
|
|
|
// Use a single connection for the transaction
|
|
let mut stream = TcpStream::connect(format!("127.0.0.1:{}", port)).await.unwrap();
|
|
|
|
// Acquire write permissions for this connection
|
|
let handshake = "*4\r\n$6\r\nSELECT\r\n$1\r\n0\r\n$3\r\nKEY\r\n$10\r\ntest-admin\r\n";
|
|
stream.write_all(handshake.as_bytes()).await.unwrap();
|
|
let mut buffer = [0; 1024];
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
let resp = String::from_utf8_lossy(&buffer[..n]);
|
|
assert!(resp.contains("OK"));
|
|
|
|
// Test MULTI
|
|
stream.write_all("*1\r\n$5\r\nMULTI\r\n".as_bytes()).await.unwrap();
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
let response = String::from_utf8_lossy(&buffer[..n]);
|
|
assert!(response.contains("OK"));
|
|
|
|
// Test queued commands
|
|
stream.write_all("*3\r\n$3\r\nSET\r\n$4\r\nkey1\r\n$6\r\nvalue1\r\n".as_bytes()).await.unwrap();
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
let response = String::from_utf8_lossy(&buffer[..n]);
|
|
assert!(response.contains("QUEUED"));
|
|
|
|
stream.write_all("*3\r\n$3\r\nSET\r\n$4\r\nkey2\r\n$6\r\nvalue2\r\n".as_bytes()).await.unwrap();
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
let response = String::from_utf8_lossy(&buffer[..n]);
|
|
assert!(response.contains("QUEUED"));
|
|
|
|
// Test EXEC
|
|
stream.write_all("*1\r\n$4\r\nEXEC\r\n".as_bytes()).await.unwrap();
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
let response = String::from_utf8_lossy(&buffer[..n]);
|
|
assert!(response.contains("OK")); // Should contain array of OK responses
|
|
|
|
// Verify commands were executed
|
|
stream.write_all("*2\r\n$3\r\nGET\r\n$4\r\nkey1\r\n".as_bytes()).await.unwrap();
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
let response = String::from_utf8_lossy(&buffer[..n]);
|
|
assert!(response.contains("value1"));
|
|
|
|
stream.write_all("*2\r\n$3\r\nGET\r\n$4\r\nkey2\r\n".as_bytes()).await.unwrap();
|
|
let n = stream.read(&mut buffer).await.unwrap();
|
|
let response = String::from_utf8_lossy(&buffer[..n]);
|
|
assert!(response.contains("value2"));
|
|
|
|
// Stop the server
|
|
server_handle.abort();
|
|
|
|
println!("✅ All transaction operations tests passed!");
|
|
} |