#![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::{
Change, CollectionWrite, DocumentWrite, Edge, EdgeDelete, EdgePage, Fork, GraphPage, Mutation,
Node, Operation, Selection, Version,
};
use haematite_cli::serve::service::Service;
use haematite_cli::serve::storage::Storage;
use serde::Deserialize;
use serde_json::{Value, json, value::RawValue};
type TestResult<T = ()> = Result<T, Box<dyn Error>>;
#[derive(Deserialize)]
struct ReadSelection {
branch: String,
version: Version,
}
#[derive(Deserialize)]
struct Answer {
selection: ReadSelection,
edges: Vec<Edge>,
next_after: Option<String>,
}
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 service = Service::new(storage, limit(16)?)?;
let fixture = Self { service, directory };
for collection in ["records", "other"] {
fixture.apply(
"main",
&format!("declare-{collection}"),
Change::Collection(CollectionWrite {
collection: collection.to_owned(),
schema: None,
}),
)?;
}
for (collection, id) in [
("records", "a"),
("records", "b"),
("records", "c"),
("other", "a"),
] {
fixture.apply(
"main",
&format!("document-{collection}-{id}"),
Change::Put(DocumentWrite {
collection: collection.to_owned(),
document_id: id.to_owned(),
document: RawValue::from_string(serde_json::to_string(&json!({"name":id}))?)?,
}),
)?;
}
Ok(fixture)
}
fn export(&self, branch: &str) -> TestResult<Answer> {
graph(&self.service, selection(branch, None), None, 16)
}
fn apply(&self, branch: &str, id: &str, change: Change) -> TestResult<Version> {
let expected = self.export(branch)?.selection.version;
let result = self.service.dispatch(Operation::Mutate(Mutation {
request_id: id.to_owned(),
branch: branch.to_owned(),
expected,
change,
}))?;
Ok(serde_json::from_value(result["version"].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, limit(16)?)?,
directory,
})
}
}
fn limit(value: u32) -> TestResult<NonZeroU32> {
Ok(NonZeroU32::new(value).ok_or("page limit is zero")?)
}
fn selection(branch: &str, version: Option<Version>) -> Selection {
Selection {
branch: branch.to_owned(),
version,
}
}
fn node(collection: &str, document_id: &str) -> Node {
Node {
collection: collection.to_owned(),
document_id: document_id.to_owned(),
}
}
fn edge(from: &str, relation: &str, to: &str) -> Edge {
Edge {
from: node("records", from),
to: node("records", to),
relation: relation.to_owned(),
properties: json!({"weight":1}),
}
}
fn graph(service: &Service, at: Selection, after: Option<String>, rows: u32) -> TestResult<Answer> {
Ok(serde_json::from_value(service.dispatch(
Operation::GraphExport(GraphPage {
selection: at,
after,
limit: limit(rows)?,
relations: None,
collections: None,
members: None,
records: None,
unrelated: None,
}),
)?)?)
}
fn at_node(
service: &Service,
incoming: bool,
at: Selection,
node: Node,
relation: Option<&str>,
after: Option<String>,
rows: u32,
) -> TestResult<Answer> {
let page = EdgePage {
selection: at,
node,
relation: relation.map(str::to_owned),
after,
limit: limit(rows)?,
};
let operation = if incoming {
Operation::EdgesIn(page)
} else {
Operation::EdgesOut(page)
};
Ok(serde_json::from_value(service.dispatch(operation)?)?)
}
#[test]
fn both_directions_replace_properties_survive_reopen_and_delete_together() -> TestResult {
let fixture = Fixture::new()?;
let mut offered = edge("a", "owns", "b");
offered.to = node("other", "a");
let written = fixture.apply("main", "edge", Change::EdgePut(offered.clone()))?;
for (incoming, node) in [(false, offered.from.clone()), (true, offered.to.clone())] {
let page = at_node(
&fixture.service,
incoming,
selection("main", None),
node,
None,
None,
16,
)?;
assert_eq!(page.edges, vec![offered.clone()]);
assert_eq!(page.selection.version, written);
assert_eq!(page.selection.branch, "main");
assert!(page.next_after.is_none());
}
offered.properties = json!({"weight":2,"label":"updated"});
fixture.apply("main", "replace", Change::EdgePut(offered.clone()))?;
let fixture = fixture.reopen()?;
assert_eq!(fixture.export("main")?.edges, vec![offered.clone()]);
let incoming = at_node(
&fixture.service,
true,
selection("main", None),
offered.to.clone(),
None,
None,
16,
)?;
assert_eq!(incoming.edges, vec![offered.clone()]);
fixture.apply(
"main",
"delete",
Change::EdgeDelete(EdgeDelete {
relation: offered.relation,
from: offered.from.clone(),
to: offered.to.clone(),
}),
)?;
for (incoming, node) in [(false, offered.from), (true, offered.to)] {
assert!(
at_node(
&fixture.service,
incoming,
selection("main", None),
node,
None,
None,
16
)?
.edges
.is_empty()
);
}
assert!(fixture.export("main")?.edges.is_empty());
Ok(())
}
#[test]
fn relation_filter_and_continuations_keep_the_captured_version() -> TestResult {
let fixture = Fixture::new()?;
let original = [
edge("a", "a", "b"),
edge("a", "a", "c"),
edge("a", "z", "b"),
];
for (index, offered) in original.iter().enumerate() {
fixture.apply(
"main",
&format!("edge-{index}"),
Change::EdgePut(offered.clone()),
)?;
}
let filtered = at_node(
&fixture.service,
false,
selection("main", None),
node("records", "a"),
Some("a"),
None,
16,
)?;
assert_eq!(filtered.edges, original[..2]);
let incoming = at_node(
&fixture.service,
true,
selection("main", None),
node("records", "b"),
Some("z"),
None,
16,
)?;
assert_eq!(incoming.edges, vec![original[2].clone()]);
let first = at_node(
&fixture.service,
false,
selection("main", None),
node("records", "a"),
None,
None,
1,
)?;
let version = first.selection.version;
let mut after = first.next_after;
let mut seen = first.edges;
fixture.apply("main", "later", Change::EdgePut(edge("a", "0", "b")))?;
let refused = fixture
.service
.dispatch(Operation::EdgesOut(EdgePage {
selection: selection("main", None),
node: node("records", "a"),
relation: None,
after: after.clone(),
limit: NonZeroU32::MIN,
}))
.err()
.ok_or("fresh-head continuation was accepted")?;
assert_eq!(refused.code, "version_required");
for _ in 0..2 {
let cursor = after
.take()
.ok_or("continuation ended before all captured edges")?;
let page = at_node(
&fixture.service,
false,
selection("main", Some(version)),
node("records", "a"),
None,
Some(cursor),
1,
)?;
assert_eq!(page.selection.version, version);
seen.extend(page.edges);
after = page.next_after;
}
assert_eq!(seen, original);
assert!(after.is_none());
assert_eq!(fixture.export("main")?.edges.len(), 4);
Ok(())
}
#[test]
fn missing_ends_invalid_parts_and_nonobject_properties_refuse_without_a_version() -> TestResult {
let fixture = Fixture::new()?;
let before = fixture.export("main")?.selection.version;
for (side, incoming) in [("from", false), ("to", true)] {
let mut offered = edge("a", "owns", "b");
if incoming {
offered.to.document_id = "missing".to_owned();
} else {
offered.from.document_id = "missing".to_owned();
}
let error = fixture
.apply("main", side, Change::EdgePut(offered))
.err()
.ok_or("edge with a missing endpoint was accepted")?;
let failure = error
.downcast_ref::<haematite_cli::serve::protocol::ResponseFailure>()
.ok_or("endpoint refusal lost its type")?;
assert_eq!(failure.code, "document_missing");
assert_eq!(failure.path.as_deref(), Some(format!("/{side}").as_str()));
assert!(failure.message.contains("records/missing"));
}
for (index, properties) in [
Value::Null,
json!([]),
json!(true),
json!(42),
json!("text"),
]
.into_iter()
.enumerate()
{
let mut offered = edge("a", "owns", "b");
offered.properties = properties;
let error = fixture
.apply(
"main",
&format!("properties-{index}"),
Change::EdgePut(offered),
)
.err()
.ok_or("non-object properties were accepted")?;
assert_eq!(
error
.downcast_ref::<haematite_cli::serve::protocol::ResponseFailure>()
.ok_or("refusal lost its type")?
.code,
"edge_properties"
);
}
for index in 0..5 {
let mut offered = edge("a", "owns", "b");
match index {
0 => offered.from.collection = "bad\0collection".to_owned(),
1 => offered.from.document_id.clear(),
2 => offered.relation = "r".repeat(513),
3 => offered.to.collection.clear(),
_ => offered.to.document_id = "bad\0id".to_owned(),
}
let error = fixture
.apply(
"main",
&format!("identifier-{index}"),
Change::EdgePut(offered),
)
.err()
.ok_or("invalid edge identifier was accepted")?;
assert_eq!(
error
.downcast_ref::<haematite_cli::serve::protocol::ResponseFailure>()
.ok_or("refusal lost its type")?
.code,
"invalid_identifier"
);
}
let now = fixture.export("main")?;
assert_eq!(now.selection.version, before);
assert!(now.edges.is_empty());
Ok(())
}
#[test]
fn export_emits_one_copy_and_fork_edges_do_not_leak_to_the_source() -> TestResult {
let fixture = Fixture::new()?;
let original = edge("a", "owns", "b");
let from = fixture.apply("main", "original", Change::EdgePut(original.clone()))?;
fixture.apply(
"main",
"fork",
Change::Fork(Fork {
name: "draft".to_owned(),
from,
}),
)?;
let draft = edge("b", "links", "c");
fixture.apply("draft", "draft-edge", Change::EdgePut(draft.clone()))?;
assert_eq!(fixture.export("main")?.edges, vec![original.clone()]);
let first = graph(&fixture.service, selection("draft", None), None, 1)?;
assert_eq!(first.edges, vec![original]);
let cursor = first.next_after.ok_or("export omitted its continuation")?;
let second = graph(
&fixture.service,
selection("draft", Some(first.selection.version)),
Some(cursor),
1,
)?;
assert_eq!(second.edges, vec![draft]);
assert_eq!(second.selection.version, first.selection.version);
assert!(second.next_after.is_none());
assert_eq!(fixture.export("draft")?.edges.len(), 2);
Ok(())
}
#[test]
fn graph_pages_enforce_byte_budgets_and_validate_every_cursor_component() -> TestResult {
let fixture = Fixture::new()?;
let mut offered = edge("a", "owns", "b");
offered.properties = json!({"blob":"x".repeat(512)});
let version = fixture.apply("main", "large", Change::EdgePut(offered))?;
let error = fixture
.service
.dispatch_bounded(
Operation::GraphExport(GraphPage {
selection: selection("main", None),
after: None,
limit: NonZeroU32::MIN,
relations: None,
collections: None,
members: None,
records: None,
unrelated: None,
}),
NonZeroUsize::MIN,
)
.err()
.ok_or("oversized graph row was accepted")?;
assert_eq!(error.code, "response_limit");
for after in [
"records\0a\0owns\0records\0",
"records\0a\0owns\0records\0b\0extra",
] {
assert!(
fixture
.service
.dispatch(Operation::GraphExport(GraphPage {
selection: selection("main", Some(version)),
after: Some(after.to_owned()),
limit: NonZeroU32::MIN,
relations: None,
collections: None,
members: None,
records: None,
unrelated: None,
}))
.is_err()
);
}
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
}