use super::StreamingDecoder;
use crate::io::Read;
fn frame_declaring(fcs: u64, single_segment: bool, payload: &[u8]) -> alloc::vec::Vec<u8> {
let mut f = alloc::vec![0x28, 0xB5, 0x2F, 0xFD];
if single_segment {
f.push(0xE0);
} else {
f.extend_from_slice(&[0xC0, 0x00]);
}
f.extend_from_slice(&fcs.to_le_bytes());
let header = ((payload.len() as u32) << 3) | 1;
f.extend_from_slice(&header.to_le_bytes()[..3]);
f.extend_from_slice(payload);
f
}
fn drain_with_read(
mut decoder: StreamingDecoder<&[u8], crate::decoding::FrameDecoder>,
) -> Result<alloc::vec::Vec<u8>, crate::io::Error> {
let mut out = alloc::vec::Vec::new();
let mut buf = [0u8; 4096];
loop {
let n = decoder.read(&mut buf)?;
if n == 0 {
return Ok(out);
}
out.extend_from_slice(&buf[..n]);
}
}
fn frame_of(payload: &[u8]) -> alloc::vec::Vec<u8> {
crate::encoding::compress_to_vec(payload, crate::encoding::CompressionLevel::Fastest)
}
fn skippable_frame(payload: &[u8]) -> alloc::vec::Vec<u8> {
let mut f = alloc::vec![0x50, 0x2A, 0x4D, 0x18];
f.extend_from_slice(&(payload.len() as u32).to_le_bytes());
f.extend_from_slice(payload);
f
}
#[test]
fn read_continues_into_following_frames() {
let mut stream = frame_of(b"first frame, ");
stream.extend(frame_of(b"second frame, "));
stream.extend(frame_of(b"third"));
let decoder = StreamingDecoder::new(stream.as_slice());
assert_eq!(
drain_with_read(decoder).unwrap(),
b"first frame, second frame, third"
);
}
#[test]
fn read_skips_skippable_frames_anywhere_in_the_stream() {
let mut stream = skippable_frame(b"leading metadata");
stream.extend(frame_of(b"one "));
stream.extend(skippable_frame(b""));
stream.extend(frame_of(b"two"));
stream.extend(skippable_frame(b"trailer"));
let decoder = StreamingDecoder::new(stream.as_slice());
assert_eq!(drain_with_read(decoder).unwrap(), b"one two");
}
#[test]
fn a_stream_of_only_skippable_frames_decodes_to_nothing() {
let mut stream = skippable_frame(b"first");
stream.extend(skippable_frame(b"second"));
let decoder = StreamingDecoder::new(stream.as_slice());
assert!(drain_with_read(decoder).unwrap().is_empty());
let mut decoder = StreamingDecoder::new(stream.as_slice());
let mut out = alloc::vec::Vec::new();
decoder.read_to_end(&mut out).unwrap();
assert!(out.is_empty());
}
#[test]
fn read_rejects_bytes_after_the_last_frame_that_are_not_a_frame() {
let mut stream = frame_of(b"payload");
stream.extend_from_slice(b"not a frame");
let decoder = StreamingDecoder::new(stream.as_slice());
assert!(drain_with_read(decoder).is_err());
}
#[test]
fn read_applies_the_constructor_dictionary_to_following_frames() {
use crate::decoding::{Dictionary, DictionaryHandle};
use crate::encoding::{CompressionLevel, FrameCompressor};
let content = b"shared words the frames reuse again and again".to_vec();
let handle = DictionaryHandle::from_dictionary(
Dictionary::from_raw_content(7, content.clone()).unwrap(),
);
let mut compressor: FrameCompressor = FrameCompressor::new(CompressionLevel::Default);
compressor
.set_dictionary(Dictionary::from_raw_content(7, content).unwrap())
.unwrap();
let mut stream = compressor.compress_independent_frame(b"the frames reuse words");
stream.extend(compressor.compress_independent_frame(b" and again the shared words"));
let decoder = StreamingDecoder::new_with_dictionary_handle(stream.as_slice(), &handle).unwrap();
assert_eq!(
drain_with_read(decoder).unwrap(),
b"the frames reuse words and again the shared words"
);
}
struct OpenPipe<'a> {
data: &'a [u8],
}
impl Read for OpenPipe<'_> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize, crate::io::Error> {
if self.data.is_empty() {
return Err(crate::io::Error::from(crate::io::ErrorKind::WouldBlock));
}
self.data.read(buf)
}
}
struct Trickle<'a> {
data: &'a [u8],
block_next: bool,
}
impl Read for Trickle<'_> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize, crate::io::Error> {
self.block_next = !self.block_next;
if !self.block_next {
return Err(crate::io::Error::from(crate::io::ErrorKind::WouldBlock));
}
let n = buf.len().min(1).min(self.data.len());
buf[..n].copy_from_slice(&self.data[..n]);
self.data = &self.data[n..];
Ok(n)
}
}
#[test]
fn read_resumes_wherever_a_would_block_interrupts_it() {
let payload: alloc::vec::Vec<u8> = (0..300_000u32).map(|i| (i % 251) as u8).collect();
let mut compressor =
crate::encoding::FrameCompressor::new(crate::encoding::CompressionLevel::Fastest);
compressor.set_content_checksum(true);
compressor.set_source(payload.as_slice());
let mut stream = alloc::vec::Vec::new();
compressor.set_drain(&mut stream);
compressor.compress();
stream.extend(skippable_frame(b"metadata between frames"));
stream.extend(frame_of(b"second"));
let mut decoder = StreamingDecoder::new(Trickle {
data: &stream,
block_next: false,
});
#[cfg(feature = "hash")]
decoder
.decoder_mut()
.set_content_checksum(crate::decoding::ContentChecksum::Verify);
let mut out = alloc::vec::Vec::new();
let mut buf = [0u8; 4096];
loop {
match decoder.read(&mut buf) {
Ok(0) => break,
Ok(n) => out.extend_from_slice(&buf[..n]),
Err(e) if e.kind() == crate::io::ErrorKind::WouldBlock => {}
Err(e) => panic!("the stream must survive WouldBlock: {e:?}"),
}
}
let mut expected = payload;
expected.extend_from_slice(b"second");
assert!(out == expected, "decoded {} bytes", out.len());
}
#[test]
fn read_continues_into_following_magicless_frames() {
let first = frame_declaring(5, false, b"first");
let second = frame_declaring(6, false, b"second");
let mut stream = first[4..].to_vec();
stream.extend_from_slice(&second[4..]);
let mut frame_decoder = crate::decoding::FrameDecoder::new();
frame_decoder.set_magicless(true);
let source = Trickle {
data: &stream,
block_next: false,
};
let mut decoder = StreamingDecoder::new_with_decoder(source, frame_decoder);
let mut out = alloc::vec::Vec::new();
let mut buf = [0u8; 64];
loop {
match decoder.read(&mut buf) {
Ok(0) => break,
Ok(n) => out.extend_from_slice(&buf[..n]),
Err(e) if e.kind() == crate::io::ErrorKind::WouldBlock => {}
Err(e) => panic!("magicless frames must decode in turn: {e:?}"),
}
}
assert_eq!(out, b"firstsecond");
}
#[test]
fn into_inner_after_a_frame_returns_the_rest_of_the_source() {
let payload = b"one frame of content";
let mut stream = frame_of(payload);
stream.extend_from_slice(b"bytes after the frame");
let mut decoder = StreamingDecoder::new(stream.as_slice());
let mut out = [0u8; 20];
decoder.read_exact(&mut out).unwrap();
assert_eq!(&out, payload);
assert_eq!(decoder.into_inner(), b"bytes after the frame");
}
#[test]
fn into_inner_after_a_checksummed_frame_returns_the_rest_of_the_source() {
let payload = b"one frame of content";
let mut compressor =
crate::encoding::FrameCompressor::new(crate::encoding::CompressionLevel::Fastest);
compressor.set_content_checksum(true);
compressor.set_source(&payload[..]);
let mut stream = alloc::vec::Vec::new();
compressor.set_drain(&mut stream);
compressor.compress();
stream.extend_from_slice(b"bytes after the frame");
let mut decoder = StreamingDecoder::new(stream.as_slice());
let mut out = [0u8; 20];
decoder.read_exact(&mut out).unwrap();
assert_eq!(&out, payload);
assert_eq!(decoder.into_inner(), b"bytes after the frame");
}
#[test]
fn a_small_frame_buffers_no_more_than_its_block() {
let payload: alloc::vec::Vec<u8> = (0..1000u32)
.map(|i| (i.wrapping_mul(2654435761) >> 24) as u8)
.collect();
let frame = frame_of(&payload);
let mut decoder = StreamingDecoder::new(frame.as_slice());
let mut out = alloc::vec::Vec::new();
let mut buf = [0u8; 256];
loop {
let n = decoder.read(&mut buf).unwrap();
if n == 0 {
break;
}
out.extend_from_slice(&buf[..n]);
}
assert_eq!(out, payload);
assert!(
decoder.input.buf.len() <= frame.len(),
"a {}-byte frame used a {}-byte input buffer",
frame.len(),
decoder.input.buf.len()
);
}
struct Interrupting<'a> {
data: &'a [u8],
interrupt_next: bool,
}
impl Read for Interrupting<'_> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize, crate::io::Error> {
self.interrupt_next = !self.interrupt_next;
if self.interrupt_next {
return Err(crate::io::Error::from(crate::io::ErrorKind::Interrupted));
}
self.data.read(buf)
}
}
#[test]
fn read_retries_an_interrupted_source_everywhere() {
let mut stream = skippable_frame(b"leading");
stream.extend(frame_of(b"one "));
stream.extend(skippable_frame(b"between"));
stream.extend(frame_of(b"two"));
let source = Interrupting {
data: &stream,
interrupt_next: false,
};
let mut decoder = StreamingDecoder::new(source);
let mut out = alloc::vec::Vec::new();
let mut buf = [0u8; 64];
loop {
let n = decoder.read(&mut buf).unwrap();
if n == 0 {
break;
}
out.extend_from_slice(&buf[..n]);
}
assert_eq!(out, b"one two");
}
#[test]
fn a_truncated_skippable_frame_is_an_error() {
let mut leading = skippable_frame(b"metadata");
leading.truncate(leading.len() - 3);
let mut trailing = frame_of(b"payload");
trailing.extend_from_slice(&leading);
for stream in [leading.as_slice(), trailing.as_slice()] {
let decoder = StreamingDecoder::new(stream);
let err =
drain_with_read(decoder).expect_err("a skippable frame cut short must not end cleanly");
assert!(
alloc::format!("{err:?}").contains("FailedToSkipFrame"),
"{err:?}"
);
}
}
#[test]
fn a_source_error_around_a_leading_skippable_frame_fails_the_read() {
struct FailsAfter<'a> {
data: &'a [u8],
}
impl Read for FailsAfter<'_> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize, crate::io::Error> {
if self.data.is_empty() {
return Err(crate::io::Error::from(crate::io::ErrorKind::Other));
}
self.data.read(buf)
}
}
let skippable = skippable_frame(b"metadata");
for data in [&skippable[..skippable.len() - 3], &skippable[..]] {
let mut decoder = StreamingDecoder::new(FailsAfter { data });
let err = decoder
.read(&mut [0u8; 16])
.expect_err("a failing source must fail the read");
assert_eq!(err.kind(), crate::io::ErrorKind::Other);
assert!(
err.get_ref().is_none(),
"the source's own error, not a decode error: {err:?}"
);
}
}
#[test]
fn into_inner_before_any_read_returns_the_source_untouched() {
use crate::decoding::{Dictionary, DictionaryHandle};
let mut leading_skippable = skippable_frame(b"metadata");
leading_skippable.extend(frame_of(b"payload"));
let plain = frame_of(b"payload");
for stream in [plain.as_slice(), leading_skippable.as_slice()] {
let mut source = Counting {
data: stream,
reads: 0,
};
let returned = StreamingDecoder::new(&mut source).into_inner();
assert_eq!(returned.reads, 0);
assert_eq!(returned.data, stream);
}
let handle = DictionaryHandle::from_dictionary(
Dictionary::from_raw_content(7, b"dictionary content".to_vec()).unwrap(),
);
let decoder = StreamingDecoder::new_with_dictionary_handle(plain.as_slice(), &handle).unwrap();
assert_eq!(decoder.into_inner(), plain.as_slice());
}
#[test]
fn a_refused_dictionary_hands_the_source_back_unread() {
use crate::decoding::{Dictionary, DictionaryHandle, FrameDecoder};
let mut refused = Dictionary::from_raw_content(7, b"dictionary content".to_vec()).unwrap();
refused.offset_hist[0] = 0;
let refused = DictionaryHandle::from_dictionary(refused);
let plain = frame_of(b"payload");
let counting = || Counting {
data: plain.as_slice(),
reads: 0,
};
let unread = |source: &Counting<'_>| source.reads == 0 && source.data == plain.as_slice();
let error = StreamingDecoder::new_with_dictionary_handle(counting(), &refused)
.err()
.expect("a zero repeat offset is refused");
assert!(
alloc::format!("{error}").contains("epeat"),
"the decoder's error shows through: {error}"
);
assert!(unread(&error.into_source()));
let mut decoder = FrameDecoder::new();
decoder.set_max_window_size(1 << 20).unwrap();
let (source, decoder, error) =
StreamingDecoder::new_with_decoder_and_dictionary_handle(counting(), decoder, &refused)
.err()
.expect("a zero repeat offset is refused")
.into_parts();
assert!(unread(&source));
assert_eq!(
decoder.max_window_size(),
1 << 20,
"the same decoder comes back"
);
assert!(alloc::format!("{error:?}").contains("ZeroRepeatOffset"));
let truncated = 0xEC30_A437u32.to_le_bytes();
let error = StreamingDecoder::new_with_dictionary_bytes(counting(), &truncated)
.err()
.expect("a truncated dictionary is refused");
assert!(unread(&error.into_source()));
}
#[test]
fn a_damaged_first_frame_fails_the_read_and_keeps_the_source() {
let mut stream = alloc::vec![0x28, 0xB5, 0x2F, 0xFD, 0x00, 0xF8];
stream.extend_from_slice(b"the next entry");
let mut decoder = StreamingDecoder::new(stream.as_slice());
let err = decoder
.read(&mut [0u8; 16])
.expect_err("a frame header no decoder accepts must fail");
assert!(
alloc::format!("{err:?}").contains("WindowSizeTooBig")
|| alloc::format!("{err:?}").contains("FrameHeaderError"),
"{err:?}"
);
assert_eq!(decoder.into_inner(), b"the next entry");
}
#[test]
fn an_empty_source_is_an_error() {
let err = StreamingDecoder::new(&[][..])
.read(&mut [0u8; 16])
.expect_err("an empty source is not a stream");
assert!(
alloc::format!("{err:?}").contains("MagicNumberReadError"),
"{err:?}"
);
let mut out = alloc::vec::Vec::new();
let err = StreamingDecoder::new(&[][..])
.read_to_end(&mut out)
.expect_err("an empty source is not a stream");
assert!(
alloc::format!("{err:?}").contains("MagicNumberReadError"),
"{err:?}"
);
}
#[test]
fn a_reused_decoder_applies_the_forced_dictionary_to_every_first_frame() {
use crate::decoding::{Dictionary, DictionaryHandle, FrameDecoder};
use crate::encoding::{CompressionLevel, FrameCompressor};
let content = b"shared words the frames reuse again and again".to_vec();
let handle = DictionaryHandle::from_dictionary(
Dictionary::from_raw_content(0, content.clone()).unwrap(),
);
let mut compressor: FrameCompressor = FrameCompressor::new(CompressionLevel::Default);
compressor
.set_dictionary(Dictionary::from_raw_content(0, content).unwrap())
.unwrap();
let frame = compressor.compress_independent_frame(b"the frames reuse shared words");
let mut frame_decoder = FrameDecoder::new();
for _ in 0..3 {
let mut decoder = StreamingDecoder::new_with_decoder_and_dictionary_handle(
frame.as_slice(),
&mut frame_decoder,
&handle,
)
.unwrap();
let mut out = alloc::vec::Vec::new();
let mut buf = [0u8; 64];
loop {
let n = decoder.read(&mut buf).unwrap();
if n == 0 {
break;
}
out.extend_from_slice(&buf[..n]);
}
assert_eq!(out, b"the frames reuse shared words");
}
}
#[test]
fn read_rejects_a_stream_cut_inside_the_next_frame_header() {
let mut stream = frame_of(b"payload");
let next = frame_of(b"next");
stream.extend_from_slice(&next[..5]);
let decoder = StreamingDecoder::new(stream.as_slice());
assert!(drain_with_read(decoder).is_err());
}
#[test]
fn read_rejects_a_block_that_does_not_decode() {
let frame = [
0x28, 0xB5, 0x2F, 0xFD, 0x20, 16, 0x25, 0x00, 0x00, 0xFF, 0xFF, 0xFF, 0xFF,
];
let decoder = StreamingDecoder::new(&frame[..]);
assert!(drain_with_read(decoder).is_err());
}
#[test]
fn read_rejects_a_reserved_block_without_waiting_for_its_body() {
let header = (10u32 << 3) | (3 << 1) | 1;
let mut frame = alloc::vec![0x28, 0xB5, 0x2F, 0xFD, 0x20, 16];
frame.extend_from_slice(&header.to_le_bytes()[..3]);
let mut decoder = StreamingDecoder::new(OpenPipe { data: &frame });
let mut buf = [0u8; 64];
let err = decoder
.read(&mut buf)
.expect_err("a reserved block type must fail");
assert_ne!(err.kind(), crate::io::ErrorKind::WouldBlock, "{err:?}");
}
#[test]
fn read_rejects_an_oversized_rle_block_without_waiting_for_its_byte() {
let header = (100u32 << 3) | (1 << 1) | 1;
let mut frame = alloc::vec![0x28, 0xB5, 0x2F, 0xFD, 0x20, 16];
frame.extend_from_slice(&header.to_le_bytes()[..3]);
let mut decoder = StreamingDecoder::new(OpenPipe { data: &frame });
let mut buf = [0u8; 64];
let err = decoder
.read(&mut buf)
.expect_err("an RLE block over the block maximum must fail");
assert_ne!(err.kind(), crate::io::ErrorKind::WouldBlock, "{err:?}");
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_releases_the_compressed_stream() {
let payload: alloc::vec::Vec<u8> = (0..50_000u32)
.map(|i| (i.wrapping_mul(2654435761) >> 24) as u8)
.collect();
let stream = frame_of(&payload);
let mut decoder = StreamingDecoder::new(stream.as_slice());
let mut out = alloc::vec::Vec::new();
decoder.read_to_end(&mut out).unwrap();
assert_eq!(out, payload);
assert_eq!(decoder.input.buf.capacity(), 0);
}
struct Counting<'a> {
data: &'a [u8],
reads: usize,
}
impl Read for Counting<'_> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize, crate::io::Error> {
self.reads += 1;
self.data.read(buf)
}
}
#[test]
fn a_leading_skippable_frame_is_skipped_in_large_chunks() {
let mut stream = skippable_frame(&alloc::vec![0u8; 64 * 1024]);
stream.extend(frame_of(b"after the metadata"));
let mut source = Counting {
data: &stream,
reads: 0,
};
let mut decoder = StreamingDecoder::new(&mut source);
let mut out = alloc::vec::Vec::new();
let mut buf = [0u8; 64];
loop {
let n = decoder.read(&mut buf).unwrap();
if n == 0 {
break;
}
out.extend_from_slice(&buf[..n]);
}
assert_eq!(out, b"after the metadata");
let chunks = 64 * 1024 / super::SKIP_CHUNK as usize;
assert!(
source.reads <= chunks + 8,
"{} reads for a 64 KiB skippable frame",
source.reads
);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_rejects_a_non_zstd_source_before_buffering_it() {
let not_zstd = alloc::vec![0xABu8; 1 << 20];
let mut source = Counting {
data: ¬_zstd,
reads: 0,
};
let mut out = alloc::vec::Vec::new();
let err = StreamingDecoder::new(&mut source)
.read_to_end(&mut out)
.expect_err("bytes that are not a frame must be refused");
assert!(
alloc::format!("{err:?}").contains("BadMagicNumber"),
"{err:?}"
);
let taken = not_zstd.len() - source.data.len();
assert!(taken <= 18, "{taken} bytes taken to refuse a magic number");
assert!(out.is_empty());
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_after_the_end_of_the_stream_appends_nothing() {
let frame = frame_of(b"one frame");
let mut decoder = StreamingDecoder::new(frame.as_slice());
let mut out = alloc::vec::Vec::new();
decoder.read_to_end(&mut out).unwrap();
assert_eq!(out, b"one frame");
assert_eq!(decoder.read_to_end(&mut out).unwrap(), 0);
assert_eq!(out, b"one frame");
}
fn incompressible(len: usize) -> alloc::vec::Vec<u8> {
let mut x = 0x9E37_79B9_7F4A_7C15u64;
(0..len)
.map(|_| {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
(x >> 32) as u8
})
.collect()
}
#[test]
fn read_resumes_raw_blocks_wherever_a_would_block_interrupts_them() {
let payload = incompressible(300_000);
let mut compressor =
crate::encoding::FrameCompressor::new(crate::encoding::CompressionLevel::Fastest);
compressor.set_content_checksum(true);
compressor.set_source(payload.as_slice());
let mut stream = alloc::vec::Vec::new();
compressor.set_drain(&mut stream);
compressor.compress();
assert!(
stream.len() > payload.len(),
"the frame must hold Raw blocks"
);
stream.extend(frame_of(b"next"));
let descriptor = crate::decoding::frame::FrameDescriptor(stream[4]);
let first_header = 5
+ usize::from(!descriptor.single_segment_flag())
+ usize::from(descriptor.dictionary_id_bytes().unwrap())
+ usize::from(descriptor.frame_content_size_bytes().unwrap());
let stops: alloc::vec::Vec<usize> = (first_header..stream.len()).collect();
let source = Arrivals {
data: &stream,
served: 0,
stops: &stops,
};
let mut decoder = StreamingDecoder::new(source);
#[cfg(feature = "hash")]
decoder
.decoder_mut()
.set_content_checksum(crate::decoding::ContentChecksum::Verify);
let mut out = alloc::vec::Vec::new();
let mut buf = [0u8; 4096];
loop {
match decoder.read(&mut buf) {
Ok(0) => break,
Ok(n) => out.extend_from_slice(&buf[..n]),
Err(e) if e.kind() == crate::io::ErrorKind::WouldBlock => {}
Err(e) => panic!("Raw blocks must survive WouldBlock: {e:?}"),
}
}
let mut expected = payload;
expected.extend_from_slice(b"next");
assert!(out == expected, "decoded {} bytes", out.len());
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_after_a_part_read_raw_block_is_complete() {
let payload = incompressible(20_000);
let frame = frame_of(&payload);
let descriptor = crate::decoding::frame::FrameDescriptor(frame[4]);
let header = 5
+ usize::from(!descriptor.single_segment_flag())
+ usize::from(descriptor.dictionary_id_bytes().unwrap())
+ usize::from(descriptor.frame_content_size_bytes().unwrap());
let stops = [header + 3 + 100];
let source = Arrivals {
data: &frame,
served: 0,
stops: &stops,
};
let mut decoder = StreamingDecoder::new(source);
let mut buf = [0u8; 64];
let mut out = alloc::vec::Vec::new();
loop {
match decoder.read(&mut buf) {
Err(e) if e.kind() == crate::io::ErrorKind::WouldBlock => break,
Ok(n) => out.extend_from_slice(&buf[..n]),
Err(e) => panic!("{e:?}"),
}
}
assert_eq!(
out,
payload[..100],
"every byte is handed out as it arrives"
);
decoder.read_to_end(&mut out).unwrap();
assert!(out == payload, "decoded {} bytes", out.len());
}
#[test]
fn read_rejects_a_frame_cut_inside_a_raw_block() {
let payload = incompressible(20_000);
let mut frame = frame_of(&payload);
frame.truncate(frame.len() - 100);
let decoder = StreamingDecoder::new(frame.as_slice());
let err = drain_with_read(decoder).expect_err("a cut Raw block must fail");
assert!(
alloc::format!("{err:?}").contains("UnexpectedEof"),
"{err:?}"
);
}
#[cfg(all(feature = "lsm", feature = "hash"))]
#[test]
fn read_records_per_block_checksums_for_every_block() {
let mut payload = b"compressible line of text, again and again\n".repeat(8_000);
payload.extend(incompressible(300_000));
let frame = crate::encoding::compress_to_vec(
payload.as_slice(),
crate::encoding::CompressionLevel::Fastest,
);
let mut whole = crate::decoding::FrameDecoder::new();
whole.enable_per_block_checksums();
let mut out = alloc::vec![0u8; payload.len()];
whole.decode_all(&frame, &mut out).unwrap();
let expected = whole.computed_block_checksums().to_vec();
assert!(expected.len() > 2, "the frame must hold several blocks");
let mut frame_decoder = crate::decoding::FrameDecoder::new();
frame_decoder.enable_per_block_checksums();
let mut decoder = StreamingDecoder::new_with_decoder(frame.as_slice(), frame_decoder);
let mut decoded = alloc::vec::Vec::new();
let mut buf = [0u8; 4096];
loop {
let n = decoder.read(&mut buf).unwrap();
if n == 0 {
break;
}
decoded.extend_from_slice(&buf[..n]);
}
assert!(decoded == payload);
assert_eq!(
decoder.decoder.computed_block_checksums(),
expected.as_slice()
);
}
struct Arrivals<'a> {
data: &'a [u8],
served: usize,
stops: &'a [usize],
}
impl Read for Arrivals<'_> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize, crate::io::Error> {
let limit = match self.stops.first() {
Some(&stop) if stop == self.served => {
self.stops = &self.stops[1..];
return Err(crate::io::Error::from(crate::io::ErrorKind::WouldBlock));
}
Some(&stop) => stop,
None => self.data.len(),
};
let n = buf.len().min(limit - self.served);
buf[..n].copy_from_slice(&self.data[self.served..self.served + n]);
self.served += n;
Ok(n)
}
}
#[test]
fn a_flushed_frame_in_progress_hands_over_everything_so_far() {
let payload: alloc::vec::Vec<u8> = (0..50_000u32).flat_map(|i| (i / 7).to_le_bytes()).collect();
let mut context =
crate::encoding::CompressionContext::new(crate::encoding::CompressionLevel::Default);
let mut frame = alloc::vec::Vec::new();
context.write(&mut frame, &payload).unwrap();
context.flush(&mut frame).unwrap();
let stops = [frame.len()];
let source = Arrivals {
data: &frame,
served: 0,
stops: &stops,
};
let mut decoder = StreamingDecoder::new(source);
let mut buf = [0u8; 4096];
let mut out = alloc::vec::Vec::new();
loop {
match decoder.read(&mut buf) {
Err(e) if e.kind() == crate::io::ErrorKind::WouldBlock => break,
Ok(n) => out.extend_from_slice(&buf[..n]),
Err(e) => panic!("{e:?}"),
}
}
assert!(
out == payload,
"handed over {} of {} bytes",
out.len(),
payload.len()
);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_resumes_after_the_source_would_block() {
let payload: alloc::vec::Vec<u8> = (0..20_000u32).map(|i| (i % 251) as u8).collect();
let frame = frame_of(&payload);
let descriptor = crate::decoding::frame::FrameDescriptor(frame[4]);
let header = 5
+ usize::from(!descriptor.single_segment_flag())
+ usize::from(descriptor.dictionary_id_bytes().unwrap())
+ usize::from(descriptor.frame_content_size_bytes().unwrap());
let stops = [header + 10, header + 50];
let source = Arrivals {
data: &frame,
served: 0,
stops: &stops,
};
let mut decoder = StreamingDecoder::new(source);
let mut buf = [0u8; 64];
let err = decoder
.read(&mut buf)
.expect_err("the first block is not all here");
assert_eq!(err.kind(), crate::io::ErrorKind::WouldBlock);
let mut out = alloc::vec::Vec::new();
let err = decoder
.read_to_end(&mut out)
.expect_err("the source stops again");
assert_eq!(err.kind(), crate::io::ErrorKind::WouldBlock);
out.clear();
decoder.read_to_end(&mut out).unwrap();
assert!(out == payload, "decoded {} bytes", out.len());
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_returns_the_source_error() {
let frame = frame_of(b"complete frame");
let mut decoder = StreamingDecoder::new(OpenPipe { data: &frame });
let mut out = alloc::vec::Vec::new();
let err = decoder
.read_to_end(&mut out)
.expect_err("a source that would block cannot be read to its end");
assert_eq!(err.kind(), crate::io::ErrorKind::WouldBlock);
}
#[test]
fn read_rejects_a_frame_cut_inside_its_checksum() {
let mut compressor =
crate::encoding::FrameCompressor::new(crate::encoding::CompressionLevel::Fastest);
compressor.set_content_checksum(true);
compressor.set_source(&b"checksummed payload"[..]);
let mut frame = alloc::vec::Vec::new();
compressor.set_drain(&mut frame);
compressor.compress();
frame.truncate(frame.len() - 2);
let decoder = StreamingDecoder::new(frame.as_slice());
let err = drain_with_read(decoder).expect_err("a frame missing checksum bytes must fail");
assert!(
alloc::format!("{err:?}").contains("FailedToReadChecksum"),
"{err:?}"
);
}
#[test]
fn a_dictionary_decoder_rejects_bytes_that_are_not_a_frame() {
use crate::decoding::{Dictionary, DictionaryHandle};
let handle = DictionaryHandle::from_dictionary(
Dictionary::from_raw_content(7, b"dictionary content".to_vec()).unwrap(),
);
let decoder =
StreamingDecoder::new_with_dictionary_handle(&b"not a frame"[..], &handle).unwrap();
assert!(drain_with_read(decoder).is_err());
}
#[test]
fn read_delivers_a_finished_frame_before_the_next_frame_arrives() {
let frame = frame_of(b"complete frame");
let mut decoder = StreamingDecoder::new(OpenPipe { data: &frame });
let mut buf = [0u8; 64];
let n = decoder.read(&mut buf).unwrap();
assert_eq!(&buf[..n], b"complete frame");
}
#[test]
fn read_rejects_a_frame_shorter_than_its_declared_size() {
for single_segment in [false, true] {
let frame = frame_declaring(64 << 20, single_segment, &[]);
let decoder = StreamingDecoder::new(frame.as_slice());
let err = drain_with_read(decoder)
.expect_err("a frame that produced less than it declared must not end cleanly");
assert!(
alloc::format!("{err:?}").contains("FrameContentSizeMismatch"),
"single_segment={single_segment}: {err:?}"
);
}
}
#[cfg(feature = "std")]
#[test]
fn read_after_a_direct_read_to_end_ends_cleanly() {
use crate::encoding::{CompressionLevel, compress_to_vec};
let payload: alloc::vec::Vec<u8> = (0..20_000u32).map(|i| (i % 251) as u8).collect();
let frame = compress_to_vec(payload.as_slice(), CompressionLevel::Fastest);
let mut decoder = StreamingDecoder::new(frame.as_slice());
let mut out = alloc::vec::Vec::new();
decoder.read_to_end(&mut out).unwrap();
assert_eq!(out, payload);
let mut buf = [0u8; 16];
assert_eq!(decoder.read(&mut buf).unwrap(), 0);
}
#[test]
fn read_rejects_a_frame_longer_than_its_declared_size() {
let frame = frame_declaring(1, false, b"abcd");
let decoder = StreamingDecoder::new(frame.as_slice());
let err = drain_with_read(decoder)
.expect_err("a frame that produced more than it declared must not end cleanly");
assert!(
alloc::format!("{err:?}").contains("FrameContentSizeMismatch"),
"{err:?}"
);
}
#[cfg(feature = "hash")]
#[test]
fn read_delivering_bytes_defers_checksum_error_to_next_call() {
use crate::decoding::ContentChecksum;
use crate::encoding::{CompressionLevel, FrameCompressor};
use crate::io::ErrorKind;
use alloc::vec;
use alloc::vec::Vec;
let payload: Vec<u8> = (0..8192u32).map(|i| (i & 0xFF) as u8).collect();
let mut compressor = FrameCompressor::new(CompressionLevel::Default);
compressor.set_content_checksum(true);
compressor.set_source(payload.as_slice());
let mut compressed = Vec::new();
compressor.set_drain(&mut compressed);
compressor.compress();
let last = compressed.len() - 1;
compressed[last] ^= 0xFF;
let mut decoder = StreamingDecoder::new(compressed.as_slice());
decoder
.decoder
.set_content_checksum(ContentChecksum::Verify);
let mut buf = vec![0u8; payload.len() + 4096];
let n = decoder
.read(&mut buf)
.expect("a read that delivered bytes must not return the checksum Err");
assert_eq!(n, payload.len());
assert_eq!(&buf[..n], payload.as_slice());
let err = decoder
.read(&mut buf)
.expect_err("deferred checksum mismatch must surface on the terminating read");
assert_eq!(err.kind(), ErrorKind::Other);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_decode_in_place_matches_and_takes_direct_path() {
use crate::encoding::{CompressionLevel, FrameCompressor};
use alloc::vec::Vec;
let payload: Vec<u8> = (0..20_000u32)
.map(|i| (i.wrapping_mul(2654435761) >> 24) as u8)
.collect();
let mut compressor = FrameCompressor::new(CompressionLevel::Default);
compressor.set_source(payload.as_slice());
let mut compressed = Vec::new();
compressor.set_drain(&mut compressed);
compressor.compress();
let mut decoder = StreamingDecoder::new(compressed.as_slice());
let mut out = Vec::new();
let n = decoder.read_to_end(&mut out).unwrap();
assert_eq!(n, payload.len());
assert_eq!(out, payload);
assert_eq!(decoder.decoder.direct_frames(), 1);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_after_partial_read_is_complete() {
use crate::encoding::{CompressionLevel, FrameCompressor};
use alloc::vec;
use alloc::vec::Vec;
let payload: Vec<u8> = (0..20_000u32).map(|i| (i & 0xFF) as u8).collect();
let mut compressor = FrameCompressor::new(CompressionLevel::Default);
compressor.set_source(payload.as_slice());
let mut compressed = Vec::new();
compressor.set_drain(&mut compressed);
compressor.compress();
let mut decoder = StreamingDecoder::new(compressed.as_slice());
let mut head = vec![0u8; 4096];
let got = decoder.read(&mut head).unwrap();
assert!(got > 0 && got <= head.len());
let mut out = Vec::new();
out.extend_from_slice(&head[..got]);
decoder.read_to_end(&mut out).unwrap();
assert_eq!(out, payload);
assert_eq!(decoder.decoder.direct_frames(), 0);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_decodes_all_concatenated_frames() {
use crate::encoding::{CompressionLevel, compress_slice_to_vec};
use alloc::vec::Vec;
let a: Vec<u8> = (0..5000u32).map(|i| (i & 0xFF) as u8).collect();
let b: Vec<u8> = (0..3000u32)
.map(|i| ((i.wrapping_mul(7)) & 0xFF) as u8)
.collect();
let mut stream = compress_slice_to_vec(&a, CompressionLevel::Level(3));
stream.extend_from_slice(&compress_slice_to_vec(&b, CompressionLevel::Level(3)));
let mut decoder = StreamingDecoder::new(stream.as_slice());
let mut out = Vec::new();
decoder.read_to_end(&mut out).unwrap();
let mut expected = a.clone();
expected.extend_from_slice(&b);
assert_eq!(out, expected);
assert_eq!(decoder.decoder.direct_frames(), 2);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_after_partial_read_decodes_all_concatenated_frames() {
use crate::encoding::{CompressionLevel, compress_slice_to_vec};
use alloc::vec;
use alloc::vec::Vec;
let a: Vec<u8> = (0..6000u32).map(|i| (i & 0xFF) as u8).collect();
let b: Vec<u8> = (0..4000u32)
.map(|i| ((i.wrapping_mul(11)) & 0xFF) as u8)
.collect();
let mut stream = compress_slice_to_vec(&a, CompressionLevel::Level(3));
stream.extend_from_slice(&compress_slice_to_vec(&b, CompressionLevel::Level(3)));
let mut decoder = StreamingDecoder::new(stream.as_slice());
let mut head = vec![0u8; 2048];
let got = decoder.read(&mut head).unwrap();
assert!(got > 0 && got <= head.len());
let mut out = Vec::new();
out.extend_from_slice(&head[..got]);
decoder.read_to_end(&mut out).unwrap();
let mut expected = a.clone();
expected.extend_from_slice(&b);
assert_eq!(
out, expected,
"fallback path must decode frame 2 too, not stop at frame 1 EOF"
);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_concatenated_dict_frames_decode_with_dictionary() {
use crate::encoding::{CompressionLevel, FrameCompressor};
use alloc::vec::Vec;
let dict_raw = include_bytes!("../../../dict_tests/dictionary");
let compress_with_dict = |payload: &[u8]| -> Vec<u8> {
let mut compressor = FrameCompressor::new(CompressionLevel::Default);
compressor
.set_dictionary_from_bytes(dict_raw)
.expect("dict load");
compressor.set_source(payload);
let mut compressed = Vec::new();
compressor.set_drain(&mut compressed);
compressor.compress();
compressed
};
let a = b"first dictionary-compressed frame payload".to_vec();
let b = b"second dictionary-compressed frame payload".to_vec();
let mut stream = compress_with_dict(&a);
stream.extend_from_slice(&compress_with_dict(&b));
let mut decoder =
StreamingDecoder::new_with_dictionary_bytes(stream.as_slice(), dict_raw).unwrap();
let mut out = Vec::new();
decoder
.read_to_end(&mut out)
.expect("both dict frames must decode with the forced dictionary");
let mut expected = a.clone();
expected.extend_from_slice(&b);
assert_eq!(out, expected);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_truncates_output_on_direct_decode_error() {
use crate::encoding::{CompressionLevel, FrameCompressor};
use alloc::vec::Vec;
let payload: Vec<u8> = (0..5000u32).map(|i| (i & 0xFF) as u8).collect();
let mut compressor = FrameCompressor::new(CompressionLevel::Default);
compressor.set_source(payload.as_slice());
let mut compressed = Vec::new();
compressor.set_drain(&mut compressed);
compressor.compress();
compressed.truncate(compressed.len() - 40);
let mut decoder = StreamingDecoder::new(compressed.as_slice());
let mut out = b"SENTINEL".to_vec();
let result = decoder.read_to_end(&mut out);
assert!(result.is_err(), "truncated block must fail the decode");
assert_eq!(
out, b"SENTINEL",
"failed direct decode must not append non-decoded bytes to output"
);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_truncates_output_on_midframe_fallback_error() {
use crate::encoding::{CompressionLevel, CompressionParameters, FrameCompressor};
use alloc::vec;
use alloc::vec::Vec;
let payload: Vec<u8> = (0..320_000u32)
.map(|i| (i.wrapping_mul(2654435761) >> 24) as u8)
.collect();
let params = CompressionParameters::builder(CompressionLevel::Default)
.window_log(17)
.build()
.expect("window_log within bounds");
let mut compressor = FrameCompressor::new(CompressionLevel::Default);
compressor.set_parameters(¶ms);
compressor.set_source(payload.as_slice());
let mut compressed = Vec::new();
compressor.set_drain(&mut compressed);
compressor.compress();
compressed.truncate(compressed.len() - 40);
let mut decoder = StreamingDecoder::new(compressed.as_slice());
let mut head = vec![0u8; 4096];
let got = decoder.read(&mut head).unwrap();
assert!(got > 0);
let mut out = Vec::new();
out.extend_from_slice(&head[..got]);
let result = decoder.read_to_end(&mut out);
assert!(
result.is_err(),
"truncated current frame must fail the decode"
);
assert!(
out.len() <= payload.len(),
"failed fallback read must not leave a zero-filled tail (len {} > payload {})",
out.len(),
payload.len()
);
assert_eq!(
out.as_slice(),
&payload[..out.len()],
"decoded prefix must match the payload, with no appended non-decoded bytes"
);
}
#[cfg(feature = "std")]
#[test]
fn read_to_end_empty_frame_decodes_to_empty() {
use crate::encoding::{CompressionLevel, compress_slice_to_vec};
use alloc::vec::Vec;
let compressed = compress_slice_to_vec(&[], CompressionLevel::Level(3));
let mut decoder = StreamingDecoder::new(compressed.as_slice());
let mut out = Vec::new();
decoder.read_to_end(&mut out).unwrap();
assert!(out.is_empty());
}
#[test]
fn every_finished_frame_is_reported_once_with_its_sizes() {
use super::FinishedFrame;
use crate::encoding::{CompressionLevel, compress_slice_to_vec};
use alloc::vec::Vec;
let first_payload = b"the first frame of the stream".repeat(40);
let second_payload = b"and the second one".repeat(3);
let first = compress_slice_to_vec(&first_payload, CompressionLevel::Level(3));
let second = compress_slice_to_vec(&second_payload, CompressionLevel::Level(1));
let mut stream = first.clone();
stream.extend_from_slice(&0x184D_2A50u32.to_le_bytes());
stream.extend_from_slice(&3u32.to_le_bytes());
stream.extend_from_slice(b"pad");
stream.extend_from_slice(&second);
let mut decoder = StreamingDecoder::new(stream.as_slice());
let mut buf = [0u8; 7];
let mut finished = Vec::new();
let mut decoded = 0usize;
loop {
let n = decoder
.read_reporting_frames(&mut buf, |frame| finished.push(frame))
.unwrap();
if n == 0 {
break;
}
decoded += n;
}
assert_eq!(decoded, first_payload.len() + second_payload.len());
assert_eq!(
finished,
[
FinishedFrame {
compressed_size: first.len() as u64,
decompressed_size: first_payload.len() as u64,
},
FinishedFrame {
compressed_size: second.len() as u64,
decompressed_size: second_payload.len() as u64,
},
]
);
let mut again = Vec::new();
let end = decoder
.read_reporting_frames(&mut buf, |frame| again.push(frame))
.unwrap();
assert_eq!((end, again.as_slice()), (0, &[][..]), "reported once");
}
#[test]
fn consecutive_empty_frames_are_each_reported() {
use super::FinishedFrame;
use crate::encoding::{CompressionLevel, compress_slice_to_vec};
use alloc::vec::Vec;
let empty = compress_slice_to_vec(b"", CompressionLevel::Level(3));
let payload = b"after two empty frames";
let last = compress_slice_to_vec(payload, CompressionLevel::Level(3));
let mut stream = empty.clone();
stream.extend_from_slice(&empty);
stream.extend_from_slice(&last);
let mut decoder = StreamingDecoder::new(stream.as_slice());
let mut buf = [0u8; 64];
let mut finished = Vec::new();
while decoder
.read_reporting_frames(&mut buf, |frame| finished.push(frame))
.unwrap()
> 0
{}
let empty_frame = FinishedFrame {
compressed_size: empty.len() as u64,
decompressed_size: 0,
};
assert_eq!(
finished,
[
empty_frame,
empty_frame,
FinishedFrame {
compressed_size: last.len() as u64,
decompressed_size: payload.len() as u64,
},
]
);
}