structured-zstd 0.2.5

use super::StreamingDecoder;
use crate::io::Read;

/// A frame with an 8-byte `Frame_Content_Size` of `fcs` followed by one raw
/// last block carrying `payload`. Multi-segment frames take a 1 KiB window
/// descriptor; single-segment ones size their window by the FCS.
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
}

/// Drain `decoder` through plain `read` calls the way `io::copy` does, up to
/// the first `Ok(0)` or error.
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]);
    }
}

/// One compressed frame of `payload`.
fn frame_of(payload: &[u8]) -> alloc::vec::Vec<u8> {
    crate::encoding::compress_to_vec(payload, crate::encoding::CompressionLevel::Fastest)
}

/// A skippable frame (RFC 8878 3.1.2) carrying `payload`.
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
}

/// RFC 8878 3: a stream is one or more frames. Plain `read`, the path
/// `io::copy` takes, must deliver every frame's content, not stop at the end
/// of the first one.
#[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"
    );
}

/// Skippable frames carry no content and are skipped wherever they stand,
/// first in the stream included; a skippable frame last ends the stream
/// cleanly.
#[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");
}

/// A stream of skippable frames alone holds no content: it decodes to
/// nothing, through `read` and through `read_to_end`, as upstream decodes it.
#[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());
}

/// Bytes after the last frame that do not form a frame are an error, not a
/// clean end of stream.
#[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());
}

/// A decoder built with a dictionary applies it to every frame, not just the
/// first, as `read_to_end` already does.
#[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"
    );
}

/// A source that hands out its bytes and then reports that nothing more is
/// available yet, as an open pipe or socket would.
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)
    }
}

/// A non-blocking source that hands out one byte per call and reports
/// `WouldBlock` on every other call, so every frame header, skippable frame
/// and block arrives in pieces.
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)
    }
}

/// A `WouldBlock` anywhere, inside the first frame's header, a block, a block
/// header, a checksum, a skippable frame or the next frame's header, loses
/// nothing: retrying the read picks up where the source stopped. Every byte
/// of the stream trickles in between `WouldBlock`s.
#[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());
}

/// Magicless frames carry no magic number, so a frame's header is as long as
/// its descriptor says from its first byte. Ten-byte magicless headers
/// (window descriptor and an eight-byte content size), the first one
/// included, handed over a byte at a time between `WouldBlock`s must still
/// start their frames.
#[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");
}

/// The reader takes from its source only what the next step of the frame
/// needs, as upstream's `ZSTD_decompressStream` loads only the next step's
/// input. Once a frame's content is delivered the source stands right after
/// that frame, so `into_inner` hands back the rest of the stream untouched.
#[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");
}

/// A checksummed frame read into a buffer of exactly its content size takes
/// its four-byte checksum with its last block, so the source stands right
/// after the frame, not at the checksum.
#[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");
}

/// A small frame is read into a buffer the size of its block, not one sized
/// for the largest block any frame may hold.
#[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()
    );
}

/// A source that reports `Interrupted` before every read it serves, so each
/// read path's retry runs: the header reads, the skip of a skippable frame,
/// the boundary probe and the input fill.
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)
    }
}

/// `Interrupted` is not an error (the `Read` contract): every read path
/// retries it, whether the source is at a frame header, inside a skippable
/// frame, at a frame boundary or between blocks.
#[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");
}

/// A skippable frame that ends before its declared length is a truncated
/// stream, wherever it stands: in front of the first frame and after one.
#[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:?}"
        );
    }
}

/// A source that fails part-way through a leading skippable frame, or at the
/// boundary after it, fails the read with the source's own error, not with a
/// frame cut short and not with a clean empty stream.
#[test]
fn a_source_error_around_a_leading_skippable_frame_fails_the_read() {
    /// Serves `data`, then fails every read with `Other`.
    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:?}"
        );
    }
}

/// Construction reads nothing, so a decoder that was never read hands its
/// source back untouched: a plain stream, one opening with a skippable frame,
/// and one under a forced dictionary alike. This is what lets a container
/// reader hand the raw bytes on after deciding not to decompress them.
#[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());
}

