use std::collections::{BTreeMap, BTreeSet};
use std::error::Error;
use std::num::NonZeroUsize;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{json, value::RawValue};
use super::*;
use crate::serve::protocol::{
Cardinality, Change, CollectionWrite, DocumentDelete, DocumentWrite, Edge, Fork, Mutation,
Relation, Selection, Version,
};
type TestResult<T = ()> = Result<T, Box<dyn Error>>;
struct Fixture {
storage: Storage,
directory: tempfile::TempDir,
serial: usize,
}
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,
fixture_long_wait_interval(),
)?;
let mut fixture = Self {
storage,
directory,
serial: 0,
};
fixture.seed()?;
fixture.prepare_branches()?;
Ok(fixture)
}
fn seed(&mut self) -> TestResult {
self.apply(
"main",
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
)?;
for id in [
"first",
"second",
"same",
"unchanged",
"removed",
"end",
"target_only",
"conflict",
"deleted",
"deleted_with_target_edge",
] {
self.put("main", id, 0)?;
}
self.apply(
"main",
Change::RelationPut(Relation {
relation: "links".to_owned(),
from_collection: "records".to_owned(),
to_collection: "records".to_owned(),
cardinality: Cardinality::ManyToMany,
}),
)?;
for (from, to) in [
("removed", "end"),
("removed", "unchanged"),
("end", "removed"),
("unchanged", "removed"),
("removed", "removed"),
("deleted", "end"),
("deleted_with_target_edge", "end"),
] {
self.apply(
"main",
Change::EdgePut(Edge {
relation: "links".to_owned(),
from: node(from),
to: node(to),
properties: json!({}),
}),
)?;
}
self.apply(
"main",
Change::EdgePut(Edge {
relation: "links".to_owned(),
from: node("first"),
to: node("end"),
properties: json!({"weight": 0}),
}),
)?;
Ok(())
}
fn prepare_branches(&mut self) -> TestResult {
for name in ["feature", "foreign"] {
let from = self.live("main")?;
self.apply(
"main",
Change::Fork(Fork {
name: name.to_owned(),
from,
}),
)?;
}
for (id, value) in [("first", 1), ("second", 2), ("same", 3)] {
self.put("feature", id, value)?;
}
self.put("main", "same", 3)?;
self.apply(
"main",
Change::Delete(DocumentDelete {
collection: "records".to_owned(),
document_id: "target_only".to_owned(),
}),
)?;
self.put("foreign", "first", 9)?;
self.prepare_coverage_inputs()?;
self.remove_stored_endpoint()?;
let from = self.live("main")?;
self.apply(
"main",
Change::Fork(Fork {
name: "empty".to_owned(),
from,
}),
)?;
for (id, branch) in [
("review", "feature"),
("other", "feature"),
("foreign", "foreign"),
("new_removed", "feature"),
("new_kept", "feature"),
("coverage", "feature"),
("nothing", "empty"),
] {
let source = MergePin {
branch: branch.to_owned(),
version: self.live(branch)?,
};
let target = MergePin {
branch: "main".to_owned(),
version: self.live("main")?,
};
self.storage.merge_open(id, &source, &target)?;
}
Ok(())
}
fn prepare_coverage_inputs(&mut self) -> TestResult {
for id in ["deleted", "deleted_with_target_edge"] {
self.apply(
"feature",
Change::Delete(DocumentDelete {
collection: "records".to_owned(),
document_id: id.to_owned(),
}),
)?;
}
self.apply(
"main",
Change::EdgePut(Edge {
relation: "links".to_owned(),
from: node("end"),
to: node("deleted_with_target_edge"),
properties: json!({}),
}),
)?;
self.put("feature", "created", 7)?;
self.put("feature", "created_kept", 7)?;
self.put("feature", "conflict", 1)?;
self.put("main", "conflict", 2)?;
for (branch, from, weight) in [
("feature", "created", 0),
("feature", "created_kept", 0),
("feature", "first", 1),
("main", "first", 2),
] {
self.apply(
branch,
Change::EdgePut(Edge {
relation: "links".to_owned(),
from: node(from),
to: node("end"),
properties: json!({"weight": weight}),
}),
)?;
}
let (outgoing, incoming) = edges::keys(&node("ghost"), "links", &node("end"))?;
self.write_stored_source_rows(&[
(outgoing, Some(b"{}".to_vec())),
(incoming, Some(b"{}".to_vec())),
])
}
fn live(&self, branch: &str) -> TestResult<Version> {
Ok(self
.storage
.capture(&Selection {
branch: branch.to_owned(),
version: None,
})?
.version)
}
fn apply(&mut self, branch: &str, change: Change) -> TestResult {
self.serial += 1;
let expected = self.live(branch)?;
self.storage.mutate(&Mutation {
request_id: format!("journal-{}", self.serial),
branch: branch.to_owned(),
expected,
change,
})?;
Ok(())
}
fn put(&mut self, branch: &str, id: &str, value: u64) -> TestResult {
self.apply(
branch,
Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: id.to_owned(),
document: RawValue::from_string(json!({"n": value}).to_string())?,
}),
)
}
fn remove_stored_endpoint(&mut self) -> TestResult {
self.write_stored_source(document_key("records", "removed"), None)
}
fn write_stored_source(&mut self, key: Vec<u8>, value: Option<Vec<u8>>) -> TestResult {
self.write_stored_source_rows(&[(key, value)])
}
fn write_stored_source_rows(&mut self, rows: &[(Vec<u8>, Option<Vec<u8>>)]) -> TestResult {
self.write_stored_rows("feature", rows)
}
fn write_stored_rows(
&mut self,
branch: &str,
rows: &[(Vec<u8>, Option<Vec<u8>>)],
) -> TestResult {
use haematite::branch::{CommitDurability, CommitRequest, current_timestamp, open};
let name = branch_name(branch);
let record = self
.storage
.refs
.get(&name)
.ok_or("missing source")?
.clone();
self.storage.retain_current(branch, &record)?;
let branch = open(&name, &self.storage.refs, &self.storage.registry)?;
for (key, value) in rows {
let shard = self.storage.writer.shard_for(key);
match value {
Some(value) => branch.put(shard, key, value)?,
None => branch.delete(shard, key)?,
}
}
self.storage.writer.commit(
&branch,
&self.storage.registry,
CommitRequest {
durability: CommitDurability::Durable {
refs: &mut self.storage.refs,
},
extra_parents: &[],
timestamp: current_timestamp(),
},
)?;
self.storage.publish_record(&name)?;
Ok(())
}
fn request(
&self,
merge_id: &str,
owner: MergeOwner,
choice: MergeChoice,
) -> TestResult<MergeChoose> {
let plan = self.storage.merge_plan(merge_id)?.ok_or("missing plan")?;
let mut roots = Vec::new();
for pin in [&plan.origin, &plan.target, &plan.source] {
roots.push(self.storage.capture(&Selection {
branch: pin.branch.clone(),
version: Some(pin.version),
})?);
}
let key = owner.key()?;
let mut values = Vec::new();
for at in &roots {
values.push(reads::get_bytes(&self.storage.reader, at, &key)?);
}
let token = choice::Pins::new(&roots[0], &roots[1], &roots[2])?.token(
&key,
values[0].as_deref(),
values[1].as_deref(),
values[2].as_deref(),
)?;
Ok(MergeChoose {
merge_id: merge_id.to_owned(),
owner,
token,
choice,
})
}
}
fn node(id: &str) -> Node {
Node {
collection: "records".to_owned(),
document_id: id.to_owned(),
}
}
fn document(id: &str) -> MergeOwner {
MergeOwner::Document {
collection: "records".to_owned(),
document_id: id.to_owned(),
}
}
fn edge(from: &str) -> MergeOwner {
edge_between(from, "end")
}
fn edge_between(from: &str, to: &str) -> MergeOwner {
MergeOwner::Edge {
relation: "links".to_owned(),
from: node(from),
to: node(to),
}
}
fn new_endpoint_removed(fixture: &mut Fixture) -> TestResult {
let mut request = fixture.request("new_removed", edge("created"), MergeChoice::TakeSource)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&request))?,
"choice_depends_on_endpoint"
);
let document = fixture.request("new_removed", document("created"), MergeChoice::TakeTarget)?;
fixture.storage.merge_choose(&document)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&request))?,
"choice_requires_remove_edge"
);
request.choice = MergeChoice::RemoveEdge;
fixture.storage.merge_choose(&request)?;
Ok(())
}
fn new_endpoint_kept(fixture: &mut Fixture) -> TestResult {
let kept = fixture.request("new_kept", document("created"), MergeChoice::TakeSource)?;
fixture.storage.merge_choose(&kept)?;
let mut request = fixture.request("new_kept", edge("created"), MergeChoice::RemoveEdge)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&request))?,
"choice_not_needed"
);
request.choice = MergeChoice::TakeSource;
fixture.storage.merge_choose(&request)?;
let remove = fixture.request("new_kept", document("created"), MergeChoice::TakeTarget)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&remove))?,
"choice_conflict"
);
let mut ghost = fixture.request("new_kept", edge("ghost"), MergeChoice::TakeSource)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&ghost))?,
"choice_requires_remove_edge"
);
ghost.choice = MergeChoice::RemoveEdge;
fixture.storage.merge_choose(&ghost)?;
Ok(())
}
struct CoverageListing {
entries: Vec<MissingChoice>,
resumed_incident: BTreeSet<(bool, String)>,
}
impl CoverageListing {
fn by_owner(self) -> TestResult<BTreeMap<Vec<u8>, MissingChoice>> {
self.entries
.into_iter()
.map(|missing| Ok((missing.owner.key()?, missing)))
.collect()
}
}
fn coverage_entries(fixture: &Fixture, id: &str) -> TestResult<CoverageListing> {
let mut found = CoverageListing {
entries: Vec::new(),
resumed_incident: BTreeSet::new(),
};
let mut after: Option<String> = None;
for _ in 0..128 {
if let Some(after) = &after {
let position: serde_json::Value = cursor::decode(after)?;
let incident = &position["incident"];
if incident["node"]["document_id"] == "removed"
&& let Some(suffix) = incident["after"].as_str()
{
let incoming = incident["incoming"]
.as_bool()
.ok_or("missing incident direction")?;
found.resumed_incident.insert((incoming, suffix.to_owned()));
}
}
let before = fixture.storage.merge_service_selection()?.version;
let page = fixture
.storage
.merge_coverage(id, after.as_deref(), NonZeroUsize::MIN)?;
assert_eq!(fixture.storage.merge_service_selection()?.version, before);
found.entries.extend(page.missing);
if page.end {
assert!(page.next_after.is_none());
return Ok(found);
}
let next = page.next_after.ok_or("unterminated page has no cursor")?;
assert_ne!(after.as_ref(), Some(&next));
after = Some(next);
}
Err("coverage listing did not end".into())
}
fn coverage_missing(fixture: &Fixture) -> TestResult<BTreeMap<Vec<u8>, MissingChoice>> {
coverage_entries(fixture, "coverage")?.by_owner()
}
fn choose_missing(fixture: &mut Fixture, found: BTreeMap<Vec<u8>, MissingChoice>) -> TestResult {
for missing in found.into_values() {
let choice = match &missing.owner {
MergeOwner::Document { .. } => MergeChoice::TakeSource,
MergeOwner::Edge { from, .. } if from.document_id == "first" => MergeChoice::TakeSource,
MergeOwner::Edge { .. } => MergeChoice::RemoveEdge,
};
let request = fixture.request("coverage", missing.owner, choice)?;
assert_eq!(request.token, missing.token);
fixture.storage.merge_choose(&request)?;
}
Ok(())
}
fn forged_coverage_is_only_a_listing(fixture: &Fixture) -> TestResult {
let page = fixture
.storage
.merge_coverage("coverage", None, NonZeroUsize::MIN)?;
let Some(after) = page.next_after else {
return Err("fixture ended before a continuation".into());
};
let mut forged: serde_json::Value = cursor::decode(&after)?;
forged["comparison"] = serde_json::Value::Null;
forged["diff_done"] = json!(true);
forged["incident"] = serde_json::Value::Null;
forged["pending"] = json!([]);
let forged = cursor::encode(&forged)?;
let before = fixture.storage.merge_service_selection()?.version;
let page = fixture
.storage
.merge_coverage("coverage", Some(&forged), NonZeroUsize::MIN)?;
assert!(page.end);
assert!(page.missing.is_empty());
assert!(serde_json::to_value(&page)?.get("complete").is_none());
assert_eq!(fixture.storage.merge_service_selection()?.version, before);
Ok(())
}
fn coverage_listing(fixture: &mut Fixture) -> TestResult {
let before = fixture.storage.merge_service_selection()?.version;
let nothing = fixture
.storage
.merge_coverage("nothing", None, NonZeroUsize::MIN)?;
assert!(nothing.end);
assert!(nothing.missing.is_empty());
assert_eq!(fixture.storage.merge_service_selection()?.version, before);
forged_coverage_is_only_a_listing(fixture)?;
let found = coverage_missing(fixture)?;
assert!(!found.contains_key(&document("deleted").key()?));
assert!(!found.contains_key(&edge("deleted").key()?));
let expected: BTreeSet<_> = [
document("conflict"),
document("removed"),
document("deleted_with_target_edge"),
edge("first"),
edge("ghost"),
]
.iter()
.map(MergeOwner::key)
.collect::<Result<_, _>>()?;
assert_eq!(found.keys().cloned().collect::<BTreeSet<_>>(), expected);
choose_missing(fixture, found)?;
let remove = fixture.request("coverage", document("created"), MergeChoice::TakeTarget)?;
fixture.storage.merge_choose(&remove)?;
let listing = coverage_entries(fixture, "coverage")?;
incident_cursor(&listing)?;
let found = listing.by_owner()?;
let expected: BTreeSet<_> = [
edge("created"),
edge("removed"),
edge_between("removed", "unchanged"),
edge_between("end", "removed"),
edge_between("unchanged", "removed"),
edge_between("removed", "removed"),
edge_between("end", "deleted_with_target_edge"),
]
.iter()
.map(MergeOwner::key)
.collect::<Result<_, _>>()?;
assert_eq!(found.keys().cloned().collect::<BTreeSet<_>>(), expected);
choose_missing(fixture, found)?;
assert!(coverage_missing(fixture)?.is_empty());
unchosen_created_has_no_rows(fixture)?;
Ok(())
}
fn unchosen_created_has_no_rows(fixture: &Fixture) -> TestResult {
let service = fixture.storage.merge_service_selection()?;
for owner in [document("created_kept"), edge("created_kept")] {
let request = fixture.request("coverage", owner, MergeChoice::TakeSource)?;
assert!(
fixture
.storage
.read_choice_row(&service, "coverage", &request.token)?
.is_none()
);
}
Ok(())
}
fn incident_cursor(listing: &CoverageListing) -> TestResult {
let positions: BTreeSet<_> = [
(false, "end"),
(false, "removed"),
(false, "unchanged"),
(true, "end"),
(true, "unchanged"),
]
.into_iter()
.map(|(incoming, other)| (incoming, format!("links\0records\0{other}")))
.collect();
assert_eq!(listing.resumed_incident, positions);
let mut seen = Vec::new();
for missing in &listing.entries {
if let MergeOwner::Edge { from, to, .. } = &missing.owner
&& (from.document_id == "removed" || to.document_id == "removed")
{
seen.push(missing.owner.key()?);
}
}
let expected: BTreeSet<_> = [
edge("removed"),
edge_between("removed", "unchanged"),
edge_between("end", "removed"),
edge_between("unchanged", "removed"),
edge_between("removed", "removed"),
]
.iter()
.map(MergeOwner::key)
.collect::<Result<_, _>>()?;
assert_eq!(seen.len(), expected.len());
assert_eq!(seen.into_iter().collect::<BTreeSet<_>>(), expected);
Ok(())
}
fn unsupported_owners(fixture: &mut Fixture) -> TestResult {
for (parent, name) in [
("main", "unsupported_target"),
("unsupported_target", "unsupported_source"),
] {
let from = fixture.live(parent)?;
fixture.apply(
parent,
Change::Fork(Fork {
name: name.to_owned(),
from,
}),
)?;
}
for (branch, name) in [
("unsupported_source", "source_only"),
("unsupported_target", "target_only"),
] {
let version = fixture.live(branch)?;
let collection = crate::serve::storage::Collection {
name: name.to_owned(),
schema: None,
changed_in: branch.to_owned(),
changed_at: version,
};
let relation = Relation {
relation: name.to_owned(),
from_collection: "records".to_owned(),
to_collection: "records".to_owned(),
cardinality: Cardinality::ManyToMany,
};
fixture.write_stored_rows(
branch,
&[
(
format!("c\0{name}").into_bytes(),
Some(serde_json::to_vec(&collection)?),
),
(
format!("n\0{name}").into_bytes(),
Some(serde_json::to_vec(&relation)?),
),
(
format!("v\0records\0{name}\0end").into_bytes(),
Some(1.0_f32.to_le_bytes().to_vec()),
),
(
format!("w\0records\0{name}").into_bytes(),
Some(b"1".to_vec()),
),
],
)?;
}
open_unsupported(fixture, "unsupported_one_side")?;
assert!(
coverage_entries(fixture, "unsupported_one_side")?
.entries
.is_empty()
);
for (branch, schema) in [("unsupported_source", true), ("unsupported_target", false)] {
fixture.apply(
branch,
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: Some(json!(schema)),
}),
)?;
}
open_unsupported(fixture, "unsupported_conflict")?;
let failed = unsupported_refusal(fixture)?;
assert_eq!(
(failed.code.as_str(), failed.message.as_str()),
(
"choice_owner_unsupported",
"family c owner records has conflicting values and no choice kind"
)
);
Ok(())
}
fn open_unsupported(fixture: &mut Fixture, id: &str) -> TestResult {
let source = MergePin {
branch: "unsupported_source".to_owned(),
version: fixture.live("unsupported_source")?,
};
let target = MergePin {
branch: "unsupported_target".to_owned(),
version: fixture.live("unsupported_target")?,
};
fixture.storage.merge_open(id, &source, &target)?;
Ok(())
}
fn unsupported_refusal(fixture: &Fixture) -> TestResult<Failure> {
let mut after: Option<String> = None;
for _ in 0..128 {
match fixture.storage.merge_coverage(
"unsupported_conflict",
after.as_deref(),
NonZeroUsize::MIN,
) {
Err(error) => return Ok(error),
Ok(page) if page.end => return Err("unsupported conflict reached the end".into()),
Ok(page) => after = Some(page.next_after.ok_or("missing coverage continuation")?),
}
}
Err("unsupported conflict was not reached".into())
}
fn refusal<T>(result: Result<T, Failure>) -> TestResult<String> {
Ok(result.err().ok_or("invalid choice accepted")?.code)
}
fn admission(fixture: &mut Fixture) -> TestResult {
let mut first = fixture.request("review", document("first"), MergeChoice::TakeSource)?;
let before = fixture.storage.merge_service_selection()?.version;
let admitted = fixture.storage.merge_choose(&first)?;
let after = fixture.storage.merge_service_selection()?.version;
assert_eq!(after.sequence, before.sequence + 1);
assert_eq!(fixture.storage.merge_choose(&first)?, admitted);
assert_eq!(fixture.storage.merge_service_selection()?.version, after);
first.choice = MergeChoice::TakeTarget;
assert_eq!(
refusal(fixture.storage.merge_choose(&first))?,
"choice_conflict"
);
first.choice = MergeChoice::TakeSource;
first.owner = document("second");
assert_eq!(
refusal(fixture.storage.merge_choose(&first))?,
"choice_token_mismatch"
);
first.token = fixture
.request("foreign", document("first"), MergeChoice::TakeSource)?
.token;
first.owner = document("first");
assert_eq!(
refusal(fixture.storage.merge_choose(&first))?,
"choice_token_mismatch"
);
let mut malformed = first.clone();
malformed.token = "merge-choice-v1:INVALID".to_owned();
assert_eq!(
refusal(fixture.storage.merge_choose(&malformed))?,
"choice_token_invalid"
);
malformed.token = first.token.clone();
malformed.owner = document("");
assert_eq!(
refusal(fixture.storage.merge_choose(&malformed))?,
"invalid_identifier"
);
for (id, expected) in [
("same", "choice_same_on_both"),
("unchanged", "choice_not_needed"),
] {
let request = fixture.request("review", document(id), MergeChoice::TakeSource)?;
assert_eq!(refusal(fixture.storage.merge_choose(&request))?, expected);
}
let mut invalid = fixture.request("review", document("second"), MergeChoice::RemoveEdge)?;
let revision = fixture
.storage
.merge_choices("review", None, NonZeroUsize::MIN)?
.revision;
assert_eq!(
refusal(fixture.storage.merge_choose(&invalid))?,
"choice_kind_mismatch"
);
assert_eq!(
fixture
.storage
.merge_choices("review", None, NonZeroUsize::MIN)?
.revision,
revision
);
invalid.choice = MergeChoice::TakeSource;
fixture.storage.merge_choose(&invalid)?;
Ok(())
}
fn unchanged_inputs(fixture: &Fixture) -> TestResult {
let plan = fixture
.storage
.merge_plan("review")?
.ok_or("missing plan")?;
assert_eq!(fixture.live("feature")?, plan.source.version);
assert_eq!(fixture.live("main")?, plan.target.version);
let target = fixture.storage.capture(&Selection {
branch: plan.target.branch,
version: Some(plan.target.version),
})?;
assert_eq!(
reads::get_bytes(
&fixture.storage.reader,
&target,
&document_key("records", "first")
)?,
Some(b"{\"n\":0}".to_vec())
);
let key = edges::keys(&node("removed"), "links", &node("end"))?.0;
assert!(reads::get_bytes(&fixture.storage.reader, &target, &key)?.is_some());
Ok(())
}
fn pages(fixture: &mut Fixture) -> TestResult<String> {
let first = fixture
.storage
.merge_choices("review", None, NonZeroUsize::MIN)?;
assert_eq!(first.entries.len(), 1);
let cursor = first.next_after.ok_or("missing second page")?;
let other = fixture.request("other", document("first"), MergeChoice::TakeTarget)?;
fixture.storage.merge_choose(&other)?;
let second = fixture
.storage
.merge_choices("review", Some(&cursor), NonZeroUsize::MIN)?;
assert_eq!(second.entries.len(), 1);
assert!(second.next_after.is_none());
assert_ne!(first.entries[0].token, second.entries[0].token);
assert_eq!(first.revision, second.revision);
assert_eq!(
refusal(
fixture
.storage
.merge_choices("other", Some(&cursor), NonZeroUsize::MIN)
)?,
"invalid_continuation"
);
let mut forged: ChoiceCursor = cursor::decode(&cursor)?;
forged.token = fixture
.request("foreign", document("first"), MergeChoice::TakeSource)?
.token;
let forged = cursor::encode(&forged)?;
assert_eq!(
refusal(
fixture
.storage
.merge_choices("review", Some(&forged), NonZeroUsize::MIN)
)?,
"invalid_continuation"
);
Ok(cursor)
}
fn endpoint(fixture: &mut Fixture, old_cursor: &str) -> TestResult {
let edge = MergeOwner::Edge {
relation: "links".to_owned(),
from: node("removed"),
to: node("end"),
};
let request = fixture.request("review", edge, MergeChoice::RemoveEdge)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&request))?,
"choice_depends_on_endpoint"
);
let remove = fixture.request("review", document("removed"), MergeChoice::TakeSource)?;
fixture.storage.merge_choose(&remove)?;
let mut keep_edge = request.clone();
keep_edge.choice = MergeChoice::TakeTarget;
assert_eq!(
refusal(fixture.storage.merge_choose(&keep_edge))?,
"choice_requires_remove_edge"
);
fixture.storage.merge_choose(&request)?;
assert_eq!(
refusal(
fixture
.storage
.merge_choices("review", Some(old_cursor), NonZeroUsize::MIN)
)?,
"choice_journal_moved"
);
let restore = fixture.request("review", document("removed"), MergeChoice::TakeTarget)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&restore))?,
"choice_conflict"
);
let keep = fixture.request("other", document("removed"), MergeChoice::TakeTarget)?;
fixture.storage.merge_choose(&keep)?;
let repaired = fixture.request("other", request.owner, MergeChoice::RemoveEdge)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&repaired))?,
"choice_not_needed"
);
let target_only =
fixture.request("review", document("target_only"), MergeChoice::TakeTarget)?;
fixture.storage.merge_choose(&target_only)?;
Ok(())
}
#[test]
fn choice_admission_replay_dependencies_and_cursor_share_one_fixture() -> TestResult {
let mut fixture = Fixture::new()?;
let empty = fixture
.storage
.merge_choices("review", None, NonZeroUsize::MIN)?;
assert!(empty.entries.is_empty());
assert_eq!(empty.revision, 0);
new_endpoint_removed(&mut fixture)?;
new_endpoint_kept(&mut fixture)?;
coverage_listing(&mut fixture)?;
admission(&mut fixture)?;
let cursor = pages(&mut fixture)?;
endpoint(&mut fixture, &cursor)?;
unchanged_inputs(&fixture)?;
let request = fixture.request("review", document("first"), MergeChoice::TakeSource)?;
let path = fixture.directory.path().join("database");
drop(fixture.storage);
fixture.storage = Storage::open(
&path,
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
let reopened =
fixture
.storage
.merge_choices("review", None, NonZeroUsize::MIN.saturating_add(7))?;
assert_eq!(reopened.entries.len(), 5);
let before = fixture.storage.merge_service_selection()?.version;
fixture.storage.merge_choose(&request)?;
assert_eq!(fixture.storage.merge_service_selection()?.version, before);
let foreign = fixture.request("foreign", document("first"), MergeChoice::TakeSource)?;
fixture.put("feature", "first", 17)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&request))?,
"merge_pins_moved"
);
fixture.put("main", "first", 23)?;
assert_eq!(
refusal(fixture.storage.merge_choose(&foreign))?,
"merge_pins_moved"
);
fixture.storage.merge_abandon("review")?;
assert_eq!(
refusal(fixture.storage.merge_choose(&request))?,
"merge_abandoned"
);
assert_eq!(
refusal(
fixture
.storage
.merge_choices("review", None, NonZeroUsize::MIN)
)?,
"merge_abandoned"
);
let mut missing = request;
missing.merge_id = "missing".to_owned();
assert_eq!(
refusal(fixture.storage.merge_choose(&missing))?,
"merge_missing"
);
unsupported_owners(&mut fixture)?;
Ok(())
}
const fn fixture_long_wait_interval() -> std::time::Duration {
const LONG_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
LONG_WAIT_INTERVAL
}