git-remote-object-store 0.2.4

//! In-memory [`ObjectStore`] used by every higher-layer test.
//!
//! Keeps push, fetch, locking, doctor, and LFS logic exercisable without a
//! MinIO/Azurite container. Production builds do not see this module — it
//! is gated on `cfg(any(test, feature = "test-util"))`. Crate-internal unit
//! tests pick it up via `cfg(test)`; integration tests in `tests/` opt in
//! by enabling the `test-util` Cargo feature.
//!
//! The backing store is a `BTreeMap<String, MockObject>` so
//! [`ObjectStore::list`]'s iteration is deterministic (lexicographic).
//! Prefix matching mirrors S3 `Prefix=` byte-prefix semantics —
//! `list("a")` returns `a`, `a/1`, and `aaa`.
//!
//! Fault injection is FIFO: tests call [`MockStore::arm`] to queue a
//! [`Fault`]; the next matching operation consumes it and returns the
//! corresponding [`ObjectStoreError`]. This drives stale-lock recovery
//! and similar error-path tests deterministically.

use std::collections::BTreeMap;
use std::ops::Bound;
use std::path::Path;
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use bytes::Bytes;
use time::OffsetDateTime;

use super::error::other_boxed;
use super::{GetOpts, ObjectMeta, ObjectStore, ObjectStoreError, ProgressSink, PutOpts};

/// Produce a deterministic, content-derived `ETag` string that mimics the
/// `"<hex>"` format returned by S3 `HeadObject`. Uses a simple FNV-1a
/// hash — sufficient for test-time identity checks without pulling in a
/// crypto dependency.
fn mock_etag(body: &[u8]) -> String {
    let h = body.iter().fold(0xcbf2_9ce4_8422_2325_u64, |h, &b| {
        (h ^ u64::from(b)).wrapping_mul(0x0100_0000_01b3)
    });
    format!("\"{h:016x}\"")
}

/// Single-call fault recipes consumed FIFO by [`MockStore`].
///
/// Each variant matches one trait method + key/prefix; when the matching
/// call fires, the variant is removed from the queue and the listed error
/// is returned.
#[derive(Debug, Clone)]
pub enum Fault {
    /// Force `put_if_absent(key, _)` to return [`ObjectStoreError::PreconditionFailed`]
    /// without writing.
    PreconditionFailedOnPutIfAbsent {
        /// Key being written.
        key: String,
    },
    /// Force `head(key)` to return [`ObjectStoreError::NotFound`].
    NotFoundOnHead {
        /// Key being inspected.
        key: String,
    },
    /// Force `head(key)` to return [`ObjectStoreError::Network`],
    /// simulating a transient remote failure (timeout, 5xx) on the
    /// HEAD round-trip. Distinct from [`Fault::NotFoundOnHead`]: a
    /// `NotFound` is a definitive answer ("object is absent"), whereas
    /// a `Network` error tells the caller nothing about presence and
    /// must be treated as inconclusive.
    NetworkOnHead {
        /// Key being inspected.
        key: String,
    },
    /// Force `get_bytes(key)` to return [`ObjectStoreError::Network`].
    NetworkOnGetBytes {
        /// Key being read.
        key: String,
    },
    /// Force `list(prefix)` to return [`ObjectStoreError::AccessDenied`].
    AccessDeniedOnList {
        /// Prefix being listed.
        prefix: String,
    },
    /// Force every `list(_)` call to return [`ObjectStoreError::AccessDenied`],
    /// regardless of prefix. Use this when the test wants to assert that a
    /// given code path makes ZERO list calls — armed-and-still-pending
    /// after the call proves no `list` happened, even on an unanticipated
    /// prefix that an exact-match [`Fault::AccessDeniedOnList`] would miss.
    AccessDeniedOnAnyList,
    /// Force `delete(key)` to return [`ObjectStoreError::Network`].
    NetworkOnDelete {
        /// Key being deleted.
        key: String,
    },
    /// Force `delete(key)` to return [`ObjectStoreError::NotFound`],
    /// simulating a concurrent sweeper that removed the key after we
    /// listed it but before we deleted it.
    NotFoundOnDelete {
        /// Key being deleted.
        key: String,
    },
    /// Force `put_path(key, _, _)` to return [`ObjectStoreError::Network`]
    /// without writing. Fires before the source file is read, so the
    /// fault is independent of disk state. Scoped to `put_path` (not
    /// `put_bytes`) so tests that drive multiple uploads through the
    /// same store can target one path-based upload at a time without
    /// perturbing other writes.
    NetworkOnPutPath {
        /// Key being uploaded.
        key: String,
    },
    /// Force `put_bytes(key, _, _)` to return
    /// [`ObjectStoreError::Network`] for any key starting with
    /// `prefix`. Used by tests that target a write whose final key
    /// embeds a non-deterministic component (e.g. a UUID embedded in
    /// the path) and so can't be matched exactly. Fires on
    /// `put_bytes` only; `put_path` is untouched.
    NetworkOnPutBytesPrefix {
        /// Prefix the key must start with for the fault to fire.
        prefix: String,
    },
    /// Force `get_to_file(key, _)` to return
    /// [`ObjectStoreError::PreconditionFailed`], simulating an object that was
    /// mutated between the `head` and the body download in the S3
    /// backend. Higher-layer protocol tests use this to verify that
    /// fetch surfaces a structured error rather than a corrupted bundle.
    PreconditionFailedOnGetToFile {
        /// Key being downloaded.
        key: String,
    },
    /// Force `put_if_absent(key, _)` to return `Ok(false)` even when the
    /// key is currently absent, simulating a concurrent write winning the
    /// CAS race between our absence-check and our own write. S3 surfaces
    /// this as 409 `ConditionalRequestConflict`; this fault lets tests
    /// exercise callers' contention-handling without re-inserting the key
    /// between two awaits.
    ContendedPutIfAbsent {
        /// Key being written.
        key: String,
    },
}

