extern crate std;
use self::std::panic::{AssertUnwindSafe, catch_unwind};
use super::super::tests::{InRoom, insert, source, view};
use super::*;
#[test]
fn missing_retained_history_cannot_compose_a_partial_delivery() -> Result<(), &'static str> {
let authority = source(1)?;
let maintained = view(&authority)?;
let before = maintained.revision();
let mut buffer = maintained
.delivery_buffer()
.map_err(|_| "buffer unavailable")?;
assert!(matches!(
buffer.first.insert_with_id(
7,
FirstChange {
index: 9,
marker: core::marker::PhantomData,
}
),
WarpTableInsertWithIdOutcome::Inserted { .. }
));
let Err(payload) = catch_unwind(AssertUnwindSafe(|| {
maintained.compose_delivery_changes(&mut buffer)
})) else {
return Err("missing history was accepted");
};
assert_eq!(
payload
.downcast_ref::<alloc::string::String>()
.ok_or("unexpected diagnostic")?,
"WarpTable invariant violated: \"retained history disappeared during delivery\""
);
assert_eq!(maintained.revision(), before);
assert!(buffer.changes.is_empty());
Ok(())
}
#[test]
fn seen_creation_refusals_preserve_the_column_and_allocator_cause() -> Result<(), &'static str> {
use alloc::string::ToString;
let source = Vec::<u8>::new()
.try_reserve_exact(usize::MAX)
.err()
.ok_or("impossible reservation accepted")?;
let message = source.to_string();
let mapped = seen_creation_error(
super::super::super::WarpTableCreationError::ColumnsUnavailable {
source: SeenColumnsCreationError::StorageUnavailable {
column: "marker",
source,
},
},
);
let ViewCreationError::TableUnavailable(
super::super::super::WarpTableCreationError::ColumnsUnavailable { source },
) = mapped
else {
return Err("creation refusal changed category");
};
assert_eq!(source.column, "marker");
assert_eq!(source.source.to_string(), message);
Ok(())
}
#[test]
fn revision_and_epoch_exhaustion_preserve_publication_and_definition() -> Result<(), &'static str> {
let mut authority = source(2)?;
insert(&mut authority, 7, 1, 13)?;
let mut maintained = view(&authority)?;
maintained.revision = u64::MAX;
let before = maintained.revision();
assert_eq!(
maintained.refresh(&authority),
Err(ViewRefreshError::RevisionExhausted)
);
assert_eq!(maintained.revision(), before);
assert_eq!(
maintained
.rows()
.map(|row| (row.id, *row.value))
.collect::<Vec<_>>(),
[(7, 13)]
);
maintained.epoch = u64::MAX;
let before = maintained.revision();
let Err((returned, ViewRefreshError::RevisionExhausted)) =
maintained.replace_definition(&authority, InRoom(2))
else {
return Err("exhausted epoch accepted a definition");
};
assert_eq!(returned.0, 2);
assert_eq!(maintained.definition.0, 1);
assert_eq!(maintained.revision(), before);
assert_eq!(
maintained
.rows()
.map(|row| (row.id, *row.value))
.collect::<Vec<_>>(),
[(7, 13)]
);
Ok(())
}
#[test]
fn corrupted_incarnations_and_primary_keys_do_not_publish_partial_refreshes()
-> Result<(), &'static str> {
let mut authority = source(2)?;
insert(&mut authority, 7, 1, 13)?;
let mut maintained = view(&authority)?;
let before = maintained.revision();
insert(&mut authority, 11, 1, 17)?;
authority.incarnations.truncate(1);
assert!(catch_unwind(AssertUnwindSafe(|| maintained.refresh(&authority))).is_err());
assert_eq!(maintained.revision(), before);
assert_eq!(
maintained
.rows()
.map(|row| (row.id, *row.value))
.collect::<Vec<_>>(),
[(7, 13)]
);
authority.table.ids.truncate(1);
authority.table.len = 1;
assert!(catch_unwind(AssertUnwindSafe(|| maintained.refresh(&authority))).is_err());
assert_eq!(maintained.revision(), before);
assert_eq!(
maintained
.rows()
.map(|row| (row.id, *row.value))
.collect::<Vec<_>>(),
[(7, 13)]
);
Ok(())
}
#[test]
fn underprovisioned_seen_and_delivery_tables_refuse_before_publication() -> Result<(), &'static str>
{
let mut authority = source(4)?;
let mut maintained = view(&authority)?;
let base = maintained.revision();
insert(&mut authority, 7, 1, 13)?;
insert(&mut authority, 11, 1, 17)?;
maintained.seen.maximum_rows = 1;
assert!(catch_unwind(AssertUnwindSafe(|| maintained.refresh(&authority))).is_err());
assert_eq!(maintained.revision(), base);
assert!(maintained.rows().next().is_none());
maintained.seen.maximum_rows = 4;
assert_eq!(
maintained.refresh(&authority),
Ok(ViewRefreshOutcome::Published {
rows: 2,
changes: 2
})
);
assert!(matches!(
authority.remove(7).map_err(|_| "removal unavailable")?,
WarpTableRemoveOutcome::Removed { .. }
));
insert(&mut authority, 7, 1, 19)?;
assert_eq!(
maintained.refresh(&authority),
Ok(ViewRefreshOutcome::Published {
rows: 2,
changes: 1
})
);
let current = maintained.revision();
let mut buffer = maintained
.delivery_buffer()
.map_err(|_| "buffer unavailable")?;
buffer.first.maximum_rows = 1;
assert!(
catch_unwind(AssertUnwindSafe(|| {
let _unexpected_delivery = maintained.delivery(Some(base), &mut buffer);
}))
.is_err()
);
assert_eq!(maintained.revision(), current);
buffer.first.maximum_rows = 2;
let ViewDelivery::Changes {
base: delivered_base,
through,
changes,
} = maintained
.delivery(Some(base), &mut buffer)
.map_err(|_| "delivery unavailable")?
else {
return Err("repaired buffer did not deliver changes");
};
assert_eq!(delivered_base, base);
assert_eq!(through, current);
assert_eq!(changes.len(), 2);
Ok(())
}
#[test]
fn invalid_projection_removal_and_admission_report_exact_invariants() -> Result<(), &'static str> {
let mut authority = source(1)?;
insert(&mut authority, 7, 1, 13)?;
let mut maintained = view(&authority)?;
for (change, diagnostic) in [
(
ViewChange::Remove {
id: 11,
incarnation: 17,
},
"WarpTable invariant violated: \"projected removal absent\"",
),
(
ViewChange::Upsert {
id: 11,
incarnation: 17,
value: 19,
},
"WarpTable invariant violated: \"projection exceeded source capacity\"",
),
] {
maintained.pending.clear();
maintained.pending.push(change);
let Err(reason) = catch_unwind(AssertUnwindSafe(|| maintained.commit(false, 1))) else {
return Err("invalid projection committed");
};
assert_eq!(
reason
.downcast_ref::<alloc::string::String>()
.map(|value| value.as_str()),
Some(diagnostic)
);
assert_eq!(
maintained
.rows()
.map(|row| (row.id, *row.value))
.collect::<Vec<_>>(),
[(7, 13)]
);
}
Ok(())
}
#[test]
fn clear_refuses_an_identity_absent_from_the_primary_index() -> Result<(), &'static str> {
let mut authority = source(1)?;
insert(&mut authority, 7, 1, 13)?;
authority.table.primary_index = super::super::super::dense_key_index::DenseKeyIndex::try_new(1)
.map_err(|_| "index unavailable")?;
let Err(reason) = catch_unwind(AssertUnwindSafe(|| clear(&mut authority.table))) else {
return Err("dangling identity was cleared");
};
assert_eq!(
reason
.downcast_ref::<alloc::string::String>()
.map(|value| value.as_str()),
Some("WarpTable invariant violated: \"table identity unavailable during clear\"")
);
assert_eq!(authority.table.ids(), [7]);
assert_eq!(authority.table.len(), 1);
Ok(())
}