/// A dictionary no frame can use (here a zero repeat offset, which RFC 8878
/// 5 forbids) is refused before anything is read, and every dictionary
/// constructor hands the source back unread with the error, and the decoder
/// too where it was passed by value. A blob that does not parse as a
/// dictionary is refused the same way.
#[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"));

    // The dictionary magic with nothing after it.
    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()));
}

/// A damaged first frame fails the first read, not the constructor, so the
/// source stays with the caller: `into_inner` hands back what the decoder did
/// not take, and the bytes after the damaged frame are still there.
#[test]
fn a_damaged_first_frame_fails_the_read_and_keeps_the_source() {
    // A frame magic with a window descriptor over any decoder's limit
    // (RFC 8878 3.1.1.1.2), then the rest of a container.
    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");
}

/// An empty source holds no frame: both read paths report it, as the
/// constructor reported it when it read the first header itself.
#[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:?}"
    );
}

/// A decoder reused across streams under one forced dictionary decodes each,
/// its first frame included, with that dictionary: the one armed before the
/// first header on a fresh decoder, and the one already held on a reused one.
#[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");
    }
}

/// The source ending part-way through the next frame's header is a truncated
/// stream, not its clean end.
#[test]
fn read_rejects_a_stream_cut_inside_the_next_frame_header() {
    let mut stream = frame_of(b"payload");
    // Magic and descriptor only: a frame header is longer than five bytes,
    // since a single-segment frame always carries its content size.
    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());
}

/// A block whose content does not decode fails the read that reaches it.
#[test]
fn read_rejects_a_block_that_does_not_decode() {
    // Single-segment frame of 16 bytes holding one last compressed block of
    // four bytes that are no valid literals section.
    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());
}

/// A block header of the reserved type (RFC 8878 3.1.1.2.2) is invalid on its
/// own: the reader reports it from the header's three bytes instead of waiting
/// on the source for a body the block does not have.
#[test]
fn read_rejects_a_reserved_block_without_waiting_for_its_body() {
    // Single-segment frame of 16 bytes, then a last block of type 3 stating
    // 10 bytes of content, within the frame's block maximum, that never follow.
    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:?}");
}

/// An RLE block's size field is its repeat count (RFC 8878 3.1.1.2.3): one
/// over the frame's block maximum is invalid from the header alone, so the
/// reader reports it without waiting for the block's one byte of content.
#[test]
fn read_rejects_an_oversized_rle_block_without_waiting_for_its_byte() {
    // Single-segment frame of 16 bytes, then a last RLE block repeating its
    // byte 100 times, past the frame's 16-byte block maximum.
    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:?}");
}

/// `read_to_end` gathers the rest of the stream in one buffer; it must not
/// stay allocated once the stream is decoded, or a decoder kept alive pins a
/// whole compressed archive.
#[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);
}

/// A source that counts the reads it serves.
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)
    }
}

/// A large skippable frame in front of the first frame is stepped over in
/// chunks of `SKIP_CHUNK`, as the ones after it are, not 512 bytes at a time.
#[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
    );
}

/// `read_to_end` on a stream not started yet validates the first frame header
/// before it gathers the rest of the source: a large input that is not zstd
/// fails after a header's worth of bytes, not after the whole input was
/// copied into memory.
#[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: &not_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());
}

/// `read_to_end` at the end of the stream appends nothing and succeeds, as
/// `read` returns `Ok(0)` there.
#[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");
}

/// `len` bytes no compressor shrinks, so frames of them hold Raw blocks.
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()
}

/// Raw blocks are read straight into the decode buffer, read by read: a
/// source that hands them over a byte at a time between `WouldBlock`s, with
/// the content checksum verified, still decodes every byte.
#[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());
    // Every byte after the first frame header arrives on its own, and every
    // other read would block.
    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());
}

/// `read_to_end` after a `read` that stopped inside the first Raw block must
/// carry on from the decode buffer, not decode the frame again from the
/// source as if nothing had been read.
#[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();
    // The Raw block's first 100 bytes are taken and handed out as they come,
    // then the source blocks.
    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());
}