#[derive(Debug, Clone)]
struct MockObject {
    body: Bytes,
    last_modified: OffsetDateTime,
    content_disposition: Option<String>,
    user_metadata: Vec<(String, String)>,
}

#[derive(Default)]
struct MockState {
    objects: BTreeMap<String, MockObject>,
    faults: Vec<Fault>,
    /// Chunk size used to slice progress callbacks for `put_path` /
    /// `get_to_file` / `put_bytes`. `None` means "single end-of-transfer
    /// event with the full body size", matching the on-the-wire
    /// behaviour of S3 single-PUT and Azure single-shot upload paths.
    /// Tests that exercise per-chunk progress (LFS agent) set a small
    /// chunk size so a multi-byte body produces multiple events.
    progress_chunk_size: Option<u64>,
    /// Closure that supplies a deterministic URL for
    /// [`MockStore::presigned_get_url`] when armed by
    /// [`MockStore::set_presign_stub`]. Receives the bucket-relative
    /// `key` and requested `ttl`. `None` (default) keeps the trait's
    /// default `Unsupported` behaviour so tests that don't opt in
    /// see the same error a `MockStore` would otherwise return.
    presign_stub: Option<Arc<PresignStubFn>>,
}

/// Function shape for [`MockStore::set_presign_stub`] — borrowed key
/// and copy-`Duration` so the stub can render any URL it likes.
pub type PresignStubFn =
    dyn Fn(&str, std::time::Duration) -> Result<String, ObjectStoreError> + Send + Sync;

/// In-memory [`ObjectStore`] for tests.
///
/// `Clone` is cheap — the backing `Arc<Mutex<…>>` is shared, so all clones
/// observe the same state. Instances are `Send + Sync`.
#[derive(Default, Clone)]
pub struct MockStore {
    inner: Arc<Mutex<MockState>>,
}

impl MockStore {
    /// Build an empty store with no faults armed.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Queue `fault` to fire on the next matching operation.
    pub fn arm(&self, fault: Fault) {
        self.with_state(|s| s.faults.push(fault));
    }

    /// Configure the chunk size used when slicing progress events on
    /// `put_path` / `get_to_file` / `put_bytes`. `None` (the default)
    /// produces a single end-of-transfer event with the full body size.
    /// `Some(n)` produces ⌈len / n⌉ events, each of `min(n, remaining)`
    /// bytes — used by LFS agent tests to assert that progress flows
    /// at chunk granularity.
    pub fn set_progress_chunk_size(&self, chunk_size: Option<u64>) {
        self.with_state(|s| s.progress_chunk_size = chunk_size);
    }

    /// Arm a closure that returns the URL [`presigned_get_url`] will
    /// emit. Used by `bundle-uri` dispatch tests to verify the
    /// presigning code path without a real `SigV4` / SAS backend.
    /// `None` clears the stub and reverts to the default
    /// `ObjectStoreError::Unsupported` error.
    ///
    /// [`presigned_get_url`]: ObjectStore::presigned_get_url
    pub fn set_presign_stub<F>(&self, stub: Option<F>)
    where
        F: Fn(&str, std::time::Duration) -> Result<String, ObjectStoreError>
            + Send
            + Sync
            + 'static,
    {
        self.with_state(|s| s.presign_stub = stub.map(|f| Arc::new(f) as Arc<PresignStubFn>));
    }

    /// Seed the store with `body` under `key`, stamping `last_modified` to
    /// "now". Existing entries at `key` are overwritten.
    pub fn insert(&self, key: impl Into<String>, body: impl Into<Bytes>) {
        self.insert_with(key, body, OffsetDateTime::now_utc(), PutOpts::default());
    }

    /// Seed with explicit `last_modified` and metadata. Stale-lock recovery
    /// tests use this to back-date a lock object so the staleness check
    /// fires.
    pub fn insert_with(
        &self,
        key: impl Into<String>,
        body: impl Into<Bytes>,
        last_modified: OffsetDateTime,
        opts: PutOpts,
    ) {
        let key = key.into();
        let object = MockObject {
            body: body.into(),
            last_modified,
            content_disposition: opts.content_disposition,
            user_metadata: opts.user_metadata,
        };
        self.with_state(|s| {
            s.objects.insert(key, object);
        });
    }

    /// Snapshot of every key currently stored, sorted lex.
    #[must_use]
    pub fn keys(&self) -> Vec<String> {
        self.with_state(|s| s.objects.keys().cloned().collect())
    }

    /// `true` if `key` is present.
    #[must_use]
    pub fn contains(&self, key: &str) -> bool {
        self.with_state(|s| s.objects.contains_key(key))
    }

    /// Remove `key` synchronously. Returns `true` if the key existed,
    /// `false` otherwise. Test-only mutator used to simulate a
    /// concurrent operation racing the system under test from a
    /// synchronous context (e.g., inside a [`Prompter::confirm`]
    /// callback) where the async `delete` is unavailable.
    ///
    /// [`Prompter::confirm`]: crate::manage::Prompter::confirm
    #[must_use]
    pub fn remove_key(&self, key: &str) -> bool {
        self.with_state(|s| s.objects.remove(key).is_some())
    }

    /// Number of armed faults that have not yet fired. Tests assert this
    /// is `0` to catch typos in fault keys.
    #[must_use]
    pub fn pending_faults(&self) -> usize {
        self.with_state(|s| s.faults.len())
    }

    /// Read back the [`PutOpts`] previously stored under `key`. `None` if
    /// the key is absent. Used by tests that round-trip metadata — the
    /// trait does not surface metadata on `head` (yet).
    #[must_use]
    pub fn metadata(&self, key: &str) -> Option<PutOpts> {
        self.with_state(|s| {
            s.objects.get(key).map(|o| PutOpts {
                content_disposition: o.content_disposition.clone(),
                user_metadata: o.user_metadata.clone(),
                progress: None,
            })
        })
    }

