zipatch-rs 1.6.0

//! `ApplyObserver` integration coverage — moved from `src/lib.rs` inline tests
//! during the API-overhaul pass. Verifies on-chunk-applied event fields,
//! observer-driven cancellation, and the documented "apply runs before event
//! fires" ordering.

use std::io::Cursor;
use std::ops::ControlFlow;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use zipatch_rs::test_utils::{MAGIC, make_chunk};
use zipatch_rs::{ApplyConfig, ApplyError, ApplyObserver, ChunkEvent, ZiPatchReader};

struct LoggingObserver {
    log: Arc<std::sync::Mutex<Vec<ChunkEvent>>>,
}

impl ApplyObserver for LoggingObserver {
    fn on_chunk_applied(&mut self, ev: ChunkEvent) -> ControlFlow<(), ()> {
        self.log.lock().unwrap().push(ev);
        ControlFlow::Continue(())
    }
}

struct CountingBreaker {
    count: Arc<AtomicUsize>,
}

impl ApplyObserver for CountingBreaker {
    fn on_chunk_applied(&mut self, _ev: ChunkEvent) -> ControlFlow<(), ()> {
        self.count.fetch_add(1, Ordering::Relaxed);
        ControlFlow::Break(())
    }
}

struct BreakAfter {
    count: Arc<AtomicUsize>,
    threshold: usize,
}

impl ApplyObserver for BreakAfter {
    fn on_chunk_applied(&mut self, _ev: ChunkEvent) -> ControlFlow<(), ()> {
        let n = self.count.fetch_add(1, Ordering::Relaxed) + 1;
        if n >= self.threshold {
            ControlFlow::Break(())
        } else {
            ControlFlow::Continue(())
        }
    }
}

#[test]
fn observer_fires_for_each_non_eof_chunk_with_correct_fields() {
    // Two ADIR chunks — observer must receive exactly two events, in order,
    // with 0-based index, correct tag, and a monotonically increasing
    // bytes_read that matches the exact wire-frame sizes.
    let log: Arc<std::sync::Mutex<Vec<ChunkEvent>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
    let log_clone = log.clone();

    let mut a = Vec::new();
    a.extend_from_slice(&1u32.to_be_bytes());
    a.extend_from_slice(b"a");
    let mut b = Vec::new();
    b.extend_from_slice(&1u32.to_be_bytes());
    b.extend_from_slice(b"b");

    let mut patch = Vec::new();
    patch.extend_from_slice(&MAGIC);
    patch.extend_from_slice(&make_chunk(b"ADIR", &a));
    patch.extend_from_slice(&make_chunk(b"ADIR", &b));
    patch.extend_from_slice(&make_chunk(b"EOF_", &[]));

    let tmp = tempfile::tempdir().unwrap();
    let ctx = ApplyConfig::new(tmp.path()).with_observer(LoggingObserver { log: log_clone });
    let reader = ZiPatchReader::new(Cursor::new(patch)).unwrap();
    ctx.apply_patch(reader).unwrap();

    let events = log.lock().unwrap();
    assert_eq!(
        events.len(),
        2,
        "two non-EOF chunks must fire exactly two events"
    );
    // Index must be 0-based and monotonically increasing.
    assert_eq!(events[0].index, 0, "first event index must be 0");
    assert_eq!(events[1].index, 1, "second event index must be 1");
    // Tag must reflect the chunk wire tag.
    assert_eq!(events[0].kind, *b"ADIR");
    assert_eq!(events[1].kind, *b"ADIR");
    // ADIR body for name "a": 4 (name_len) + 1 (byte) = 5
    // Frame: 4(size) + 4(tag) + 5(body) + 4(crc) = 17
    assert_eq!(
        events[0].bytes_read,
        12 + 17,
        "bytes_read after first ADIR must be magic + one 17-byte frame"
    );
    assert_eq!(
        events[1].bytes_read,
        12 + 17 + 17,
        "bytes_read after second ADIR must be magic + two 17-byte frames"
    );
    // Strict monotonicity.
    assert!(
        events[0].bytes_read < events[1].bytes_read,
        "bytes_read must strictly increase between events"
    );
}

