crates/gunnar-pktline/tests/stream.rs
//! `Decoder` tests for the socket case.
//!
//! A real connection hands over whatever the kernel had ready, so the framing
//! must be invariant under *every* chunking of the same byte stream. The tests
//! here drive the same bytes at one byte per push, at odd chunk sizes, and in
//! one shot, and require the identical packet sequence from all of them.
use gunnar_pktline::*;
/// The stream a receive-pack session actually sees: framed command list, a
/// flush-pkt, then a raw packfile that is not framed at all.
fn receive_pack_stream() -> (Vec<u8>, Vec<Packet>, Vec<u8>) {
let packets = vec![
Packet::Data(b"0000000000000000000000000000000000000000 1111111111111111111111111111111111111111 refs/heads/main\0report-status side-band-64k\n".to_vec()),
Packet::Data(b"2222222222222222222222222222222222222222 3333333333333333333333333333333333333333 refs/heads/next\n".to_vec()),
Packet::Flush,
];
let mut wire = Vec::new();
for p in &packets {
write_packet(&mut wire, p).unwrap();
}
// Deliberately packfile-shaped, and deliberately full of bytes that look
// like framing: a NUL run, an ASCII "0000", and a high byte.
let mut raw = b"PACK\x00\x00\x00\x02\x00\x00\x00\x030000".to_vec();
raw.extend_from_slice(&[0xff, 0x00, 0x41, 0x0a]);
raw.extend_from_slice(b"tail");
let mut all = wire.clone();
all.extend_from_slice(&raw);
(all, packets, raw)
}
/// Frame `stream` fed in `chunk`-sized pushes, stopping at the first flush-pkt
/// exactly as receive-pack does, and return the packets plus the untouched tail.
fn read_until_flush(stream: &[u8], chunk: usize) -> (Vec<Packet>, Vec<u8>) {
let mut d = Decoder::new();
let mut got = Vec::new();
let mut framed = true;
for part in stream.chunks(chunk) {
d.push(part);
while framed {
match d.next_packet().expect("decode failed") {
Some(p) => {
framed = !p.is_flush();
got.push(p);
}
None => break,
}
}
}
(got, d.into_remainder())
}
#[test]
fn one_byte_at_a_time_yields_the_exact_packet_sequence() {
let (stream, expected, raw) = receive_pack_stream();
let mut d = Decoder::new();
let mut got = Vec::new();
let mut framed = true;
for (i, b) in stream.iter().enumerate() {
d.push(&[*b]);
// Stop framing at the flush. Past it the bytes are a raw packfile and
// are nobody's packet. (That the decoder *errors* on them rather than
// inventing packets out of pack bytes is asserted separately.)
while framed {
match d.next_packet().expect("decode failed mid-stream") {
Some(p) => {
framed = !p.is_flush();
got.push((i, p));
}
None => break,
}
}
}
let seq: Vec<Packet> = got.iter().map(|(_, p)| p.clone()).collect();
assert_eq!(seq, expected);
// A packet must not appear before its final byte has been pushed: the index
// at which each packet was produced is exactly the index of its last byte.
let mut end = 0usize;
let mut ends = Vec::new();
for p in &expected {
let mut one = Vec::new();
write_packet(&mut one, p).unwrap();
end += one.len();
ends.push(end - 1);
}
assert_eq!(got.iter().map(|(i, _)| *i).collect::<Vec<_>>(), ends);
assert_eq!(d.remainder(), &raw[..]);
assert_eq!(d.into_remainder(), raw);
}
#[test]
fn every_chunking_of_the_same_stream_gives_the_same_packets() {
let (stream, expected, raw) = receive_pack_stream();
for chunk in [1usize, 2, 3, 5, 7, 13, 64, 4096, stream.len()] {
let (got, tail) = read_until_flush(&stream, chunk);
assert_eq!(got, expected, "chunk size {chunk}");
assert_eq!(tail, raw, "chunk size {chunk}");
}
}
#[test]
fn into_remainder_returns_post_flush_bytes_byte_exact() {
let (stream, expected, raw) = receive_pack_stream();
let mut d = Decoder::new();
d.push(&stream);
// Stop at the flush, exactly as receive-pack does before handing the socket
// to the pack reader.
let mut got = Vec::new();
loop {
let p = d.next_packet().unwrap().expect("stream ended before flush");
let done = p.is_flush();
got.push(p);
if done {
break;
}
}
assert_eq!(got, expected);
assert_eq!(d.pending(), raw.len());
let tail = d.into_remainder();
assert_eq!(tail, raw, "the packfile's trailing hash covers these bytes");
}
#[test]
fn a_decoder_that_consumed_nothing_returns_everything() {
// Not one byte may be swallowed on the way in, whatever the bytes look like.
for raw in [
&b"\x00\x01\x02\x03raw"[..],
&b"PACK-not-framed-at-all"[..],
&b""[..],
] {
let mut d = Decoder::new();
d.push(raw);
assert_eq!(d.pending(), raw.len());
assert_eq!(d.into_remainder(), raw.to_vec());
}
}
#[test]
fn remainder_survives_internal_compaction() {
// Long enough to cross the decoder's compaction threshold, which is where a
// buffer-shifting bug would silently eat the head of the tail.
let mut wire = Vec::new();
let mut expected = Vec::new();
for i in 0..8000u32 {
let payload = format!("line {i:06}\n");
write_text(&mut wire, payload.as_bytes()).unwrap();
expected.push(Packet::Data(payload.into_bytes()));
}
write_flush(&mut wire).unwrap();
expected.push(Packet::Flush);
let raw: Vec<u8> = (0..5000u32).map(|i| (i % 251) as u8).collect();
wire.extend_from_slice(&raw);
assert!(
wire.len() > 128 * 1024,
"must exceed the compaction threshold"
);
let (got, tail) = read_until_flush(&wire, 997);
assert_eq!(got.len(), expected.len());
assert_eq!(got, expected);
assert_eq!(tail, raw);
}
#[test]
fn drain_stops_cleanly_at_a_partial_packet_and_resumes() {
let mut d = Decoder::new();
d.push(b"0006hi0009hel");
assert_eq!(d.drain().unwrap(), vec![Packet::Data(b"hi".to_vec())]);
assert_eq!(d.pending(), 7, "the partial packet stays buffered");
d.push(b"lo");
assert_eq!(d.drain().unwrap(), vec![Packet::Data(b"hello".to_vec())]);
assert!(d.is_empty());
assert_eq!(d.drain().unwrap(), Vec::new());
}
#[test]
fn next_packet_is_none_while_a_header_is_split() {
let mut d = Decoder::new();
d.push(b"00");
assert_eq!(d.next_packet().unwrap(), None);
d.push(b"06");
assert_eq!(
d.next_packet().unwrap(),
None,
"header complete, payload is not"
);
d.push(b"h");
assert_eq!(d.next_packet().unwrap(), None);
d.push(b"i");
assert_eq!(d.next_packet().unwrap(), Some(Packet::Data(b"hi".to_vec())));
assert_eq!(d.next_packet().unwrap(), None);
}
#[test]
fn an_error_is_sticky_and_does_not_resynchronise() {
// Once the stream is garbage, "skip ahead until something parses" would
// invent packets out of payload bytes. The bad header stays put.
for bad in [&b"0003"[..], &b"zzzz"[..], &b"ffff"[..]] {
let mut d = Decoder::new();
d.push(bad);
d.push(b"0006hi");
assert!(d.next_packet().is_err());
assert!(d.next_packet().is_err(), "error must not be transient");
assert!(d.drain().is_err());
// The bytes were not consumed, so nothing was silently discarded.
assert_eq!(d.pending(), bad.len() + 6);
}
}
#[test]
fn a_good_stream_after_a_bad_prefix_is_still_readable_from_the_remainder() {
// The failure mode this guards: a decoder that consumed the bad header would
// hand back a remainder missing four bytes, and the recovery path would then
// be reading from the wrong offset.
let mut d = Decoder::new();
d.push(b"0003 leftover");
assert!(d.next_packet().is_err());
assert_eq!(d.into_remainder(), b"0003 leftover".to_vec());
}
#[test]
fn pending_and_is_empty_track_only_unconsumed_bytes() {
let mut d = Decoder::new();
assert!(d.is_empty());
assert_eq!(d.pending(), 0);
d.push(b"0006hi0000");
assert_eq!(d.pending(), 10);
d.next_packet().unwrap();
assert_eq!(d.pending(), 4);
d.next_packet().unwrap();
assert!(d.is_empty());
}