    fn with_state<R>(&self, f: impl FnOnce(&mut MockState) -> R) -> R {
        let mut guard = self.inner.lock().expect("mock mutex poisoned");
        f(&mut guard)
    }

    /// Drive `sink` with `body_len` bytes' worth of `report` calls,
    /// honouring the configured `progress_chunk_size`. Empty bodies
    /// emit no events (matches the real backends — a 0-byte download
    /// short-circuits the GET entirely).
    fn emit_progress(&self, sink: &ProgressSink, body_len: u64) {
        if body_len == 0 {
            return;
        }
        let chunk_size = self.with_state(|s| s.progress_chunk_size);
        let Some(chunk_size) = chunk_size else {
            sink.report(body_len);
            return;
        };
        if chunk_size == 0 {
            sink.report(body_len);
            return;
        }
        let mut remaining = body_len;
        while remaining > 0 {
            let step = remaining.min(chunk_size);
            sink.report(step);
            remaining -= step;
        }
    }

    /// Pop the first fault for which `map` returns `Some(err)` and bubble
    /// that error out. The closure runs twice on the matching fault — once
    /// to locate it, once to extract the error — but the queue is tiny in
    /// tests, so the duplicated match is cheaper than threading the
    /// destructured payload through the callsite.
    fn check_fault(
        state: &mut MockState,
        map: impl Fn(&Fault) -> Option<ObjectStoreError>,
    ) -> Result<(), ObjectStoreError> {
        let Some(position) = state.faults.iter().position(|f| map(f).is_some()) else {
            return Ok(());
        };
        Err(map(&state.faults.remove(position)).expect("position guarantees match"))
    }
}

/// Lexicographic successor of `prefix`, used as the exclusive upper bound
/// for byte-prefix range queries on a `BTreeMap<String, _>`.
///
/// Returns `Bound::Unbounded` for the empty prefix and for prefixes that
/// are entirely `0xFF` bytes (no successor exists in the byte-string
/// order). For any other prefix, returns `Bound::Excluded(succ)`, where
/// `succ` is `prefix` with the last non-`0xFF` byte incremented and any
/// trailing `0xFF` bytes truncated.
fn next_lex(prefix: &str) -> Bound<String> {
    let bytes = prefix.as_bytes();
    let Some(pivot) = bytes.iter().rposition(|&b| b != 0xFF) else {
        return Bound::Unbounded;
    };
    let mut next = bytes[..=pivot].to_vec();
    next[pivot] += 1;
    // The increment may produce an invalid UTF-8 byte; fall back to an
    // unbounded upper if so. In the wire-format-invariant key space
    // (`<prefix>/<ref>/...`) this never fires because every legal prefix
    // ends with a printable ASCII byte.
    String::from_utf8(next).map_or(Bound::Unbounded, Bound::Excluded)
}

#[async_trait]
impl ObjectStore for MockStore {
    async fn list(&self, prefix: &str) -> Result<Vec<ObjectMeta>, ObjectStoreError> {
        self.with_state(|s| {
            Self::check_fault(s, |f| match f {
                Fault::AccessDeniedOnList { prefix: p } if p == prefix => {
                    Some(ObjectStoreError::AccessDenied(p.clone()))
                }
                Fault::AccessDeniedOnAnyList => {
                    Some(ObjectStoreError::AccessDenied(prefix.to_string()))
                }
                _ => None,
            })?;
            let bounds = (Bound::Included(prefix.to_string()), next_lex(prefix));
            Ok(s.objects
                .range(bounds)
                .map(|(key, object)| ObjectMeta {
                    key: key.clone(),
                    size: object.body.len() as u64,
                    last_modified: object.last_modified,
                    // S3 ListObjectsV2 returns ETags, but S3Store sets
                    // `etag: None` (not consumed by any caller). Match
                    // that to avoid mock/prod fidelity drift.
                    etag: None,
                })
                .collect())
        })
    }

    async fn get_to_file(
        &self,
        key: &str,
        dest: &Path,
        opts: GetOpts,
    ) -> Result<(), ObjectStoreError> {
        self.with_state(|s| {
            Self::check_fault(s, |f| match f {
                Fault::PreconditionFailedOnGetToFile { key: k } if k == key => {
                    Some(ObjectStoreError::PreconditionFailed(k.clone()))
                }
                _ => None,
            })
        })?;
        let bytes = self.get_bytes(key).await?;
        let body_len = bytes.len() as u64;
        tokio::fs::write(dest, &bytes).await.map_err(other_boxed)?;
        if let Some(sink) = opts.progress.as_ref() {
            self.emit_progress(sink, body_len);
        }
        Ok(())
    }

    async fn get_bytes(&self, key: &str) -> Result<Bytes, ObjectStoreError> {
        self.with_state(|s| {
            Self::check_fault(s, |f| match f {
                Fault::NetworkOnGetBytes { key: k } if k == key => Some(ObjectStoreError::Network(
                    Box::new(std::io::Error::other(format!("mock network: {k}"))),
                )),
                _ => None,
            })?;
            s.objects
                .get(key)
                .map(|o| o.body.clone())
                .ok_or_else(|| ObjectStoreError::NotFound(key.to_string()))
        })
    }