#[test]
fn observer_break_on_first_chunk_aborts_immediately_leaving_first_applied() {
    // Observer that always breaks: only the first chunk's apply runs, then
    // apply_to returns Cancelled. Second and third chunks are never reached.
    let mut a = Vec::new();
    a.extend_from_slice(&1u32.to_be_bytes());
    a.extend_from_slice(b"a");
    let mut b_body = Vec::new();
    b_body.extend_from_slice(&1u32.to_be_bytes());
    b_body.extend_from_slice(b"b");
    let mut c = Vec::new();
    c.extend_from_slice(&1u32.to_be_bytes());
    c.extend_from_slice(b"c");

    let mut patch = Vec::new();
    patch.extend_from_slice(&MAGIC);
    patch.extend_from_slice(&make_chunk(b"ADIR", &a));
    patch.extend_from_slice(&make_chunk(b"ADIR", &b_body));
    patch.extend_from_slice(&make_chunk(b"ADIR", &c));
    patch.extend_from_slice(&make_chunk(b"EOF_", &[]));

    let count = Arc::new(AtomicUsize::new(0));
    let count_clone = count.clone();

    let tmp = tempfile::tempdir().unwrap();
    let ctx = ApplyConfig::new(tmp.path()).with_observer(CountingBreaker { count: count_clone });
    let reader = ZiPatchReader::new(Cursor::new(patch)).unwrap();
    let err = ctx.apply_patch(reader).unwrap_err();

    assert!(
        matches!(err, ApplyError::Cancelled),
        "observer Break must produce ApplyError::Cancelled, got {err:?}"
    );
    assert_eq!(
        count.load(Ordering::Relaxed),
        1,
        "exactly one on_chunk_applied call fires before the abort takes effect"
    );
    // The first ADIR's apply completed before the event fired.
    assert!(
        tmp.path().join("a").is_dir(),
        "first ADIR must have been applied before Cancelled was returned"
    );
    // Second and third ADIRs were never reached.
    assert!(
        !tmp.path().join("b").exists(),
        "second ADIR must NOT have been applied after Cancelled"
    );
    assert!(
        !tmp.path().join("c").exists(),
        "third ADIR must NOT have been applied after Cancelled"
    );
}

#[test]
fn observer_break_on_last_chunk_before_eof_leaves_all_earlier_applied() {
    // Three ADIRs: observer continues for the first two, breaks on the third.
    // After Cancelled, a/ and b/ must exist; c/ was the breaker's chunk
    // (its apply ran before the event fired) and d/ (hypothetical fourth) never runs.
    let make_adir_chunk = |name: &[u8]| -> Vec<u8> {
        let mut body = Vec::new();
        body.extend_from_slice(&(name.len() as u32).to_be_bytes());
        body.extend_from_slice(name);
        make_chunk(b"ADIR", &body)
    };

    let mut patch = Vec::new();
    patch.extend_from_slice(&MAGIC);
    patch.extend_from_slice(&make_adir_chunk(b"a"));
    patch.extend_from_slice(&make_adir_chunk(b"b"));
    patch.extend_from_slice(&make_adir_chunk(b"c"));
    patch.extend_from_slice(&make_chunk(b"EOF_", &[]));

    let call_count = Arc::new(AtomicUsize::new(0));
    let cc = call_count.clone();
    let tmp = tempfile::tempdir().unwrap();
    let ctx = ApplyConfig::new(tmp.path()).with_observer(BreakAfter {
        count: cc,
        threshold: 3,
    });

    let reader = ZiPatchReader::new(Cursor::new(patch)).unwrap();
    let err = ctx.apply_patch(reader).unwrap_err();

    assert!(
        matches!(err, ApplyError::Cancelled),
        "expected Cancelled, got {err:?}"
    );
    // First two ADIRs fully applied.
    assert!(tmp.path().join("a").is_dir(), "a/ must exist");
    assert!(tmp.path().join("b").is_dir(), "b/ must exist");
    // Third ADIR's apply ran before the event — c/ exists.
    assert!(
        tmp.path().join("c").is_dir(),
        "c/ must exist (apply ran before event fired)"
    );
}