use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};
use std::sync::Arc;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{json, value::RawValue};
use super::Service;
use crate::serve::access::TestSource;
use crate::serve::protocol::{
Change, CollectionWrite, DocumentImport, Edge, EdgePage, GraphPage, ImportedDocument, Mutation,
Node, Operation, Selection, Version,
};
use crate::serve::storage::Storage;
type TestResult = Result<(), Box<dyn Error>>;
fn selection(version: Option<Version>) -> Selection {
Selection {
branch: "main".to_owned(),
version,
}
}
fn node(id: &str) -> Node {
Node {
collection: "records".to_owned(),
document_id: id.to_owned(),
}
}
fn edge(from: &str, to: &str) -> Edge {
Edge {
from: node(from),
to: node(to),
relation: "linked".to_owned(),
properties: json!({"weight":7, "label":format!("{from}-{to}")}),
}
}
fn apply(storage: &mut Storage, request_id: &str, change: Change) -> TestResult {
let expected = storage.capture(&selection(None))?.version;
storage.mutate_bounded(
&Mutation {
request_id: request_id.to_owned(),
branch: "main".to_owned(),
expected,
change,
},
NonZeroU32::MIN,
)?;
Ok(())
}
fn with_edges(test: impl FnOnce(Storage) -> TestResult) -> TestResult {
let directory = tempfile::tempdir()?;
let mut storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 2,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
apply(
&mut storage,
"collection",
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
)?;
let documents = ["a", "b", "c", "d", "e"]
.into_iter()
.map(|id| {
Ok(ImportedDocument {
document_id: id.to_owned(),
document: RawValue::from_string("{}".to_owned())?,
})
})
.collect::<Result<Vec<_>, serde_json::Error>>()?;
apply(
&mut storage,
"documents",
Change::PutMany(DocumentImport {
collection: "records".to_owned(),
documents,
}),
)?;
for id in ["b", "c", "d", "e"] {
apply(
&mut storage,
&format!("out-{id}"),
Change::EdgePut(edge("a", id)),
)?;
apply(
&mut storage,
&format!("in-{id}"),
Change::EdgePut(edge(id, "a")),
)?;
}
test(storage)
}
fn page(
incoming: bool,
id: &str,
after: Option<String>,
version: Option<Version>,
rows: u32,
) -> Result<Operation, Box<dyn Error>> {
let page = EdgePage {
selection: selection(version),
node: node(id),
relation: None,
after,
limit: NonZeroU32::new(rows).ok_or("zero page limit")?,
};
Ok(if incoming {
Operation::EdgesIn(page)
} else {
Operation::EdgesOut(page)
})
}
fn export(
after: Option<String>,
version: Option<Version>,
rows: u32,
) -> Result<Operation, Box<dyn Error>> {
Ok(Operation::GraphExport(GraphPage {
selection: selection(version),
after,
limit: NonZeroU32::new(rows).ok_or("zero graph limit")?,
relations: None,
collections: None,
members: None,
records: None,
unrelated: None,
}))
}
fn response_limit() -> Result<NonZeroUsize, Box<dyn Error>> {
Ok(NonZeroUsize::new(4096).ok_or("zero byte limit")?)
}
fn service(
storage: Storage,
denied: &[(&str, &str)],
rows: u32,
) -> Result<Service, Box<dyn Error>> {
Ok(Service::with_access_source(
storage,
NonZeroU32::new(rows).ok_or("zero page limit")?,
None,
Arc::new(TestSource::denying(denied)),
)?)
}
fn denied_is_absent(incoming: bool) -> TestResult {
with_edges(|storage| {
let service = service(storage, &[("records", "b")], 8)?;
let denied =
service.dispatch_bounded(page(incoming, "b", None, None, 8)?, response_limit()?)?;
let absent = service
.dispatch_bounded(page(incoming, "absent", None, None, 8)?, response_limit()?)?;
assert_eq!(denied["edges"], json!([]));
assert_eq!(serde_json::to_vec(&denied)?, serde_json::to_vec(&absent)?);
Ok(())
})
}
#[test]
fn denied_outgoing_node_is_byte_equal_to_absent() -> TestResult {
denied_is_absent(false)
}
#[test]
fn denied_incoming_node_is_byte_equal_to_absent() -> TestResult {
denied_is_absent(true)
}
fn fills_visible_page(incoming: bool) -> TestResult {
with_edges(|storage| {
let service = service(storage, &[("records", "b")], 2)?;
let answer =
service.dispatch_bounded(page(incoming, "a", None, None, 2)?, response_limit()?)?;
let expected = if incoming {
vec![edge("c", "a"), edge("d", "a")]
} else {
vec![edge("a", "c"), edge("a", "d")]
};
assert_eq!(answer["edges"], serde_json::to_value(expected)?);
let next = answer["next_after"]
.as_str()
.ok_or("visible continuation missing")?;
assert_eq!(next, "linked\0records\0d");
Ok(())
})
}
#[test]
fn outgoing_page_replaces_hidden_endpoints_with_visible_edges() -> TestResult {
fills_visible_page(false)
}
#[test]
fn incoming_page_replaces_hidden_endpoints_with_visible_edges() -> TestResult {
fills_visible_page(true)
}
#[test]
fn graph_export_omits_edges_with_either_endpoint_denied() -> TestResult {
with_edges(|storage| {
let service = service(storage, &[("records", "b")], 8)?;
let answer = service.dispatch_bounded(export(None, None, 8)?, response_limit()?)?;
let expected = [
edge("a", "c"),
edge("a", "d"),
edge("a", "e"),
edge("c", "a"),
edge("d", "a"),
edge("e", "a"),
];
assert_eq!(answer["edges"], serde_json::to_value(expected)?);
assert!(answer["next_after"].is_null());
Ok(())
})
}
#[test]
fn visible_cursors_cross_hidden_rows_without_repeating_or_skipping() -> TestResult {
with_edges(|storage| {
let service = service(storage, &[("records", "c")], 1)?;
for kind in 0..3 {
let mut after = None;
let mut version = None;
let mut seen = std::collections::BTreeSet::new();
let mut edges = Vec::new();
loop {
let operation = match kind {
0 => page(false, "a", after.clone(), version, 1)?,
1 => page(true, "a", after.clone(), version, 1)?,
_ => export(after.clone(), version, 1)?,
};
let answer = service.dispatch_bounded(operation, response_limit()?)?;
let rows: Vec<Edge> = serde_json::from_value(answer["edges"].clone())?;
edges.extend(rows);
version = Some(serde_json::from_value(
answer["selection"]["version"].clone(),
)?);
let Some(next) = answer["next_after"].as_str() else {
break;
};
assert!(!next.split('\0').any(|part| part == "c"));
assert!(seen.insert(next.to_owned()), "continuation repeated");
after = Some(next.to_owned());
}
let expected = match kind {
0 => vec![edge("a", "b"), edge("a", "d"), edge("a", "e")],
1 => vec![edge("b", "a"), edge("d", "a"), edge("e", "a")],
_ => vec![
edge("a", "b"),
edge("a", "d"),
edge("a", "e"),
edge("b", "a"),
edge("d", "a"),
edge("e", "a"),
],
};
assert_eq!(edges, expected);
}
Ok(())
})
}
#[test]
fn hidden_properties_do_not_spend_the_visible_byte_budget() -> TestResult {
with_edges(|mut storage| {
let mut hidden = edge("a", "b");
hidden.properties = json!({"hidden":"x".repeat(5000)});
apply(
&mut storage,
"large-hidden-properties",
Change::EdgePut(hidden),
)?;
let service = service(storage, &[("records", "b")], 2)?;
let answer =
service.dispatch_bounded(page(false, "a", None, None, 2)?, response_limit()?)?;
assert_eq!(
answer["edges"],
serde_json::to_value([edge("a", "c"), edge("a", "d")])?
);
Ok(())
})
}
#[test]
fn relationship_authority_is_resolved_once_and_its_failure_propagates() -> TestResult {
with_edges(|storage| {
let source = Arc::new(TestSource::unavailable());
let service = Service::with_access_source(storage, NonZeroU32::MIN, None, source.clone())?;
for (index, operation) in [
page(false, "a", None, None, 1)?,
page(true, "a", None, None, 1)?,
export(None, None, 1)?,
]
.into_iter()
.enumerate()
{
let failure = service
.dispatch(operation)
.err()
.ok_or("unavailable authority served edges")?;
assert_eq!(failure.code, "access_unavailable");
assert_eq!(source.calls(), index + 1);
}
Ok(())
})
}
#[test]
fn denied_edge_run_over_the_scan_cap_is_refused_by_name() -> TestResult {
with_edges(|storage| {
let service = service(
storage,
&[
("records", "b"),
("records", "c"),
("records", "d"),
("records", "e"),
],
1,
)?;
let failure = service
.dispatch_bounded(export(None, None, 1)?, response_limit()?)
.err()
.ok_or("over-cap hidden graph run returned an answer")?;
assert_eq!(failure.code, "access_scan_limit");
Ok(())
})
}
#[test]
fn one_visible_page_records_its_store_call_cost_and_denied_node_reads_nothing() -> TestResult {
use crate::serve::storage::{edges, reads};
with_edges(|storage| {
let reader = storage.reader()?;
let at = storage.capture(&selection(None))?;
reads::reset_page_read_counts();
let raw = edges::out(
&reader,
&at,
&node("a"),
None,
None,
NonZeroUsize::new(2).ok_or("zero row limit")?,
None,
)?;
let before = reads::page_read_counts();
assert_eq!(raw.entries.len(), 2);
assert_eq!(before.points, 0);
assert_eq!(before.ranges, at.shards.len() + 2);
let service = service(storage, &[("records", "b")], 2)?;
reads::reset_page_read_counts();
let answer =
service.dispatch_bounded(page(false, "a", None, None, 2)?, response_limit()?)?;
let after = reads::page_read_counts();
assert_eq!(
answer["edges"],
serde_json::to_value([edge("a", "c"), edge("a", "d")])?
);
assert_eq!(after.points, 1);
assert!(after.ranges <= before.ranges + 1);
eprintln!("relationship page store calls: raw={before:?}; visible={after:?}");
reads::reset_page_read_counts();
let denied =
service.dispatch_bounded(page(false, "b", None, None, 2)?, response_limit()?)?;
assert_eq!(denied["edges"], json!([]));
assert_eq!(reads::page_read_counts(), reads::PageReadCounts::default());
reads::reset_page_read_counts();
let absent =
service.dispatch_bounded(page(false, "absent", None, None, 2)?, response_limit()?)?;
assert_eq!(absent, denied);
assert_eq!(
reads::page_read_counts(),
reads::PageReadCounts {
points: 1,
ranges: 0
}
);
Ok(())
})
}
#[test]
fn a_named_record_denied_is_refused_as_one_not_stored() -> TestResult {
with_edges(|storage| {
let service = service(storage, &[("records", "b")], 8)?;
let named = |ids: &[&str]| -> Result<Operation, Box<dyn Error>> {
Ok(Operation::GraphExport(GraphPage {
selection: selection(None),
after: None,
limit: NonZeroU32::new(8).ok_or("zero graph limit")?,
relations: None,
collections: None,
members: None,
records: Some(ids.iter().copied().map(node).collect()),
unrelated: None,
}))
};
let denied = service
.dispatch_bounded(named(&["b"])?, response_limit()?)
.err()
.ok_or("denied answered")?;
let absent = service
.dispatch_bounded(named(&["a", "absent"])?, response_limit()?)
.err()
.ok_or("absent answered")?;
let mut denied = serde_json::to_value(&denied)?;
let absent = serde_json::to_value(&absent)?;
assert_eq!(denied["code"], "graph_record_unknown");
assert_eq!(denied["path"], "/records/0");
assert_eq!(absent["path"], "/records/1");
denied["path"] = json!("/records/1");
assert_eq!(
serde_json::to_string(&denied)?,
serde_json::to_string(&absent)?
);
Ok(())
})
}
const fn fixture_long_wait_interval() -> std::time::Duration {
std::time::Duration::from_secs(1)
}