#![forbid(unsafe_code)]
use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use haematite_cli::serve::protocol::ResponseFailure;
use haematite_cli::serve::protocol::{
Cardinality, Change, CollectionWrite, CountReason, DocumentWrite, Edge, EdgeDelete, Fork,
GraphPage, Mutation, Node, Operation, Relation, RelationDelete, RelationTotal, RelationsRead,
Selection, Version,
};
use haematite_cli::serve::service::Service;
use haematite_cli::serve::storage::Storage;
use serde::Deserialize;
use serde_json::{json, value::RawValue};
type TestResult<T = ()> = Result<T, Box<dyn Error>>;
#[derive(Deserialize)]
struct ReadSelection {
branch: String,
version: Version,
}
#[derive(Deserialize)]
struct RelationsAnswer {
selection: ReadSelection,
#[serde(deserialize_with = "relation_totals")]
relations: Vec<RelationTotal>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RelationRow {
relation: String,
from_collection: String,
to_collection: String,
cardinality: Cardinality,
relationships: Option<u64>,
relationships_reason: Option<String>,
}
fn relation_totals<'de, D>(decoder: D) -> Result<Vec<RelationTotal>, D::Error>
where
D: serde::Deserializer<'de>,
{
Vec::<RelationRow>::deserialize(decoder)?
.into_iter()
.map(|row| {
let reason = match row.relationships_reason.as_deref() {
None => None,
Some("withheld") => Some(CountReason::Withheld),
Some("counts_not_ready") => Some(CountReason::CountsNotReady),
Some(reason) => {
return Err(serde::de::Error::custom(format!(
"unknown relationships reason: {reason}"
)));
}
};
Ok(RelationTotal {
declaration: Relation {
relation: row.relation,
from_collection: row.from_collection,
to_collection: row.to_collection,
cardinality: row.cardinality,
},
relationships: row.relationships,
relationships_reason: reason,
})
})
.collect()
}
impl RelationsAnswer {
fn declarations(&self) -> Vec<Relation> {
self.relations
.iter()
.map(|row| row.declaration.clone())
.collect()
}
}
struct Fixture {
service: Service,
directory: tempfile::TempDir,
}
impl Fixture {
fn new() -> TestResult<Self> {
let directory = tempfile::tempdir()?;
let storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 2,
distributed: None,
executor_threads: Some(2),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
long_wait_interval(),
)?;
let fixture = Self {
service: Service::new(
storage,
NonZeroU32::MIN
.checked_add(15)
.ok_or("page limit overflow")?,
)?,
directory,
};
for collection in ["froms", "tos", "other"] {
fixture.apply(
"main",
&format!("collection-{collection}"),
Change::Collection(CollectionWrite {
collection: collection.to_owned(),
schema: None,
}),
)?;
}
for (collection, document_id) in [
("froms", "a"),
("froms", "b"),
("tos", "x"),
("tos", "y"),
("other", "z"),
] {
fixture.apply(
"main",
&format!("document-{collection}-{document_id}"),
Change::Put(DocumentWrite {
collection: collection.to_owned(),
document_id: document_id.to_owned(),
document: RawValue::from_string("{}".to_owned())?,
}),
)?;
}
Ok(fixture)
}
fn relations(&self, branch: &str, version: Option<Version>) -> TestResult<RelationsAnswer> {
Ok(serde_json::from_value(self.service.dispatch(
Operation::Relations(RelationsRead {
selection: selection(branch, version),
}),
)?)?)
}
fn head(&self, branch: &str) -> TestResult<Version> {
Ok(self.relations(branch, None)?.selection.version)
}
fn apply(&self, branch: &str, request_id: &str, change: Change) -> TestResult<Version> {
let result = self.service.dispatch(Operation::Mutate(Mutation {
request_id: request_id.to_owned(),
branch: branch.to_owned(),
expected: self.head(branch)?,
change,
}))?;
Ok(serde_json::from_value(result["version"].clone())?)
}
fn reject(&self, request_id: &str, change: Change) -> TestResult<ResponseFailure> {
let before = self.head("main")?;
let failure = self
.service
.dispatch(Operation::Mutate(Mutation {
request_id: request_id.to_owned(),
branch: "main".to_owned(),
expected: before,
change,
}))
.err()
.ok_or("invalid change was accepted")?;
assert_eq!(self.head("main")?, before);
Ok(failure)
}
fn edges(&self, branch: &str) -> TestResult<Vec<Edge>> {
let answer = self.service.dispatch(Operation::GraphExport(GraphPage {
selection: selection(branch, None),
after: None,
limit: NonZeroU32::MIN
.checked_add(15)
.ok_or("page limit overflow")?,
relations: None,
collections: None,
members: None,
records: None,
unrelated: None,
}))?;
Ok(serde_json::from_value(answer["edges"].clone())?)
}
fn reopen(self) -> TestResult<Self> {
let Self { service, directory } = self;
drop(service);
let storage = Storage::open(
&directory.path().join("database"),
NodeCacheBudget::Unlimited,
long_wait_interval(),
)?;
Ok(Self {
service: Service::new(
storage,
NonZeroU32::MIN
.checked_add(15)
.ok_or("page limit overflow")?,
)?,
directory,
})
}
}
fn selection(branch: &str, version: Option<Version>) -> Selection {
Selection {
branch: branch.to_owned(),
version,
}
}
fn relation(cardinality: Cardinality) -> Relation {
Relation {
relation: "links".to_owned(),
from_collection: "froms".to_owned(),
to_collection: "tos".to_owned(),
cardinality,
}
}
fn node(collection: &str, document_id: &str) -> Node {
Node {
collection: collection.to_owned(),
document_id: document_id.to_owned(),
}
}
fn edge(from: &str, to: &str) -> Edge {
Edge {
relation: "links".to_owned(),
from: node("froms", from),
to: node("tos", to),
properties: json!({"weight":1}),
}
}
#[test]
fn relation_wire_shapes_match_the_studio_and_refuse_unknown_fields() -> TestResult {
let declaration = relation(Cardinality::OneToMany);
let value = serde_json::to_value(Change::RelationPut(declaration.clone()))?;
assert_eq!(
value,
json!({"kind":"relation.put","value":{
"relation":"links","from_collection":"froms","to_collection":"tos","cardinality":"one_to_many"
}})
);
let decoded: Change = serde_json::from_value(value)?;
let Change::RelationPut(decoded) = decoded else {
return Err("relation.put lost its variant".into());
};
assert_eq!(decoded, declaration);
assert_eq!(
serde_json::to_value(Change::RelationDelete(RelationDelete {
relation: "links".to_owned()
}))?,
json!({"kind":"relation.delete","value":{"relation":"links"}})
);
assert_eq!(
serde_json::to_value(Operation::Relations(RelationsRead {
selection: selection("main", None)
}))?,
json!({"method":"relations.list","params":{"selection":{"branch":"main","version":null}}})
);
let read: Operation = serde_json::from_value(json!({
"method":"relations.list", "params":{"selection":{"branch":"main"}}
}))?;
let Operation::Relations(read) = read else {
return Err("relations.list lost its variant".into());
};
assert!(read.selection.version.is_none());
for cardinality in ["one_to_one", "one_to_many", "many_to_many"] {
let declaration: Relation = serde_json::from_value(json!({
"relation":"links", "from_collection":"froms", "to_collection":"tos", "cardinality":cardinality,
}))?;
assert_eq!(
serde_json::to_value(declaration)?["cardinality"],
cardinality
);
}
for invalid in [
json!({"relation":"links","from_collection":"froms","to_collection":"tos","cardinality":"one"}),
json!({"relation":"links","from_collection":"froms","to_collection":"tos","cardinality":"one_to_one","extra":true}),
] {
assert!(serde_json::from_value::<Relation>(invalid).is_err());
}
Ok(())
}
#[test]
fn declarations_list_in_order_survive_reopen_and_select_the_old_version() -> TestResult {
let fixture = Fixture::new()?;
let empty = fixture.head("main")?;
let mut z = relation(Cardinality::ManyToMany);
z.relation = "z".to_owned();
fixture.apply("main", "z", Change::RelationPut(z.clone()))?;
let a = relation(Cardinality::OneToOne);
let written = fixture.apply("main", "a", Change::RelationPut(a.clone()))?;
let answer = fixture.relations("main", None)?;
assert_eq!(answer.selection.branch, "main");
assert_eq!(answer.selection.version, written);
assert_eq!(answer.declarations(), vec![a.clone(), z.clone()]);
for row in &answer.relations {
assert_eq!(row.relationships, Some(0));
assert_eq!(row.relationships_reason, None);
}
let wire = fixture
.service
.dispatch(Operation::Relations(RelationsRead {
selection: selection("main", Some(written)),
}))?;
let fields: std::collections::BTreeSet<_> = wire
.as_object()
.ok_or("relation answer is not an object")?
.keys()
.map(String::as_str)
.collect();
assert_eq!(fields, ["relations", "selection"].into_iter().collect());
assert_eq!(
wire["selection"]["roots"]
.as_array()
.ok_or("captured roots are missing")?
.len(),
2
);
assert!(fixture.relations("main", Some(empty))?.relations.is_empty());
let fixture = fixture.reopen()?;
assert_eq!(fixture.relations("main", None)?.declarations(), vec![a, z]);
let failure = fixture
.service
.dispatch_bounded(
Operation::Relations(RelationsRead {
selection: selection("main", None),
}),
NonZeroUsize::MIN,
)
.err()
.ok_or("relation list exceeded its budget")?;
assert_eq!(failure.code, "response_limit");
Ok(())
}
#[test]
fn declaration_identifiers_and_both_missing_collections_are_refused() -> TestResult {
let fixture = Fixture::new()?;
for (index, field) in ["from_collection", "to_collection"].into_iter().enumerate() {
let mut declaration = relation(Cardinality::ManyToMany);
if index == 0 {
declaration.from_collection = "missing".to_owned();
} else {
declaration.to_collection = "missing".to_owned();
}
let failure = fixture.reject(field, Change::RelationPut(declaration))?;
assert_eq!(failure.code, "collection_missing");
assert_eq!(failure.path.as_deref(), Some(format!("/{field}").as_str()));
assert!(failure.message.contains("missing"));
}
for index in 0..3 {
let mut declaration = relation(Cardinality::ManyToMany);
match index {
0 => declaration.relation.clear(),
1 => declaration.from_collection = "bad\0name".to_owned(),
_ => declaration.to_collection = "x".repeat(513),
}
assert_eq!(
fixture
.reject(
&format!("invalid-{index}"),
Change::RelationPut(declaration)
)?
.code,
"invalid_identifier"
);
}
assert_eq!(
fixture
.reject(
"delete-invalid",
Change::RelationDelete(RelationDelete {
relation: "bad\0name".to_owned()
})
)?
.code,
"invalid_identifier"
);
assert!(fixture.relations("main", None)?.relations.is_empty());
Ok(())
}
#[test]
fn declared_collection_constraints_name_both_wrong_ends() -> TestResult {
let fixture = Fixture::new()?;
fixture.apply(
"main",
"declare",
Change::RelationPut(relation(Cardinality::ManyToMany)),
)?;
for (side, wrong_from) in [("from", true), ("to", false)] {
let mut offered = edge("a", "x");
if wrong_from {
offered.from = node("other", "z");
} else {
offered.to = node("other", "z");
}
let failure = fixture.reject(side, Change::EdgePut(offered))?;
assert_eq!(failure.code, "relation_collection");
assert_eq!(
failure.path.as_deref(),
Some(format!("/{side}/collection").as_str())
);
assert!(failure.message.contains("links"));
assert!(failure.message.contains("other/z"));
}
assert!(fixture.edges("main")?.is_empty());
Ok(())
}
#[test]
fn one_to_one_allows_disjoint_pairs_and_replacement_but_refuses_each_shared_end() -> TestResult {
let fixture = Fixture::new()?;
fixture.apply(
"main",
"declare",
Change::RelationPut(relation(Cardinality::OneToOne)),
)?;
let mut original = edge("a", "x");
fixture.apply("main", "edge", Change::EdgePut(original.clone()))?;
for (id, offered) in [
("shared-from", edge("a", "y")),
("shared-to", edge("b", "x")),
] {
let failure = fixture.reject(id, Change::EdgePut(offered))?;
assert_eq!(failure.code, "relation_cardinality");
assert!(failure.message.contains("froms/a -> tos/x"));
assert!(failure.message.contains("links"));
}
original.properties = json!({"weight":2});
fixture.apply("main", "replace", Change::EdgePut(original.clone()))?;
let disjoint = edge("b", "y");
fixture.apply("main", "disjoint", Change::EdgePut(disjoint.clone()))?;
assert_eq!(fixture.edges("main")?, vec![original.clone(), disjoint]);
fixture.apply(
"main",
"remove",
Change::EdgeDelete(EdgeDelete {
relation: original.relation,
from: original.from,
to: original.to,
}),
)?;
fixture.apply("main", "freed", Change::EdgePut(edge("a", "x")))?;
assert_eq!(fixture.edges("main")?.len(), 2);
Ok(())
}
#[test]
fn one_to_many_allows_shared_from_and_replacement_but_refuses_shared_to() -> TestResult {
let fixture = Fixture::new()?;
fixture.apply(
"main",
"declare",
Change::RelationPut(relation(Cardinality::OneToMany)),
)?;
let mut first = edge("a", "x");
fixture.apply("main", "first", Change::EdgePut(first.clone()))?;
let second = edge("a", "y");
fixture.apply("main", "second", Change::EdgePut(second.clone()))?;
let failure = fixture.reject("shared-to", Change::EdgePut(edge("b", "x")))?;
assert_eq!(failure.code, "relation_cardinality");
assert!(failure.message.contains("froms/a -> tos/x"));
first.properties = json!({"weight":3});
fixture.apply("main", "replace", Change::EdgePut(first.clone()))?;
assert_eq!(fixture.edges("main")?, vec![first, second]);
Ok(())
}
#[test]
fn many_to_many_allows_each_shared_end_and_replacement() -> TestResult {
let fixture = Fixture::new()?;
fixture.apply(
"main",
"declare",
Change::RelationPut(relation(Cardinality::ManyToMany)),
)?;
let mut first = edge("a", "x");
let others = [edge("a", "y"), edge("b", "x")];
fixture.apply("main", "first", Change::EdgePut(first.clone()))?;
for (index, offered) in others.iter().enumerate() {
fixture.apply(
"main",
&format!("other-{index}"),
Change::EdgePut(offered.clone()),
)?;
}
first.properties = json!({"weight":4});
fixture.apply("main", "replace", Change::EdgePut(first.clone()))?;
assert_eq!(
fixture.edges("main")?,
vec![first, others[0].clone(), others[1].clone()]
);
Ok(())
}
#[test]
fn constraints_apply_only_to_edges_under_the_declared_name() -> TestResult {
let fixture = Fixture::new()?;
let mut other_first = edge("a", "x");
other_first.relation = "other".to_owned();
let mut other_second = edge("a", "y");
other_second.relation = "other".to_owned();
fixture.apply("main", "other-first", Change::EdgePut(other_first))?;
fixture.apply("main", "other-second", Change::EdgePut(other_second))?;
fixture.apply(
"main",
"declare",
Change::RelationPut(relation(Cardinality::OneToOne)),
)?;
fixture.apply("main", "first", Change::EdgePut(edge("a", "x")))?;
fixture.apply("main", "second", Change::EdgePut(edge("b", "y")))?;
let edges = fixture.edges("main")?;
assert_eq!(edges.len(), 4);
assert_eq!(
edges.iter().filter(|edge| edge.relation == "links").count(),
2
);
assert_eq!(
edges.iter().filter(|edge| edge.relation == "other").count(),
2
);
Ok(())
}
#[test]
fn self_edges_in_a_same_collection_relation_can_be_replaced() -> TestResult {
let fixture = Fixture::new()?;
let mut declaration = relation(Cardinality::OneToOne);
declaration.to_collection = "froms".to_owned();
fixture.apply("main", "declare", Change::RelationPut(declaration))?;
let mut offered = edge("a", "x");
offered.to = offered.from.clone();
fixture.apply("main", "self", Change::EdgePut(offered.clone()))?;
offered.properties = json!({"weight":5});
fixture.apply("main", "replace", Change::EdgePut(offered.clone()))?;
assert_eq!(fixture.edges("main")?, vec![offered.clone()]);
let mut conflict = offered;
conflict.to.document_id = "b".to_owned();
assert_eq!(
fixture.reject("conflict", Change::EdgePut(conflict))?.code,
"relation_cardinality"
);
Ok(())
}
#[test]
fn deleting_a_declaration_leaves_edges_and_makes_the_relation_free() -> TestResult {
let fixture = Fixture::new()?;
fixture.apply(
"main",
"declare",
Change::RelationPut(relation(Cardinality::OneToOne)),
)?;
let first = edge("a", "x");
fixture.apply("main", "edge", Change::EdgePut(first.clone()))?;
let declared_version = fixture.head("main")?;
fixture.apply(
"main",
"delete",
Change::RelationDelete(RelationDelete {
relation: "links".to_owned(),
}),
)?;
assert!(fixture.relations("main", None)?.relations.is_empty());
assert_eq!(
fixture
.relations("main", Some(declared_version))?
.declarations(),
vec![relation(Cardinality::OneToOne)]
);
assert_eq!(fixture.edges("main")?, vec![first]);
fixture.apply("main", "shared", Change::EdgePut(edge("a", "y")))?;
let mut unrestricted = edge("b", "x");
unrestricted.from = node("other", "z");
fixture.apply("main", "wrong-kind-now-free", Change::EdgePut(unrestricted))?;
assert_eq!(fixture.edges("main")?.len(), 3);
fixture.apply(
"main",
"delete-absent",
Change::RelationDelete(RelationDelete {
relation: "links".to_owned(),
}),
)?;
assert_eq!(fixture.edges("main")?.len(), 3);
Ok(())
}
#[test]
fn declaring_over_free_edges_and_tightening_a_declaration_check_all_existing_edges() -> TestResult {
let fixture = Fixture::new()?;
fixture.apply("main", "a-x", Change::EdgePut(edge("a", "x")))?;
fixture.apply("main", "a-y", Change::EdgePut(edge("a", "y")))?;
let failure = fixture.reject(
"free-to-one",
Change::RelationPut(relation(Cardinality::OneToOne)),
)?;
assert_eq!(failure.code, "relation_cardinality");
assert!(failure.message.contains("froms/a -> tos/y"));
assert!(fixture.relations("main", None)?.relations.is_empty());
fixture.apply(
"main",
"many",
Change::RelationPut(relation(Cardinality::ManyToMany)),
)?;
fixture.apply(
"main",
"tighten-valid",
Change::RelationPut(relation(Cardinality::OneToMany)),
)?;
let failure = fixture.reject(
"tighten-invalid",
Change::RelationPut(relation(Cardinality::OneToOne)),
)?;
assert_eq!(failure.code, "relation_cardinality");
assert!(failure.message.contains("froms/a -> tos/y"));
assert_eq!(
fixture.relations("main", None)?.declarations(),
vec![relation(Cardinality::OneToMany)]
);
let mut wrong = relation(Cardinality::ManyToMany);
wrong.from_collection = "other".to_owned();
let failure = fixture.reject("change-from", Change::RelationPut(wrong))?;
assert_eq!(failure.code, "relation_collection");
assert!(failure.message.contains("froms/a -> tos/x"));
let mut wrong = relation(Cardinality::ManyToMany);
wrong.to_collection = "other".to_owned();
let failure = fixture.reject("change-to", Change::RelationPut(wrong))?;
assert_eq!(failure.code, "relation_collection");
assert!(failure.message.contains("froms/a -> tos/x"));
fixture.apply(
"main",
"loosen",
Change::RelationPut(relation(Cardinality::ManyToMany)),
)?;
fixture.apply("main", "b-x", Change::EdgePut(edge("b", "x")))?;
let failure = fixture.reject(
"shared-to-tighten",
Change::RelationPut(relation(Cardinality::OneToMany)),
)?;
assert_eq!(failure.code, "relation_cardinality");
assert!(failure.message.contains("froms/b -> tos/x"));
assert_eq!(fixture.edges("main")?.len(), 3);
Ok(())
}
#[test]
fn a_fork_has_its_own_declaration_and_source_edges_stay_free() -> TestResult {
let fixture = Fixture::new()?;
let from = fixture.head("main")?;
fixture.apply(
"main",
"fork",
Change::Fork(Fork {
name: "draft".to_owned(),
from,
}),
)?;
let declaration = relation(Cardinality::OneToOne);
fixture.apply("draft", "declare", Change::RelationPut(declaration.clone()))?;
assert!(fixture.relations("main", None)?.relations.is_empty());
let draft = fixture.relations("draft", None)?;
assert_eq!(draft.selection.branch, "draft");
assert_eq!(draft.declarations(), vec![declaration]);
fixture.apply("main", "source-first", Change::EdgePut(edge("a", "x")))?;
fixture.apply("main", "source-second", Change::EdgePut(edge("a", "y")))?;
assert!(fixture.edges("draft")?.is_empty());
fixture.apply("draft", "draft-first", Change::EdgePut(edge("a", "x")))?;
let failure = fixture
.service
.dispatch(Operation::Mutate(Mutation {
request_id: "draft-second".to_owned(),
branch: "draft".to_owned(),
expected: fixture.head("draft")?,
change: Change::EdgePut(edge("a", "y")),
}))
.err()
.ok_or("fork cardinality was not enforced")?;
assert_eq!(failure.code, "relation_cardinality");
assert_eq!(fixture.edges("main")?.len(), 2);
assert_eq!(fixture.edges("draft")?.len(), 1);
Ok(())
}
const fn long_wait_interval() -> std::time::Duration {
const LONG_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
LONG_WAIT_INTERVAL
}