/// A source that ends inside a Raw block is a truncated block body.
#[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:?}"
    );
}

/// Per-block checksums asked for on the decoder are taken for every block a
/// `read` decodes, compressed and Raw alike, in block order: the same digests
/// `decode_all` reports for the frame.
#[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()
    );
}

/// A source that serves `data` up to each of `stops` in turn, reporting
/// `WouldBlock` once at each, as a socket does between arrivals.
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)
    }
}

/// A frame flushed and not ended yet, from a source that has nothing more for
/// now, hands over every byte compressed before the flush: none of its last
/// window waits for the end of a frame that has not come.
#[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()
    );
}

/// Input a `read` gathered before a `WouldBlock`, and whatever `read_to_end`
/// takes from the source before it stops again, both stay with the decoder: a
/// retried `read_to_end` decodes the whole frame.
#[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());
}

/// `read_to_end` hands the source's own error back as it reported it.
#[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);
}

/// A frame cut inside its trailing checksum has every block decoded, and
/// still must not end as if it were whole.
#[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());
    // The cut is reported where it is, in the checksum, not as a block header
    // read out of the checksum's first bytes.
    let err = drain_with_read(decoder).expect_err("a frame missing checksum bytes must fail");
    assert!(
        alloc::format!("{err:?}").contains("FailedToReadChecksum"),
        "{err:?}"
    );
}

/// A decoder built with a dictionary still rejects a source that is not a
/// frame, at its first read.
#[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());
}

/// `read` stays a stream: a finished frame's bytes are delivered without
/// first waiting for the next frame's header from a source that is still
/// open.
#[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");
}

/// RFC 8878 3.1.1.1.4: when `Frame_Content_Size` is present the decompressed
/// data must be exactly that long. A frame that declares 64 MiB and produces
/// nothing must end the stream with an error, not a clean `Ok(0)` EOF that
/// lets `io::copy` report an empty file as decoded. Both window layouts.
#[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:?}"
        );
    }
}

/// `read_to_end` decodes a sized frame on the direct path, straight into the
/// caller's vector, outside the buffer whose counter the size check reads. A
/// `read` after it must still see the frame as complete and valid and return
/// a clean `Ok(0)`, not a size mismatch against a counter left at zero.
#[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);
}

/// The other direction: a frame that produces more than it declares. Its
/// bytes may be delivered, but the stream must not end as if it were valid.
#[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:?}"
    );
}

/// `Read::read` must not return `Err` after it has already written bytes
/// into the caller's buffer (the trait mandates that an error implies no
/// bytes were read). When a single `read` call both drains the final bytes
/// of a `Verify`-mode frame AND finishes it, a checksum mismatch must be
/// deferred: those bytes are delivered as `Ok(n)` and the error surfaces on
/// the next (zero-byte) call, where returning `Err` violates no contract.
#[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);
    // Checksum is the subject under test; the encoder default is off
    // (upstream library parity).
    compressor.set_content_checksum(true);
    compressor.set_source(payload.as_slice());
    let mut compressed = Vec::new();
    compressor.set_drain(&mut compressed);
    compressor.compress();

    // Corrupt the trailing 4-byte content checksum: the body still decodes
    // to the right bytes, but the stored digest no longer matches.
    let last = compressed.len() - 1;
    compressed[last] ^= 0xFF;

    let mut decoder = StreamingDecoder::new(compressed.as_slice());
    decoder
        .decoder
        .set_content_checksum(ContentChecksum::Verify);

    // A buffer large enough to drain the whole frame in one call: this call
    // finishes the frame AND writes every payload byte. The mismatch must
    // NOT abort it (that would drop the delivered bytes).
    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());

    // The deferred mismatch surfaces on the terminating zero-byte read.
    let err = decoder
        .read(&mut buf)
        .expect_err("deferred checksum mismatch must surface on the terminating read");
    assert_eq!(err.kind(), ErrorKind::Other);
}

