use std::error::Error;
use liminal::durability::StoredEntry;
use super::{
CodecError, StreamKey, decode_head, decode_key, decode_row, encode_head, encode_row, head_key,
row_key,
};
#[test]
fn keys_encode_the_full_stream_and_api_sequence() -> Result<(), Box<dyn Error>> {
assert_eq!(
head_key("abc")?.as_slice(),
b"liminal:routed-stream:v1:\0\0\0\x03abc\0"
);
let row = row_key("abc", 0x0102_0304_0506_0708)?;
assert_eq!(
row.as_slice(),
b"liminal:routed-stream:v1:\0\0\0\x03abc\x01\x01\x02\x03\x04\x05\x06\x07\x08"
);
assert_eq!(
decode_key(&row)?,
StreamKey::Row {
stream: "abc",
sequence: 0x0102_0304_0506_0708
}
);
assert_eq!(
decode_key(&head_key("abc")?)?,
StreamKey::Head { stream: "abc" }
);
Ok(())
}
#[test]
fn stream_identities_and_row_heads_are_disjoint() -> Result<(), Box<dyn Error>> {
let identities = ["a", "a\0", "ab", "κ"];
let mut keys = std::collections::BTreeSet::new();
for identity in identities {
assert!(keys.insert(head_key(identity)?));
for sequence in [0, 1, u64::MAX] {
assert!(keys.insert(row_key(identity, sequence)?));
}
}
assert_eq!(keys.len(), 16);
Ok(())
}
#[test]
fn key_decoder_refuses_empty_truncated_trailing_and_wrong_kind() -> Result<(), Box<dyn Error>> {
assert!(head_key("").is_err());
assert!(row_key("", 0).is_err());
let row = row_key("abc", 7)?;
for length in 0..row.len() {
assert!(decode_key(&row[..length]).is_err());
}
for mut key in [head_key("abc")?, row] {
key.push(0);
assert!(decode_key(&key).is_err());
}
let mut key = head_key("abc")?;
let last = key.len() - 1;
key[last] = 2;
assert!(decode_key(&key).is_err());
key[last] = 0;
key[0] ^= 1;
assert!(decode_key(&key).is_err());
assert!(decode_key(b"liminal:routed-stream:v1:\0\0\0\0\0").is_err());
assert!(decode_key(b"liminal:routed-stream:v1:\0\0\0\x01\xff\0").is_err());
assert!(decode_key(b"liminal:routed-stream:v1:\xff\xff\xff\xff\0").is_err());
Ok(())
}
#[test]
fn head_has_the_exact_versioned_thirteen_byte_format() -> Result<(), Box<dyn Error>> {
let bytes = encode_head(0x0102_0304_0506_0708);
assert_eq!(&bytes, b"LRSH\x01\x01\x02\x03\x04\x05\x06\x07\x08");
assert_eq!(bytes.len(), 13);
for value in [0, 1, u64::MAX] {
assert_eq!(decode_head(&encode_head(value))?, value);
}
Ok(())
}
#[test]
fn head_decoder_refuses_truncation_trailing_and_unknown_version() {
let bytes = encode_head(7);
for length in 0..bytes.len() {
assert!(decode_head(&bytes[..length]).is_err());
}
let mut trailing = bytes.to_vec();
trailing.push(0);
assert!(decode_head(&trailing).is_err());
let mut wrong = bytes;
wrong[4] = 2;
assert!(decode_head(&wrong).is_err());
wrong[4] = 1;
wrong[0] ^= 1;
assert!(decode_head(&wrong).is_err());
}
#[test]
fn row_preserves_sequence_nanosecond_timestamp_and_binary_payload() -> Result<(), Box<dyn Error>> {
let entry = StoredEntry {
sequence: 0x0102_0304_0506_0708,
timestamp: 0x1112_1314_1516_1718,
payload: vec![9, 8, 7],
};
let bytes = encode_row(&entry)?;
assert_eq!(bytes.as_slice(), b"LRSR\x01\x01\x02\x03\x04\x05\x06\x07\x08\x11\x12\x13\x14\x15\x16\x17\x18\0\0\0\0\0\0\0\x03\x09\x08\x07");
assert_eq!(bytes.len(), 32);
let restored = decode_row(&bytes, entry.sequence)?;
assert_eq!(restored.sequence, entry.sequence);
assert_eq!(restored.timestamp, entry.timestamp);
assert_eq!(restored.payload, entry.payload);
for payload in [Vec::new(), vec![0, 255]] {
let entry = StoredEntry {
sequence: u64::MAX,
timestamp: u64::MAX,
payload,
};
let restored = decode_row(&encode_row(&entry)?, entry.sequence)?;
assert_eq!(restored.payload, entry.payload);
assert_eq!(restored.sequence, entry.sequence);
assert_eq!(restored.timestamp, entry.timestamp);
}
Ok(())
}
#[test]
fn row_decoder_refuses_bad_framing_and_key_value_sequence_mismatch() -> Result<(), Box<dyn Error>> {
let bytes = encode_row(&StoredEntry {
sequence: 7,
timestamp: 19,
payload: vec![1, 2],
})?;
for length in 0..bytes.len() {
assert!(decode_row(&bytes[..length], 7).is_err());
}
assert!(matches!(
decode_row(&bytes, 8),
Err(CodecError::SequenceMismatch {
expected: 8,
actual: 7
})
));
let mut trailing = bytes.clone();
trailing.push(0);
assert!(decode_row(&trailing, 7).is_err());
let mut version = bytes.clone();
version[4] = 2;
assert!(decode_row(&version, 7).is_err());
version[4] = 1;
version[0] ^= 1;
assert!(decode_row(&version, 7).is_err());
let mut oversized = bytes;
oversized[21..29].copy_from_slice(&u64::MAX.to_be_bytes());
assert!(decode_row(&oversized, 7).is_err());
Ok(())
}