    /// Slice the in-memory body. Mirrors the trait contract:
    /// `start == end` short-circuits to `Ok(Bytes::new())` without
    /// touching the store; `start > end` and `end > body.len()` both
    /// produce [`ObjectStoreError::RangeNotSatisfiable`]. Rejecting
    /// `end > body.len()` matches what the real S3 and Azure backends
    /// surface after their post-flight length check elevates the
    /// silently-truncated body to the same error.
    async fn get_bytes_range(
        &self,
        key: &str,
        range: std::ops::Range<u64>,
    ) -> Result<Bytes, ObjectStoreError> {
        if let Some(empty) = super::precheck_range(key, &range)? {
            return Ok(empty);
        }
        self.with_state(|s| {
            let object = s
                .objects
                .get(key)
                .ok_or_else(|| ObjectStoreError::NotFound(key.to_string()))?;
            let len = object.body.len() as u64;
            if range.end > len {
                return Err(ObjectStoreError::RangeNotSatisfiable {
                    key: key.to_string(),
                    requested: range,
                });
            }
            // Both bounds fit in `usize` because `range.end <= len` and
            // `len` came from `Bytes::len() -> usize`. The
            // `try_from` is infallible on every target where len fits
            // in u64 — a 32-bit port would surface here, not silently
            // truncate.
            let start =
                usize::try_from(range.start).expect("range.start <= len; len fits in usize");
            let end = usize::try_from(range.end).expect("range.end <= len; len fits in usize");
            Ok(object.body.slice(start..end))
        })
    }

    async fn put_bytes(
        &self,
        key: &str,
        body: Bytes,
        opts: PutOpts,
    ) -> Result<(), ObjectStoreError> {
        self.with_state(|s| {
            Self::check_fault(s, |f| match f {
                Fault::NetworkOnPutBytesPrefix { prefix } if key.starts_with(prefix.as_str()) => {
                    Some(ObjectStoreError::Network(Box::new(std::io::Error::other(
                        format!("mock network on put_bytes: {key} (prefix {prefix})"),
                    ))))
                }
                _ => None,
            })
        })?;
        let body_len = body.len() as u64;
        let progress = opts.progress.clone();
        self.insert_with(key, body, OffsetDateTime::now_utc(), opts);
        if let Some(sink) = progress.as_ref() {
            self.emit_progress(sink, body_len);
        }
        Ok(())
    }

    /// Override the default trait `put_path` so the configured
    /// `progress_chunk_size` drives multi-event progress callbacks.
    /// Without this override the default streams the full body to
    /// `put_bytes` without progress and then emits a single
    /// end-of-transfer event, defeating the chunk knob.
    async fn put_path(&self, key: &str, src: &Path, opts: PutOpts) -> Result<(), ObjectStoreError> {
        self.with_state(|s| {
            Self::check_fault(s, |f| match f {
                Fault::NetworkOnPutPath { key: k } if k == key => {
                    Some(ObjectStoreError::Network(Box::new(std::io::Error::other(
                        format!("mock network on put_path: {k}"),
                    ))))
                }
                _ => None,
            })
        })?;
        let body = tokio::fs::read(src).await.map_err(other_boxed)?;
        self.put_bytes(key, Bytes::from(body), opts).await
    }

    async fn put_if_absent(&self, key: &str, body: Bytes) -> Result<bool, ObjectStoreError> {
        self.with_state(|s| {
            Self::check_fault(s, |f| match f {
                Fault::PreconditionFailedOnPutIfAbsent { key: k } if k == key => {
                    Some(ObjectStoreError::PreconditionFailed(k.clone()))
                }
                _ => None,
            })?;
            if s.objects.contains_key(key) {
                return Ok(false);
            }
            // `ContendedPutIfAbsent` fires only when the key is absent above
            // — this models the CAS race where another client writes between
            // our absence-check and our own conditional PUT.
            if let Some(pos) = s
                .faults
                .iter()
                .position(|f| matches!(f, Fault::ContendedPutIfAbsent { key: k } if k == key))
            {
                s.faults.remove(pos);
                return Ok(false);
            }
            s.objects.insert(
                key.to_string(),
                MockObject {
                    body,
                    last_modified: OffsetDateTime::now_utc(),
                    content_disposition: None,
                    user_metadata: Vec::new(),
                },
            );
            Ok(true)
        })
    }

    async fn head(&self, key: &str) -> Result<ObjectMeta, ObjectStoreError> {
        self.with_state(|s| {
            Self::check_fault(s, |f| match f {
                Fault::NotFoundOnHead { key: k } if k == key => {
                    Some(ObjectStoreError::NotFound(k.clone()))
                }
                Fault::NetworkOnHead { key: k } if k == key => Some(ObjectStoreError::Network(
                    Box::new(std::io::Error::other(format!("mock network on head: {k}"))),
                )),
                _ => None,
            })?;
            s.objects
                .get(key)
                .map(|o| ObjectMeta {
                    key: key.to_string(),
                    size: o.body.len() as u64,
                    last_modified: o.last_modified,
                    etag: Some(mock_etag(&o.body)),
                })
                .ok_or_else(|| ObjectStoreError::NotFound(key.to_string()))
        })
    }

    async fn copy(&self, src: &str, dst: &str) -> Result<(), ObjectStoreError> {
        self.with_state(|s| {
            let mut copied = s
                .objects
                .get(src)
                .cloned()
                .ok_or_else(|| ObjectStoreError::NotFound(src.to_string()))?;
            // Copy gets a fresh server-side timestamp, matching S3's
            // copy_object semantics.
            copied.last_modified = OffsetDateTime::now_utc();
            s.objects.insert(dst.to_string(), copied);
            Ok(())
        })
    }

    async fn delete(&self, key: &str) -> Result<(), ObjectStoreError> {
        self.with_state(|s| {
            Self::check_fault(s, |f| match f {
                Fault::NetworkOnDelete { key: k } if k == key => {
                    Some(ObjectStoreError::Network(Box::new(std::io::Error::other(
                        format!("mock network on delete: {k}"),
                    ))))
                }
                Fault::NotFoundOnDelete { key: k } if k == key => {
                    Some(ObjectStoreError::NotFound(k.clone()))
                }
                _ => None,
            })?;
            s.objects
                .remove(key)
                .map(|_| ())
                .ok_or_else(|| ObjectStoreError::NotFound(key.to_string()))
        })
    }