/// A fresh `read_to_end` must take the single-copy decode-in-place path
/// (FCS-declared frame decoded straight into the output `Vec`, no ring
/// drain) AND reproduce the payload byte-for-byte.
#[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);
    // FrameCompressor declares FCS, so the fresh fast path used the direct
    // (decode-in-place) route, not the ring drain.
    assert_eq!(decoder.decoder.direct_frames(), 1);
}

/// `read_to_end` after a partial `read` must still produce the full
/// payload. The decoder is mid-frame, so the fast path is skipped and the
/// generic grow-and-drain fallback runs (no direct frame).
#[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);
    // Mid-frame entry → fallback path, never the direct route.
    assert_eq!(decoder.decoder.direct_frames(), 0);
}

/// `read_to_end` reads the WHOLE source to EOF: a stream of concatenated
/// frames must decode every frame, not just the first. (The fast path
/// buffers the whole source, so dropping the trailing frame would lose
/// data.)
#[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);
    // Both FCS-declared frames took the direct path.
    assert_eq!(decoder.decoder.direct_frames(), 2);
}

/// `read_to_end` after a partial `read` must STILL consume the source to
/// EOF across concatenated frames, not stop at the current frame's end. The
/// partial read forces the mid-frame fallback path; with two concatenated
/// frames the fallback must finish frame 1, then advance through frame 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());
    // Partial read of frame 1 → mid-frame, so read_to_end takes the fallback.
    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"
    );
}

/// `read_to_end` on a stream of concatenated DICTIONARY frames must decode
/// every frame WITH the dictionary the decoder was constructed with. The
/// fast-path concatenated loop re-initialises following frames, and a plain
/// re-init resolves dictionaries by frame id only — losing the forced
/// dictionary for frames that omit (or can't resolve) the id.
#[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);
}

/// A direct-path decode error must NOT leave non-decoded bytes in `output`.
/// The fast path resizes `output` to the declared content size before
/// decoding; if decode fails, the enlarged (zeroed) tail must be truncated
/// away so callers never observe bytes that were never decoded.
#[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();
    // Truncate the block bytes (the FCS-bearing header at the front stays
    // intact) so the header parses but the direct-path block decode hits a
    // premature end → error after `output` was already resized.
    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"
    );
}

/// The mid-frame fallback grows `output` by `MAX_BLOCK_SIZE` before each
/// `self.read`. When that read errors (truncated current frame), the grown
/// zero-filled tail must be truncated away before the error propagates, so
/// the caller never observes `MAX_BLOCK_SIZE` worth of bytes that were never
/// decoded.
#[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;

    // Incompressible payload with a window (128 KiB) SMALLER than the input,
    // so the frame holds several blocks and bytes become collectable while
    // the frame is still mid-decode. Without a sub-input window the decoder
    // retains the whole input until the frame finishes, and a partial read
    // could only ever finish or error, never leave a truncated remainder for
    // the fallback to trip on.
    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(&params);
    compressor.set_source(payload.as_slice());
    let mut compressed = Vec::new();
    compressor.set_drain(&mut compressed);
    compressor.compress();
    // Truncate the tail so the final block decode fails partway through.
    compressed.truncate(compressed.len() - 40);

    let mut decoder = StreamingDecoder::new(compressed.as_slice());
    // A partial `read` first: leaves the decoder mid-frame so `read_to_end`
    // takes the grow-and-drain fallback (not the decode-in-place fast path).
    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"
    );
}

/// An empty (`Frame_Content_Size = 0`) frame decodes to nothing through the
/// `read_to_end` fast path — the declared-size validation accepts the valid
/// case (produced == 0) instead of erroring.
#[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());
}

/// Read a few bytes at a time, a stream of two frames with a skippable frame
/// between them reports each frame once, in order, with the bytes it took
/// from the source and the bytes it decoded to; the skippable frame is not a
/// frame of output and is not reported.
#[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");
}

/// Empty frames deliver no bytes, so one read passes through several of them
/// on its way to output; each is still a frame of the stream and is reported,
/// in order, before the frame whose output that read returns.
#[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,
            },
        ]
    );
}