use std::sync::Once;
use dbeel::{
args::{parse_args_from, Args},
error::{Error, Result},
tasks::db_server::ResponseError,
};
use dbeel_client::{self, DbeelClient};
use rmpv::Value;
use rstest::{fixture, rstest};
use serial_test::serial;
use test_utils::{install_logger, test_shard};
const ASSERT_AMOUNT_OF_TIMES: usize = 3;
static ONCE: Once = Once::new();
fn response_equals_error(
response: dbeel_client::error::Error,
error: &Error,
) -> bool {
if let dbeel_client::error::Error::SendRequestToCluster(errors) = response {
let re = ResponseError::new(error);
errors.iter().all(|e| {
if let dbeel_client::error::Error::ServerErr(name, message) = e {
name == &re.name && message == &re.message
} else {
panic!("Expected server error");
}
})
} else {
panic!("Expected cluster request error");
}
}
fn response_ok(value: Value) -> Result<bool> {
Ok(value == Value::String("OK".into()))
}
#[fixture]
fn args() -> Args {
ONCE.call_once(|| {
install_logger();
});
let _ = std::fs::remove_dir_all("/tmp/test");
parse_args_from(["", "--dir", "/tmp/test"])
}
#[rstest]
#[serial]
fn get_non_existing_collection(args: Args) -> Result<()> {
test_shard(args, |shard| async move {
let client = DbeelClient::from_seed_nodes(&[(
shard.args.ip.clone(),
shard.args.port,
)])
.await
.unwrap();
assert!(response_equals_error(
client.collection("test").await.unwrap_err(),
&Error::CollectionNotFound("test".to_string())
));
})?;
Ok(())
}
#[rstest]
#[serial]
fn drop_collection(args: Args) -> Result<()> {
test_shard(args, |shard| async move {
let client = DbeelClient::from_seed_nodes(&[(
shard.args.ip.clone(),
shard.args.port,
)])
.await
.unwrap();
let collection = client.create_collection("test").await.unwrap();
collection.drop().await.unwrap();
assert!(response_equals_error(
client.collection("test").await.unwrap_err(),
&Error::CollectionNotFound("test".to_string())
));
})?;
Ok(())
}
#[rstest]
#[serial]
fn get_non_existing_key(args: Args) -> Result<()> {
test_shard(args, |shard| async move {
let client = DbeelClient::from_seed_nodes(&[(
shard.args.ip.clone(),
shard.args.port,
)])
.await
.unwrap();
let collection = client.create_collection("test").await.unwrap();
let response = collection.get_from_str_key("key").await;
assert!(response_equals_error(
response.unwrap_err(),
&Error::KeyNotFound,
));
})?;
Ok(())
}
#[rstest]
#[serial]
fn set_and_get_key(args: Args) -> Result<()> {
test_shard(args, |shard| async move {
let client = DbeelClient::from_seed_nodes(&[(
shard.args.ip.clone(),
shard.args.port,
)])
.await
.unwrap();
let collection = client.create_collection("test").await.unwrap();
let response = collection
.set_from_str_key("key", Value::F32(100.0))
.await
.unwrap();
assert!(response_ok(response).unwrap());
for _ in 0..ASSERT_AMOUNT_OF_TIMES {
let value = collection.get_from_str_key("key").await.unwrap();
assert_eq!(value, Value::F32(100.0));
}
})?;
Ok(())
}
#[rstest]
#[serial]
fn set_and_get_key_after_restart(args: Args) -> Result<()> {
test_shard(args.clone(), |shard| async move {
let client = DbeelClient::from_seed_nodes(&[(
shard.args.ip.clone(),
shard.args.port,
)])
.await
.unwrap();
let collection = client.create_collection("test").await.unwrap();
let response = collection
.set_from_str_key("key", Value::F32(100.0))
.await
.unwrap();
assert!(response_ok(response).unwrap());
})?;
test_shard(args, |shard| async move {
let client = DbeelClient::from_seed_nodes(&[(
shard.args.ip.clone(),
shard.args.port,
)])
.await
.unwrap();
let collection = client.collection("test").await.unwrap();
for _ in 0..ASSERT_AMOUNT_OF_TIMES {
let value = collection.get_from_str_key("key").await.unwrap();
assert_eq!(value, Value::F32(100.0));
}
})?;
Ok(())
}
#[rstest]
#[serial]
fn delete_and_get_key(args: Args) -> Result<()> {
test_shard(args, |shard| async move {
let client = DbeelClient::from_seed_nodes(&[(
shard.args.ip.clone(),
shard.args.port,
)])
.await
.unwrap();
let collection = client.create_collection("test").await.unwrap();
let response = collection
.set_from_str_key("key", Value::F32(100.0))
.await
.unwrap();
assert!(response_ok(response).unwrap());
let response = collection.delete_from_str_key("key").await.unwrap();
assert!(response_ok(response).unwrap());
for _ in 0..ASSERT_AMOUNT_OF_TIMES {
let response = collection.get_from_str_key("key").await;
assert!(response_equals_error(
response.unwrap_err(),
&Error::KeyNotFound
));
}
})?;
Ok(())
}
#[rstest]
#[serial]
fn multiple_collections(args: Args) -> Result<()> {
test_shard(args, |shard| async move {
let client = DbeelClient::from_seed_nodes(&[(
shard.args.ip.clone(),
shard.args.port,
)])
.await
.unwrap();
let collections = vec![
client.create_collection("test1").await.unwrap(),
client.create_collection("test2").await.unwrap(),
];
for collection in &collections {
let response = collection
.set_from_str_key("key", Value::F32(100.0))
.await
.unwrap();
assert!(response_ok(response).unwrap());
}
for collection in &collections {
let value = collection.get_from_str_key("key").await.unwrap();
assert_eq!(value, Value::F32(100.0));
}
collections[0].delete_from_str_key("key").await.unwrap();
let response = collections[0].get_from_str_key("key").await;
assert!(response_equals_error(
response.unwrap_err(),
&Error::KeyNotFound
));
let value = collections[1].get_from_str_key("key").await.unwrap();
assert_eq!(value, Value::F32(100.0));
collections[1].delete_from_str_key("key").await.unwrap();
for collection in &collections {
let response = collection.get_from_str_key("key").await;
assert!(response_equals_error(
response.unwrap_err(),
&Error::KeyNotFound
));
}
})?;
Ok(())
}