use std::error::Error;
use haematite::branch::ShardRef;
use haematite::tree::Hash;
use serde_json::{Value, json};
use super::*;
use crate::serve::protocol::Version;
type TestResult = Result<(), Box<dyn Error>>;
fn captured(branch: &str, sequence: u64) -> Captured {
Captured {
branch: branch.to_owned(),
version: Version {
generation: 1,
sequence,
},
shards: vec![ShardRef {
shard_id: 0,
fork_anchor: Hash::of(b"anchor"),
head: Hash::of(b"head"),
}],
}
}
fn page(source: Vec<ObservedChange>, target: Vec<ObservedChange>) -> compare::Page {
let source_scanned = source.len();
let target_scanned = target.len();
compare::Page {
selection: captured("feature", 3),
from: ForkProvenance {
branch: "main".to_owned(),
version: captured("main", 1).version,
},
origin: captured("main", 1),
parent: captured("main", 2),
source,
target,
current_shard: Some(0),
next_after: None,
nodes_loaded: 0,
source_scanned,
target_scanned,
held: Vec::new(),
}
}
fn removed(id: &str) -> ObservedChange {
ObservedChange::Removed {
key: document_key("records", id),
value: b"{}".to_vec(),
}
}
#[test]
fn deletion_at_the_last_key_of_an_exhausted_shard_still_has_work_and_a_cursor() -> TestResult {
let compared = page(vec![removed("z")], Vec::new());
assert!(compared.next_after.is_none());
let checkpoint = comparison_cursor(&compared)?;
assert!(!checkpoint.comparison_done);
assert_eq!(checkpoint.pending.len(), 1);
assert_eq!(checkpoint.pending[0].node.document_id, "z");
let point: Value = cursor::decode(&checkpoint.comparison)?;
assert_eq!(point["review"]["shard"], 0);
assert_eq!(point["review"]["key"], json!(document_key("records", "z")));
assert_eq!(
point["review"]["version"],
serde_json::to_value(compared.selection.version)?
);
assert_eq!(
point["review"]["from"],
serde_json::to_value(&compared.from)?
);
assert_eq!(
point["parent"],
serde_json::to_value(compared.parent.version)?
);
Ok(())
}
#[test]
fn a_batch_queues_only_the_earliest_document_key() -> TestResult {
let compared = page(vec![removed("b"), removed("c")], vec![removed("c")]);
let checkpoint = comparison_cursor(&compared)?;
assert_eq!(checkpoint.pending.len(), 1);
assert_eq!(checkpoint.pending[0].node.document_id, "b");
assert!(!checkpoint.comparison_done);
let point: Value = cursor::decode(&checkpoint.comparison)?;
assert_eq!(point["review"]["key"], json!(document_key("records", "b")));
Ok(())
}
#[test]
fn both_sides_of_one_deleted_key_fit_the_existing_pending_bound() -> TestResult {
let compared = page(vec![removed("b")], vec![removed("b"), removed("c")]);
let mut checkpoint = comparison_cursor(&compared)?;
assert_eq!(checkpoint.pending.len(), 2);
let first = checkpoint.pending.pop().ok_or("missing source removal")?;
let second = checkpoint.pending.pop().ok_or("missing target removal")?;
assert_eq!(first.node.document_id, "b");
assert_eq!(second.node.document_id, "b");
assert!(matches!(first.removed_by, Side::Source));
assert!(matches!(second.removed_by, Side::Target));
Ok(())
}
#[test]
fn a_batch_without_deletions_uses_its_shared_settled_end() -> TestResult {
let mut compared = page(
vec![ObservedChange::Added {
key: document_key("records", "b"),
value: b"{}".to_vec(),
}],
Vec::new(),
);
let end = compare::after_key(&compared, &document_key("records", "b"))?;
compared.next_after = Some(end.clone());
let checkpoint = comparison_cursor(&compared)?;
assert!(checkpoint.pending.is_empty());
assert_eq!(checkpoint.comparison, end);
assert!(!checkpoint.comparison_done);
Ok(())
}
#[test]
fn selecting_a_key_outside_the_settled_page_is_refused() -> TestResult {
let compared = page(vec![removed("b")], Vec::new());
let refusal = compare::after_key(&compared, &document_key("records", "c"))
.err()
.ok_or("unsettled key was accepted")?;
assert_eq!(refusal.code, "invalid_continuation");
Ok(())
}
#[test]
fn a_target_deletion_before_the_source_deletion_is_selected_first() -> TestResult {
let compared = page(vec![removed("c")], vec![removed("b")]);
let checkpoint = comparison_cursor(&compared)?;
assert_eq!(checkpoint.pending.len(), 1);
assert_eq!(checkpoint.pending[0].node.document_id, "b");
assert!(matches!(checkpoint.pending[0].removed_by, Side::Target));
let point: Value = cursor::decode(&checkpoint.comparison)?;
assert_eq!(point["review"]["key"], json!(document_key("records", "b")));
Ok(())
}
#[test]
fn an_exhausted_shard_with_a_deletion_cannot_jump_to_the_next_shard() -> TestResult {
let mut compared = page(vec![removed("z")], Vec::new());
for at in [
&mut compared.selection,
&mut compared.origin,
&mut compared.parent,
] {
at.shards.push(ShardRef {
shard_id: 1,
fork_anchor: Hash::of(b"anchor"),
head: Hash::of(b"head"),
});
}
compared.next_after = Some(cursor::encode(&json!({
"review": {"branch":&compared.selection.branch, "version":compared.selection.version,
"from":&compared.from, "shard":1, "key":null},
"parent":compared.parent.version,
}))?);
let checkpoint = comparison_cursor(&compared)?;
assert_eq!(checkpoint.pending.len(), 1);
assert!(!checkpoint.comparison_done);
let point: Value = cursor::decode(&checkpoint.comparison)?;
assert_eq!(point["review"]["shard"], 0);
assert_eq!(point["review"]["key"], json!(document_key("records", "z")));
Ok(())
}
#[test]
fn a_page_without_a_settled_shard_cannot_mint_a_selected_cursor() -> TestResult {
let mut compared = page(vec![removed("b")], Vec::new());
compared.current_shard = None;
let refusal = compare::after_key(&compared, &document_key("records", "b"))
.err()
.ok_or("missing shard was accepted")?;
assert_eq!(refusal.code, "invalid_continuation");
Ok(())
}