use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json};
use super::Service;
use crate::serve::protocol::{
Change, CollectionWrite, Fork, Mutation, Operation, ReceiptQuery, Selection, Version,
};
use crate::serve::storage::Storage;
type TestResult = Result<(), Box<dyn Error>>;
fn fixture() -> Result<(tempfile::TempDir, Service, Version), Box<dyn Error>> {
let directory = tempfile::tempdir()?;
let storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 1,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
let version = storage
.capture(&Selection {
branch: "main".to_owned(),
version: None,
})?
.version;
Ok((directory, Service::new(storage, NonZeroU32::MIN)?, version))
}
fn mutation(expected: Version, request_id: &str) -> Mutation {
Mutation {
request_id: request_id.to_owned(),
branch: "main".to_owned(),
expected,
change: Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
}
}
fn assert_context(error: &Value, outcome: &str, request: &str, branch: &str) {
assert_eq!(error["outcome"], outcome);
assert_eq!(error["request_id"], request);
assert_eq!(error["receipt_branch"], branch);
}
fn receipt(service: &Service, request_id: &str, branch: &str) -> Result<Value, Box<dyn Error>> {
Ok(service.dispatch(Operation::Receipt(ReceiptQuery {
request_id: request_id.to_owned(),
branch: branch.to_owned(),
}))?)
}
#[test]
fn mutation_refusal_before_write_says_not_applied() -> TestResult {
let (_directory, service, version) = fixture()?;
let mut stale = version;
stale.sequence += 1;
let error = service
.dispatch(Operation::Mutate(mutation(stale, "refused")))
.err()
.ok_or("stale mutation was applied")?;
let error = serde_json::to_value(error)?;
assert_eq!(error["code"], "version_conflict");
assert_context(&error, "not_applied", "refused", "main");
assert!(receipt(&service, "refused", "main")?.is_null());
Ok(())
}
#[test]
fn mutation_lost_commit_answer_keeps_unknown_identity() -> TestResult {
let (_directory, service, version) = fixture()?;
Storage::lose_next_mutation_answer();
let error = service
.dispatch(Operation::Mutate(mutation(version, "lost")))
.err()
.ok_or("commit answer was not lost")?;
let error = serde_json::to_value(error)?;
assert_eq!(error["code"], "durability_unknown");
assert_context(&error, "unknown", "lost", "main");
let found = receipt(&service, "lost", "main")?;
assert_eq!(found["request_id"], "lost");
assert_eq!(
found["version"]["sequence"],
(version.sequence + 1).to_string()
);
Ok(())
}
#[test]
fn fork_lost_commit_answer_names_destination_receipt() -> TestResult {
let (_directory, service, version) = fixture()?;
let mut request = mutation(version, "fork-lost");
request.change = Change::Fork(Fork {
name: "child".to_owned(),
from: version,
});
Storage::lose_next_mutation_answer();
let error = service
.dispatch(Operation::Mutate(request))
.err()
.ok_or("fork answer was not lost")?;
let error = serde_json::to_value(error)?;
assert_eq!(error["code"], "durability_unknown");
assert_context(&error, "unknown", "fork-lost", "child");
assert!(receipt(&service, "fork-lost", "main")?.is_null());
let found = receipt(&service, "fork-lost", "child")?;
assert_eq!(found["branch"], "child");
assert_eq!(found["request_id"], "fork-lost");
Ok(())
}
#[test]
fn mutation_late_result_limit_keeps_unknown_identity() -> TestResult {
let (_directory, service, version) = fixture()?;
let error = service
.dispatch_bounded(
Operation::Mutate(mutation(version, "limited")),
NonZeroUsize::MIN,
)
.err()
.ok_or("receipt exceeded no result limit")?;
let error = serde_json::to_value(error)?;
assert_eq!(error["code"], "response_limit");
assert_context(&error, "unknown", "limited", "main");
assert_eq!(
receipt(&service, "limited", "main")?["request_id"],
json!("limited")
);
Ok(())
}
#[test]
fn unrecognized_mutation_error_is_unknown_and_other_verbs_keep_their_shape() -> TestResult {
use crate::serve::protocol::{MutationContext, ResponseFailure};
use crate::serve::storage::Failure;
let operation = Operation::Mutate(mutation(
Version {
generation: 1,
sequence: 0,
},
"unknown-code",
));
let original = Failure::new("new_backend_failure", "original refusal reason");
let mutation_error = ResponseFailure::new(
original.clone(),
MutationContext::from_operation(&operation),
);
let error = serde_json::to_value(mutation_error)?;
assert_context(&error, "unknown", "unknown-code", "main");
assert_eq!(error["code"], "new_backend_failure");
assert_eq!(error["message"], "original refusal reason");
assert_eq!(
serde_json::to_value(ResponseFailure::new(original.clone(), None))?,
serde_json::to_value(original)?
);
Ok(())
}
const fn fixture_long_wait_interval() -> std::time::Duration {
std::time::Duration::from_secs(1)
}
#[test]
fn batch_validation_codes_say_not_applied_and_keep_request_identity() -> TestResult {
use crate::serve::protocol::{MutationContext, ResponseFailure};
use crate::serve::storage::Failure;
let operation = Operation::Mutate(mutation(
Version {
generation: 1,
sequence: 0,
},
"batch-refused",
));
for code in [
"document_missing",
"relation_missing",
"relation_collection",
"relation_cardinality",
"edge_properties",
"vector_dimension_missing",
"vector_dimension",
"vector_empty",
"vector_non_finite",
"import_empty",
"import_duplicate",
"import_limit",
"import_limit_required",
] {
let failure = ResponseFailure::new(
Failure::new(code, "batch validation refusal"),
MutationContext::from_operation(&operation),
);
let value = serde_json::to_value(failure)?;
assert_eq!(value["code"], code);
assert_context(&value, "not_applied", "batch-refused", "main");
}
Ok(())
}