    /// Override for [`ObjectStore::presigned_get_url`]. Defers to the
    /// stub closure when [`MockStore::set_presign_stub`] has armed
    /// one; otherwise falls back to the trait's default `Unsupported`
    /// error. Lets `bundle-uri` dispatch tests assert that the
    /// presigning code path is taken without standing up a real
    /// `SigV4` / SAS implementation.
    async fn presigned_get_url(
        &self,
        key: &str,
        ttl: std::time::Duration,
    ) -> Result<String, ObjectStoreError> {
        if let Some(stub) = self.with_state(|s| s.presign_stub.clone()) {
            return stub(key, ttl);
        }
        Err(ObjectStoreError::Unsupported(
            "presigned URLs are not supported by this backend".to_owned(),
        ))
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;

    fn body(data: &[u8]) -> Bytes {
        Bytes::copy_from_slice(data)
    }

    #[tokio::test]
    async fn put_then_get_round_trips_bytes_and_size() {
        let store = MockStore::new();
        store
            .put_bytes("k", body(b"hello"), PutOpts::default())
            .await
            .unwrap();

        let got = store.get_bytes("k").await.unwrap();
        assert_eq!(&got[..], b"hello");

        let meta = store.head("k").await.unwrap();
        assert_eq!(meta.key, "k");
        assert_eq!(meta.size, 5);
    }

    #[tokio::test]
    async fn list_uses_byte_prefix_semantics() {
        let store = MockStore::new();
        for key in ["a", "a/1", "a/2", "aaa", "b/1"] {
            store.insert(key, body(b""));
        }

        let under_a = store.list("a").await.unwrap();
        let keys: Vec<&str> = under_a.iter().map(|m| m.key.as_str()).collect();
        assert_eq!(keys, vec!["a", "a/1", "a/2", "aaa"]);

        let under_a_slash = store.list("a/").await.unwrap();
        let keys: Vec<&str> = under_a_slash.iter().map(|m| m.key.as_str()).collect();
        assert_eq!(keys, vec!["a/1", "a/2"]);
    }

    #[tokio::test]
    async fn list_empty_prefix_returns_everything() {
        let store = MockStore::new();
        store.insert("a", body(b""));
        store.insert("z", body(b""));
        let all = store.list("").await.unwrap();
        let keys: Vec<&str> = all.iter().map(|m| m.key.as_str()).collect();
        assert_eq!(keys, vec!["a", "z"]);
    }

    #[tokio::test]
    async fn put_bytes_overwrites_existing_key() {
        let store = MockStore::new();
        store
            .put_bytes("k", body(b"first"), PutOpts::default())
            .await
            .unwrap();
        store
            .put_bytes("k", body(b"second-longer"), PutOpts::default())
            .await
            .unwrap();
        assert_eq!(&store.get_bytes("k").await.unwrap()[..], b"second-longer");
        let meta = store.head("k").await.unwrap();
        assert_eq!(meta.size, b"second-longer".len() as u64);
    }

    #[tokio::test]
    async fn put_if_absent_fault_fires_before_existing_key_check() {
        // Both a fault is armed AND the key is already present. The
        // implementation must consult the fault queue before the
        // contains_key short-circuit, so callers see Err(PreconditionFailed)
        // rather than Ok(false). Locks in the ordering enforced by
        // `MockStore::put_if_absent`.
        let store = MockStore::new();
        store.insert("k", body(b"existing"));
        store.arm(Fault::PreconditionFailedOnPutIfAbsent { key: "k".into() });

        let err = store.put_if_absent("k", body(b"x")).await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::PreconditionFailed(ref k) if k == "k"));
        // Body unchanged.
        assert_eq!(&store.get_bytes("k").await.unwrap()[..], b"existing");
        assert_eq!(store.pending_faults(), 0);
    }

    #[tokio::test]
    async fn put_if_absent_supports_zero_byte_lock_objects() {
        let store = MockStore::new();
        let acquired = store.put_if_absent("LOCK", Bytes::new()).await.unwrap();
        assert!(acquired);
        let meta = store.head("LOCK").await.unwrap();
        assert_eq!(meta.size, 0);
    }

    #[tokio::test]
    async fn put_if_absent_returns_false_when_key_exists() {
        let store = MockStore::new();
        assert!(store.put_if_absent("k", body(b"first")).await.unwrap());
        assert!(!store.put_if_absent("k", body(b"second")).await.unwrap());
        // Body unchanged after the rejected second call.
        assert_eq!(&store.get_bytes("k").await.unwrap()[..], b"first");
    }

    #[tokio::test]
    async fn put_if_absent_fault_returns_precondition_and_consumes_once() {
        let store = MockStore::new();
        store.arm(Fault::PreconditionFailedOnPutIfAbsent { key: "k".into() });

        let err = store.put_if_absent("k", body(b"x")).await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::PreconditionFailed(ref k) if k == "k"));
        assert!(!store.contains("k"));
        assert_eq!(store.pending_faults(), 0);

        // Subsequent call without a fault succeeds and inserts.
        assert!(store.put_if_absent("k", body(b"x")).await.unwrap());
    }

    #[tokio::test]
    async fn contended_put_if_absent_fault_returns_false_and_consumes_once() {
        // Sibling tripwire to `put_if_absent_fault_returns_precondition_...`.
        // Models the S3 409 ConditionalRequestConflict response: a CAS
        // race where the key is currently absent but a concurrent writer
        // wins between our absence-check and our own write. The fault must:
        //   1. fire on `put_if_absent(absent_key, _)` and return Ok(false)
        //   2. be consumed (`pending_faults()` drops to 0)
        //   3. NOT insert the key on the contended path
        //   4. allow a subsequent fault-free call to succeed normally
        let store = MockStore::new();
        store.arm(Fault::ContendedPutIfAbsent { key: "k".into() });

        // Key is absent; fault armed → Ok(false), no insert.
        let inserted = store.put_if_absent("k", body(b"x")).await.unwrap();
        assert!(!inserted, "contended fault must return Ok(false)");
        assert!(
            !store.contains("k"),
            "contended fault must NOT insert the key",
        );
        assert_eq!(
            store.pending_faults(),
            0,
            "contended fault must be consumed once",
        );

        // Without a fault, the same call now succeeds and inserts.
        let inserted = store.put_if_absent("k", body(b"x")).await.unwrap();
        assert!(inserted, "next put_if_absent must succeed and insert");
        assert!(store.contains("k"));
    }

    #[tokio::test]
    async fn delete_missing_key_is_not_found() {
        let store = MockStore::new();
        let err = store.delete("missing").await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::NotFound(ref k) if k == "missing"));
    }

    #[tokio::test]
    async fn copy_replicates_body_and_metadata_with_fresh_timestamp() {
        let store = MockStore::new();
        let src_time = OffsetDateTime::now_utc() - Duration::from_mins(1);
        store.insert_with(
            "src",
            body(b"payload"),
            src_time,
            PutOpts {
                content_disposition: Some("attachment; filename=foo".into()),
                user_metadata: vec![("k".into(), "v".into())],
                progress: None,
            },
        );

        store.copy("src", "dst").await.unwrap();
        assert_eq!(&store.get_bytes("dst").await.unwrap()[..], b"payload");
        // Source untouched.
        assert!(store.contains("src"));
        let opts = store.metadata("dst").expect("dst exists");
        assert_eq!(
            opts.content_disposition.as_deref(),
            Some("attachment; filename=foo")
        );
        assert_eq!(opts.user_metadata, vec![("k".to_string(), "v".to_string())]);
        // S3's copy_object semantics: dst gets a fresh server-side
        // timestamp, not the back-dated source's.
        let dst_meta = store.head("dst").await.unwrap();
        assert!(
            dst_meta.last_modified > src_time,
            "expected fresh timestamp on dst, got {} ≤ src {src_time}",
            dst_meta.last_modified,
        );
    }

    #[tokio::test]
    async fn copy_missing_source_is_not_found() {
        let store = MockStore::new();
        let err = store.copy("nope", "dst").await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::NotFound(ref k) if k == "nope"));
        assert!(!store.contains("dst"));
    }

    #[tokio::test]
    async fn copy_overwrites_existing_destination() {
        let store = MockStore::new();
        store.insert("src", body(b"new"));
        store.insert("dst", body(b"old"));
        store.copy("src", "dst").await.unwrap();
        assert_eq!(&store.get_bytes("dst").await.unwrap()[..], b"new");
    }

    #[tokio::test]
    async fn get_to_file_writes_body_to_path() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("out.bin");
        let store = MockStore::new();
        store.insert("k", body(b"file-bytes"));

        store
            .get_to_file("k", &path, GetOpts::default())
            .await
            .unwrap();
        let read = tokio::fs::read(&path).await.unwrap();
        assert_eq!(read, b"file-bytes");
    }

    #[tokio::test]
    async fn get_to_file_missing_parent_dir_yields_other() {
        let dir = tempfile::tempdir().unwrap();
        // Path under a subdir we deliberately do not create — guarantees
        // ENOENT from the host without coupling to any absolute path.
        let path = dir.path().join("missing-subdir").join("out.bin");
        let store = MockStore::new();
        store.insert("k", body(b"x"));
        let err = store
            .get_to_file("k", &path, GetOpts::default())
            .await
            .unwrap_err();
        assert!(matches!(err, ObjectStoreError::Other(_)));
    }

    #[tokio::test]
    async fn put_opts_round_trip_through_metadata_accessor() {
        let store = MockStore::new();
        let opts = PutOpts {
            content_disposition: Some("inline".into()),
            user_metadata: vec![("a".into(), "1".into()), ("b".into(), "2".into())],
            progress: None,
        };
        store.put_bytes("k", body(b""), opts.clone()).await.unwrap();
        let stored = store.metadata("k").expect("k exists");
        assert_eq!(stored.content_disposition, opts.content_disposition);
        assert_eq!(stored.user_metadata, opts.user_metadata);
    }

    #[tokio::test]
    async fn insert_with_back_dates_last_modified() {
        let store = MockStore::new();
        let then = OffsetDateTime::now_utc() - Duration::from_mins(5);
        store.insert_with("LOCK", body(b""), then, PutOpts::default());
        let meta = store.head("LOCK").await.unwrap();
        assert_eq!(meta.last_modified, then);
    }

    #[tokio::test]
    async fn faults_that_never_fire_are_observable() {
        let store = MockStore::new();
        store.arm(Fault::NotFoundOnHead {
            key: "never".into(),
        });
        // Operate on a different key — fault should remain queued.
        store.insert("other", body(b""));
        store.head("other").await.unwrap();
        assert_eq!(store.pending_faults(), 1);
    }

    #[tokio::test]
    async fn list_access_denied_fault_fires_once() {
        let store = MockStore::new();
        store.arm(Fault::AccessDeniedOnList {
            prefix: "secret/".into(),
        });
        let err = store.list("secret/").await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::AccessDenied(ref p) if p == "secret/"));
        // Second call without a queued fault returns empty.
        assert!(store.list("secret/").await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn head_not_found_fault_fires_once() {
        let store = MockStore::new();
        store.insert("k", body(b"abc"));
        store.arm(Fault::NotFoundOnHead { key: "k".into() });
        let err = store.head("k").await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::NotFound(ref k) if k == "k"));
        // Without a queued fault, head returns the inserted object's
        // metadata (key + size). Inspecting the payload guards against
        // regressions that swap the returned key or size.
        let meta = store.head("k").await.unwrap();
        assert_eq!(meta.key, "k");
        assert_eq!(meta.size, 3);
    }

    #[tokio::test]
    async fn get_bytes_range_returns_slice() {
        let store = MockStore::new();
        store.insert("k", body(b"abcdefghij")); // 10 bytes
        let got = store.get_bytes_range("k", 2..6).await.unwrap();
        assert_eq!(&got[..], b"cdef");
    }

    #[tokio::test]
    async fn get_bytes_range_empty_range_returns_empty_without_lookup() {
        let store = MockStore::new();
        // Key absent — empty range must not surface NotFound; it must
        // short-circuit before consulting the store.
        let got = store.get_bytes_range("missing", 5..5).await.unwrap();
        assert!(got.is_empty());
    }

    #[tokio::test]
    async fn get_bytes_range_inverted_returns_range_not_satisfiable() {
        let store = MockStore::new();
        store.insert("k", body(b"abcdefghij"));
        // Construct the inverted range via the struct literal so
        // clippy's `reversed_empty_ranges` lint (which only fires on
        // `5..2` written inline) does not block the test. The trait
        // contract expects this exact shape.
        let inverted = std::ops::Range {
            start: 5_u64,
            end: 2_u64,
        };
        let err = store.get_bytes_range("k", inverted).await.unwrap_err();
        assert!(
            matches!(
                err,
                ObjectStoreError::RangeNotSatisfiable {
                    ref key,
                    requested: ref r,
                } if key == "k" && r.start == 5 && r.end == 2,
            ),
            "expected RangeNotSatisfiable(key=k, 5..2), got {err:?}",
        );
    }

    #[tokio::test]
    async fn get_bytes_range_past_end_returns_range_not_satisfiable() {
        let store = MockStore::new();
        store.insert("k", body(b"abc"));
        let err = store.get_bytes_range("k", 0..10).await.unwrap_err();
        assert!(
            matches!(
                err,
                ObjectStoreError::RangeNotSatisfiable {
                    ref key,
                    requested: ref r,
                } if key == "k" && r.start == 0 && r.end == 10,
            ),
            "expected RangeNotSatisfiable(key=k, 0..10), got {err:?}",
        );
    }

    #[tokio::test]
    async fn get_bytes_range_missing_key_is_not_found() {
        let store = MockStore::new();
        let err = store.get_bytes_range("missing", 0..1).await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::NotFound(ref k) if k == "missing"));
    }

    #[tokio::test]
    async fn get_bytes_range_full_object_round_trips() {
        let store = MockStore::new();
        store.insert("k", body(b"hello"));
        let got = store.get_bytes_range("k", 0..5).await.unwrap();
        assert_eq!(&got[..], b"hello");
    }

    #[tokio::test]
    async fn get_bytes_network_fault_fires_once() {
        let store = MockStore::new();
        store.insert("k", body(b"x"));
        store.arm(Fault::NetworkOnGetBytes { key: "k".into() });
        let err = store.get_bytes("k").await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::Network(_)));
        assert_eq!(&store.get_bytes("k").await.unwrap()[..], b"x");
    }

    #[test]
    fn next_lex_covers_empty_short_and_invalid_utf8_fallback() {
        // Empty input: rposition finds no non-0xFF byte, so the function
        // returns Unbounded.
        assert!(matches!(next_lex(""), Bound::Unbounded));
        // Short ASCII inputs increment the last byte cleanly.
        assert!(matches!(next_lex("a"), Bound::Excluded(s) if s == "b"));
        assert!(matches!(next_lex("ab"), Bound::Excluded(s) if s == "ac"));
        // Unicode max code point U+10FFFF encodes as F4 8F BF BF.
        // Incrementing the trailing 0xBF yields F4 8F BF C0, which is
        // invalid UTF-8 (lone 0xC0 continuation), so the
        // String::from_utf8 fallback path returns Unbounded. This is the
        // only realistic way to exercise that branch through the &str
        // surface.
        assert!(matches!(next_lex("\u{10FFFF}"), Bound::Unbounded));
    }

    #[tokio::test]
    async fn delete_network_fault_fires_once() {
        let store = MockStore::new();
        store.insert("k", body(b"x"));
        store.arm(Fault::NetworkOnDelete { key: "k".into() });
        let err = store.delete("k").await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::Network(_)));
        // Object was not removed because the fault fired first.
        assert!(store.contains("k"));
        // Second call without a fault succeeds.
        store.delete("k").await.unwrap();
        assert!(!store.contains("k"));
    }

    #[tokio::test]
    async fn get_to_file_precondition_fault_fires_once() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("out.bin");
        let store = MockStore::new();
        store.insert("k", body(b"payload"));
        store.arm(Fault::PreconditionFailedOnGetToFile { key: "k".into() });
        let err = store
            .get_to_file("k", &path, GetOpts::default())
            .await
            .unwrap_err();
        assert!(matches!(err, ObjectStoreError::PreconditionFailed(ref k) if k == "k"));
        assert!(!path.exists(), "file must not be written on fault");
        assert_eq!(store.pending_faults(), 0);
        // Second call without a fault succeeds.
        store
            .get_to_file("k", &path, GetOpts::default())
            .await
            .unwrap();
        let read = tokio::fs::read(&path).await.unwrap();
        assert_eq!(read, b"payload");
    }

    #[tokio::test]
    async fn head_returns_deterministic_etag() {
        let store = MockStore::new();
        store.insert("k", body(b"hello"));
        let m1 = store.head("k").await.unwrap();
        let m2 = store.head("k").await.unwrap();
        assert!(m1.etag.is_some(), "etag must be present");
        assert_eq!(m1.etag, m2.etag, "same content ⇒ same etag");
        // Different content ⇒ different etag.
        store.insert("k", body(b"world"));
        let m3 = store.head("k").await.unwrap();
        assert_ne!(m1.etag, m3.etag, "different content ⇒ different etag");
    }

    #[test]
    fn mock_etag_is_deterministic() {
        let a = mock_etag(b"abc");
        let b = mock_etag(b"abc");
        assert_eq!(a, b);
        // Quoted format mimicking S3.
        assert!(a.starts_with('"') && a.ends_with('"'));
        // Different input ⇒ different output.
        assert_ne!(mock_etag(b"abc"), mock_etag(b"xyz"));
    }

    #[tokio::test]
    async fn put_path_round_trips_same_as_put_bytes() {
        let dir = tempfile::tempdir().unwrap();
        let file_path = dir.path().join("payload.bin");
        let content = b"put_path round-trip content";
        tokio::fs::write(&file_path, content).await.unwrap();

        let store = MockStore::new();
        let opts = PutOpts {
            content_disposition: Some("attachment".into()),
            user_metadata: vec![("key".into(), "val".into())],
            progress: None,
        };
        store
            .put_path("via-path", &file_path, opts.clone())
            .await
            .unwrap();
        store
            .put_bytes("via-bytes", body(content), opts)
            .await
            .unwrap();

        let path_body = store.get_bytes("via-path").await.unwrap();
        let bytes_body = store.get_bytes("via-bytes").await.unwrap();
        assert_eq!(
            path_body, bytes_body,
            "put_path and put_bytes must produce identical bodies"
        );

        let path_meta = store.metadata("via-path").expect("via-path exists");
        let bytes_meta = store.metadata("via-bytes").expect("via-bytes exists");
        assert_eq!(
            path_meta.content_disposition, bytes_meta.content_disposition,
            "metadata must match"
        );
        assert_eq!(
            path_meta.user_metadata, bytes_meta.user_metadata,
            "user_metadata must match"
        );
    }

    #[tokio::test]
    async fn put_path_missing_file_returns_error() {
        let store = MockStore::new();
        let err = store
            .put_path(
                "k",
                Path::new("/tmp/nonexistent-abc123xyz"),
                PutOpts::default(),
            )
            .await
            .unwrap_err();
        assert!(
            matches!(err, ObjectStoreError::Other(_)),
            "expected ObjectStoreError::Other, got {err:?}"
        );
    }

    #[test]
    fn mock_store_is_send_and_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<MockStore>();
    }

    #[tokio::test]
    async fn put_path_emits_chunked_progress_with_configured_chunk_size() {
        // Pin the chunk-knob-driven progress contract: with chunk=8 on a
        // 20-byte body, we get 3 events of sizes 8, 8, 4 totalling 20.
        // Higher layers (the LFS agent) depend on this for live progress
        // — see lfs/agent.rs::upload_emits_chunked_progress_for_multipart_body.
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("body.bin");
        let payload = b"abcdefghijklmnopqrst"; // 20 bytes
        tokio::fs::write(&src, payload).await.unwrap();

        let store = MockStore::new();
        store.set_progress_chunk_size(Some(8));

        let received = Arc::new(Mutex::new(Vec::<u64>::new()));
        let received_clone = Arc::clone(&received);
        let sink = ProgressSink::new(move |amount| {
            received_clone.lock().expect("lock").push(amount);
        });
        let opts = PutOpts {
            progress: Some(sink),
            ..PutOpts::default()
        };
        store.put_path("k", &src, opts).await.unwrap();

        let chunks = received.lock().unwrap().clone();
        assert_eq!(chunks, vec![8, 8, 4]);
        assert_eq!(chunks.iter().sum::<u64>(), payload.len() as u64);
    }

    #[tokio::test]
    async fn put_path_default_chunk_emits_single_event_with_full_size() {
        // With no chunk knob configured, MockStore mirrors S3 single-PUT
        // / Azure single-shot upload: one report() call with the full
        // body size. Locks in the "no chunking, but at least one event"
        // contract so the LFS agent's bytesSoFar==size assertion still
        // holds for backends without per-chunk hooks.
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("body.bin");
        tokio::fs::write(&src, b"hello").await.unwrap();

        let store = MockStore::new();
        let received = Arc::new(Mutex::new(Vec::<u64>::new()));
        let received_clone = Arc::clone(&received);
        let sink = ProgressSink::new(move |amount| {
            received_clone.lock().expect("lock").push(amount);
        });
        let opts = PutOpts {
            progress: Some(sink),
            ..PutOpts::default()
        };
        store.put_path("k", &src, opts).await.unwrap();

        assert_eq!(received.lock().unwrap().clone(), vec![5]);
    }

    #[tokio::test]
    async fn empty_body_emits_no_progress_events() {
        // A 0-byte transfer fires no callback. Real backends short-
        // circuit empty downloads (skip the GET entirely) so no chunk
        // ever lands on the wire — match that here so the LFS agent
        // doesn't emit a `"bytesSinceLast":0` stub.
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("empty.bin");
        tokio::fs::write(&src, b"").await.unwrap();
        let store = MockStore::new();
        store.set_progress_chunk_size(Some(8));

        let received = Arc::new(Mutex::new(Vec::<u64>::new()));
        let received_clone = Arc::clone(&received);
        let sink = ProgressSink::new(move |amount| {
            received_clone.lock().expect("lock").push(amount);
        });
        let opts = PutOpts {
            progress: Some(sink),
            ..PutOpts::default()
        };
        store.put_path("k", &src, opts).await.unwrap();
        assert!(received.lock().unwrap().is_empty());
    }
}