mx-cache 0.1.0

//! Comprehensive test suite for mx-cache crate.
//!
//! This module contains:
//! - Extended `LocalCache` tests
//! - Mock-based `RedisCache` tests
//! - Property-based tests using proptest
//! - Concurrent access pattern tests
//! - Edge case and stress tests

use crate::{Cache, LocalCache};
use std::time::Duration;

// ============================================================================
// LocalCache Extended Tests
// ============================================================================

mod local_cache_tests {
    use super::*;
    use futures::future::join_all;

    fn create_test_cache() -> LocalCache {
        LocalCache::new(1000, Duration::from_secs(60))
    }

    // ==================== Edge Cases ====================

    #[tokio::test]
    async fn test_empty_key() {
        let cache = create_test_cache();
        let key: &[u8] = b"";
        let value = b"value_for_empty_key";

        cache
            .set(key, value, Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(key).await.unwrap();

        assert_eq!(result, Some(value.to_vec()));
    }

    #[tokio::test]
    async fn test_empty_value() {
        let cache = create_test_cache();
        let key = b"key_with_empty_value";
        let value: &[u8] = b"";

        cache
            .set(key, value, Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(key).await.unwrap();

        assert_eq!(result, Some(Vec::new()));
    }

    #[tokio::test]
    async fn test_large_value() {
        let cache = create_test_cache();
        let key = b"large_value_key";
        let value: Vec<u8> = (0..10_000).map(|i| (i % 256) as u8).collect();

        cache
            .set(key, &value, Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(key).await.unwrap();

        assert_eq!(result, Some(value));
    }

    #[tokio::test]
    async fn test_binary_key_and_value() {
        let cache = create_test_cache();
        // Binary data with null bytes and special characters
        let key: &[u8] = &[0x00, 0xFF, 0x01, 0xFE, 0x00];
        let value: &[u8] = &[0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x00];

        cache
            .set(key, value, Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(key).await.unwrap();

        assert_eq!(result, Some(value.to_vec()));
    }

    #[tokio::test]
    async fn test_unicode_bytes_in_key() {
        let cache = create_test_cache();
        // UTF-8 encoded unicode string
        let key = b"";
        let value = b"value";

        cache
            .set(key, value, Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(key).await.unwrap();

        assert_eq!(result, Some(value.to_vec()));
    }

    #[tokio::test]
    async fn test_set_overwrites_existing() {
        let cache = create_test_cache();
        let key = b"overwrite_key";

        cache
            .set(key, b"value1", Duration::from_secs(60))
            .await
            .unwrap();
        cache
            .set(key, b"value2", Duration::from_secs(60))
            .await
            .unwrap();

        let result = cache.get(key).await.unwrap();
        assert_eq!(result, Some(b"value2".to_vec()));
    }

    #[tokio::test]
    async fn test_del_nonexistent_key() {
        let cache = create_test_cache();
        // Should not error when deleting non-existent key
        let result = cache.del(b"nonexistent").await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_set_nx_px_after_delete() {
        let cache = create_test_cache();
        let key = b"deleted_nx_key";

        cache
            .set_nx_px(key, b"value1", Duration::from_secs(60))
            .await
            .unwrap();
        cache.del(key).await.unwrap();

        // Should succeed after deletion
        let was_set = cache
            .set_nx_px(key, b"value2", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(was_set, "should succeed after deletion");

        let result = cache.get(key).await.unwrap();
        assert_eq!(result, Some(b"value2".to_vec()));
    }

    #[tokio::test]
    async fn test_contains_sync_after_delete() {
        let cache = create_test_cache();
        let key = b"contains_delete";

        cache
            .set(key, b"value", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(cache.contains_sync(key));

        cache.del(key).await.unwrap();
        assert!(!cache.contains_sync(key));
    }

    // ==================== TTL Behavior ====================

    #[tokio::test]
    async fn test_zero_ttl_expires_immediately() {
        let cache = create_test_cache();
        let key = b"zero_ttl_key";

        cache.set(key, b"value", Duration::ZERO).await.unwrap();

        // Value should expire very quickly (or immediately)
        tokio::time::sleep(Duration::from_millis(10)).await;
        let result = cache.get(key).await.unwrap();
        assert!(result.is_none(), "zero TTL should expire immediately");
    }

    #[tokio::test]
    async fn test_very_long_ttl() {
        let cache = create_test_cache();
        let key = b"long_ttl_key";

        // Set with 1 hour TTL
        cache
            .set(key, b"value", Duration::from_secs(3600))
            .await
            .unwrap();

        // Should still exist
        let result = cache.get(key).await.unwrap();
        assert!(result.is_some());
    }

    // ==================== Concurrent Access ====================

    #[tokio::test]
    async fn test_concurrent_sets() {
        let cache = create_test_cache();
        let num_tasks = 100;

        let handles: Vec<_> = (0..num_tasks)
            .map(|i| {
                let cache = cache.clone();
                tokio::spawn(async move {
                    let key = format!("concurrent_key_{i}");
                    let value = format!("value_{i}");
                    cache
                        .set(key.as_bytes(), value.as_bytes(), Duration::from_secs(60))
                        .await
                        .unwrap();
                })
            })
            .collect();

        join_all(handles).await;

        // Verify all values are set
        for i in 0..num_tasks {
            let key = format!("concurrent_key_{i}");
            let expected = format!("value_{i}");
            let result = cache.get(key.as_bytes()).await.unwrap();
            assert_eq!(result, Some(expected.into_bytes()));
        }
    }

    #[tokio::test]
    async fn test_concurrent_set_nx_px_same_key() {
        let cache = create_test_cache();
        let key = b"race_key";
        let num_tasks = 100;

        let handles: Vec<_> = (0..num_tasks)
            .map(|i| {
                let cache = cache.clone();
                tokio::spawn(async move {
                    let value = format!("value_{i}");
                    cache
                        .set_nx_px(key, value.as_bytes(), Duration::from_secs(60))
                        .await
                        .unwrap()
                })
            })
            .collect();

        let results: Vec<bool> = join_all(handles)
            .await
            .into_iter()
            .map(|r| r.unwrap())
            .collect();

        // Exactly one should succeed (be true)
        let success_count = results.iter().filter(|&&x| x).count();
        assert_eq!(success_count, 1, "exactly one set_nx_px should succeed");
    }

    #[tokio::test]
    async fn test_concurrent_reads_and_writes() {
        let cache = create_test_cache();
        let key = b"rw_key";

        // Pre-populate
        cache
            .set(key, b"initial", Duration::from_secs(60))
            .await
            .unwrap();

        let mut handles = Vec::new();

        // Spawn readers
        for _ in 0..50 {
            let cache = cache.clone();
            handles.push(tokio::spawn(async move {
                let result = cache.get(key).await.unwrap();
                assert!(result.is_some());
            }));
        }

        // Spawn writers
        for i in 0..50 {
            let cache = cache.clone();
            handles.push(tokio::spawn(async move {
                let value = format!("value_{i}");
                cache
                    .set(key, value.as_bytes(), Duration::from_secs(60))
                    .await
                    .unwrap();
            }));
        }

        // All should complete without error
        for handle in join_all(handles).await {
            handle.unwrap();
        }
    }

    // ==================== Capacity ====================

    #[tokio::test]
    async fn test_capacity_limit() {
        // Create cache with small capacity
        // Note: Moka's eviction is asynchronous and may not happen immediately.
        // We use a slightly larger capacity and insert many more items to ensure
        // eviction eventually occurs.
        let cache = LocalCache::new(50, Duration::from_secs(60));

        // Insert significantly more items than capacity
        for i in 0..200 {
            let key = format!("cap_key_{i}");
            let value = format!("value_{i}");
            cache
                .set(key.as_bytes(), value.as_bytes(), Duration::from_secs(60))
                .await
                .unwrap();
        }

        // Give Moka time to run eviction - it processes evictions asynchronously
        tokio::time::sleep(Duration::from_millis(500)).await;

        // Count how many entries remain
        let mut count = 0;
        for i in 0..200 {
            let key = format!("cap_key_{i}");
            if cache.get(key.as_bytes()).await.unwrap().is_some() {
                count += 1;
            }
        }

        // Moka uses weighted eviction, so we just verify that some eviction occurred
        // We inserted 200 items into a cache with capacity 50, so we expect significant eviction
        assert!(
            count < 200,
            "cache should evict some entries when significantly over capacity, found {count} entries"
        );
    }

    // ==================== Clone Behavior ====================

    #[tokio::test]
    async fn test_clone_shares_state() {
        let cache1 = create_test_cache();
        let cache2 = cache1.clone();

        cache1
            .set(b"shared_key", b"value", Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache2.get(b"shared_key").await.unwrap();

        assert_eq!(result, Some(b"value".to_vec()), "clones should share state");
    }

    #[tokio::test]
    async fn test_clone_delete_propagates() {
        let cache1 = create_test_cache();
        let cache2 = cache1.clone();

        cache1
            .set(b"shared_key", b"value", Duration::from_secs(60))
            .await
            .unwrap();
        cache2.del(b"shared_key").await.unwrap();

        let result = cache1.get(b"shared_key").await.unwrap();
        assert_eq!(result, None, "delete should propagate to clones");
    }

    // ==================== Stress Tests ====================

    #[tokio::test]
    async fn test_many_keys() {
        let cache = LocalCache::new(10_000, Duration::from_secs(60));
        let num_keys = 5000;

        // Insert many keys
        for i in 0..num_keys {
            let key = format!("key_{i:05}");
            let value = format!("value_{i:05}");
            cache
                .set(key.as_bytes(), value.as_bytes(), Duration::from_secs(60))
                .await
                .unwrap();
        }

        // Verify all keys exist
        for i in 0..num_keys {
            let key = format!("key_{i:05}");
            let expected = format!("value_{i:05}");
            let result = cache.get(key.as_bytes()).await.unwrap();
            assert_eq!(result, Some(expected.into_bytes()), "key {i} should exist");
        }
    }

    #[tokio::test]
    async fn test_rapid_set_get_cycles() {
        let cache = create_test_cache();
        let key = b"rapid_key";

        for i in 0..1000 {
            let value = format!("value_{i}");
            cache
                .set(key, value.as_bytes(), Duration::from_secs(60))
                .await
                .unwrap();
            let result = cache.get(key).await.unwrap();
            assert_eq!(result, Some(value.into_bytes()));
        }
    }
}

// ============================================================================
// RedisCache Namespace Tests (unit tests without real Redis)
// ============================================================================

mod redis_cache_namespace_tests {
    use std::borrow::Cow;

    /// Test the namespacing logic without actual Redis connection.
    /// We replicate the private `namespaced` method logic here for testing.
    fn namespaced<'a>(prefix: &str, key: &[u8], stack: &'a mut [u8; 96]) -> Cow<'a, [u8]> {
        let needed = prefix.len() + key.len();
        if needed <= stack.len() {
            let mut offset = 0;
            stack[..prefix.len()].copy_from_slice(prefix.as_bytes());
            offset += prefix.len();
            stack[offset..offset + key.len()].copy_from_slice(key);
            return Cow::Borrowed(&stack[..needed]);
        }
        let mut buf = Vec::with_capacity(needed);
        buf.extend_from_slice(prefix.as_bytes());
        buf.extend_from_slice(key);
        Cow::Owned(buf)
    }

    #[test]
    fn test_namespace_short_key() {
        let prefix = "myapp:";
        let key = b"mykey";
        let mut stack = [0u8; 96];

        let result = namespaced(prefix, key, &mut stack);

        assert!(
            matches!(result, Cow::Borrowed(_)),
            "should use stack buffer"
        );
        assert_eq!(result.as_ref(), b"myapp:mykey");
    }

    #[test]
    fn test_namespace_empty_prefix() {
        let prefix = "";
        let key = b"mykey";
        let mut stack = [0u8; 96];

        let result = namespaced(prefix, key, &mut stack);

        assert_eq!(result.as_ref(), b"mykey");
    }

    #[test]
    fn test_namespace_empty_key() {
        let prefix = "myapp:";
        let key = b"";
        let mut stack = [0u8; 96];

        let result = namespaced(prefix, key, &mut stack);

        assert_eq!(result.as_ref(), b"myapp:");
    }

    #[test]
    fn test_namespace_long_key_uses_heap() {
        let prefix = "myapp:";
        // Create key longer than 96 bytes
        let key: Vec<u8> = (0..100).map(|i| (i % 256) as u8).collect();
        let mut stack = [0u8; 96];

        let result = namespaced(prefix, &key, &mut stack);

        assert!(
            matches!(result, Cow::Owned(_)),
            "should use heap for long keys"
        );
        assert_eq!(result.len(), prefix.len() + key.len());
        assert!(result.starts_with(prefix.as_bytes()));
        assert!(result.ends_with(&key));
    }

    #[test]
    fn test_namespace_exactly_96_bytes() {
        let prefix = "prefix:"; // 7 bytes
        let key = vec![b'x'; 89]; // 89 bytes = total 96
        let mut stack = [0u8; 96];

        let result = namespaced(prefix, &key, &mut stack);

        assert!(
            matches!(result, Cow::Borrowed(_)),
            "exactly 96 bytes should use stack"
        );
        assert_eq!(result.len(), 96);
    }

    #[test]
    fn test_namespace_97_bytes_uses_heap() {
        let prefix = "prefix:"; // 7 bytes
        let key = vec![b'x'; 90]; // 90 bytes = total 97
        let mut stack = [0u8; 96];

        let result = namespaced(prefix, &key, &mut stack);

        assert!(matches!(result, Cow::Owned(_)), "97 bytes should use heap");
        assert_eq!(result.len(), 97);
    }

    #[test]
    fn test_namespace_binary_key() {
        let prefix = "bin:";
        let key: &[u8] = &[0x00, 0xFF, 0xDE, 0xAD, 0xBE, 0xEF];
        let mut stack = [0u8; 96];

        let result = namespaced(prefix, key, &mut stack);

        assert_eq!(&result[..4], b"bin:");
        assert_eq!(&result[4..], key);
    }
}

// ============================================================================
// Property-Based Tests
// ============================================================================

mod proptest_tests {
    use super::*;
    use proptest::prelude::*;

    proptest! {
        /// Test that set followed by get always returns the same value
        #[test]
        fn prop_set_get_roundtrip(
            key in prop::collection::vec(any::<u8>(), 0..100),
            value in prop::collection::vec(any::<u8>(), 0..1000)
        ) {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let cache = LocalCache::new(1000, Duration::from_secs(60));

                cache.set(&key, &value, Duration::from_secs(60)).await.unwrap();
                let result = cache.get(&key).await.unwrap();

                prop_assert_eq!(result, Some(value));
                Ok(())
            })?;
        }

        /// Test that set_nx_px returns false on second call with same key
        #[test]
        fn prop_set_nx_px_idempotent(
            key in prop::collection::vec(any::<u8>(), 1..50),
            value1 in prop::collection::vec(any::<u8>(), 0..100),
            value2 in prop::collection::vec(any::<u8>(), 0..100)
        ) {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let cache = LocalCache::new(1000, Duration::from_secs(60));

                let first = cache.set_nx_px(&key, &value1, Duration::from_secs(60)).await.unwrap();
                let second = cache.set_nx_px(&key, &value2, Duration::from_secs(60)).await.unwrap();

                prop_assert!(first, "first set_nx_px should succeed");
                prop_assert!(!second, "second set_nx_px should fail");

                // Value should be the first one
                let result = cache.get(&key).await.unwrap();
                prop_assert_eq!(result, Some(value1));
                Ok(())
            })?;
        }

        /// Test that delete makes key unavailable
        #[test]
        fn prop_delete_removes_key(
            key in prop::collection::vec(any::<u8>(), 1..50),
            value in prop::collection::vec(any::<u8>(), 0..100)
        ) {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let cache = LocalCache::new(1000, Duration::from_secs(60));

                cache.set(&key, &value, Duration::from_secs(60)).await.unwrap();
                cache.del(&key).await.unwrap();
                let result = cache.get(&key).await.unwrap();

                prop_assert_eq!(result, None);
                Ok(())
            })?;
        }

        /// Test that contains_sync is consistent with get
        #[test]
        fn prop_contains_sync_consistent_with_get(
            key in prop::collection::vec(any::<u8>(), 1..50),
            value in prop::collection::vec(any::<u8>(), 0..100),
            should_set in any::<bool>()
        ) {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let cache = LocalCache::new(1000, Duration::from_secs(60));

                if should_set {
                    cache.set(&key, &value, Duration::from_secs(60)).await.unwrap();
                }

                let contains = cache.contains_sync(&key);
                let get_result = cache.get(&key).await.unwrap();

                prop_assert_eq!(contains, get_result.is_some());
                Ok(())
            })?;
        }

        /// Test that set overwrites previous value
        #[test]
        fn prop_set_overwrites(
            key in prop::collection::vec(any::<u8>(), 1..50),
            value1 in prop::collection::vec(any::<u8>(), 0..100),
            value2 in prop::collection::vec(any::<u8>(), 0..100)
        ) {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let cache = LocalCache::new(1000, Duration::from_secs(60));

                cache.set(&key, &value1, Duration::from_secs(60)).await.unwrap();
                cache.set(&key, &value2, Duration::from_secs(60)).await.unwrap();
                let result = cache.get(&key).await.unwrap();

                prop_assert_eq!(result, Some(value2));
                Ok(())
            })?;
        }

        /// Test that get on non-existent key returns None
        #[test]
        fn prop_get_nonexistent_returns_none(
            key in prop::collection::vec(any::<u8>(), 1..50)
        ) {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let cache = LocalCache::new(1000, Duration::from_secs(60));
                let result = cache.get(&key).await.unwrap();
                prop_assert_eq!(result, None);
                Ok(())
            })?;
        }

        /// Test namespace concatenation (unit test for the logic)
        #[test]
        fn prop_namespace_concatenation(
            prefix in "[a-z]{0,10}:",
            key in prop::collection::vec(any::<u8>(), 0..50)
        ) {
            let mut stack = [0u8; 96];
            let needed = prefix.len() + key.len();

            // Replicate namespaced logic
            let result: std::borrow::Cow<'_, [u8]> = if needed <= stack.len() {
                stack[..prefix.len()].copy_from_slice(prefix.as_bytes());
                stack[prefix.len()..needed].copy_from_slice(&key);
                std::borrow::Cow::Borrowed(&stack[..needed])
            } else {
                let mut buf = Vec::with_capacity(needed);
                buf.extend_from_slice(prefix.as_bytes());
                buf.extend_from_slice(&key);
                std::borrow::Cow::Owned(buf)
            };

            // Verify result is correct concatenation
            prop_assert_eq!(result.len(), needed);
            prop_assert!(result.starts_with(prefix.as_bytes()));
            prop_assert!(result.ends_with(&key));
        }
    }
}

// ============================================================================
// Cache Trait Invariant Tests
// ============================================================================

mod cache_trait_invariants {
    use super::*;

    /// Test that all Cache trait implementations behave consistently
    async fn test_cache_invariants<C: Cache>(cache: &C, prefix: &str) {
        let key = format!("{prefix}invariant_key").into_bytes();
        let value = b"invariant_value";

        // Invariant 1: get on non-existent key returns None
        let result = cache.get(&key).await.unwrap();
        assert_eq!(result, None, "{prefix}: get on new key should be None");

        // Invariant 2: set followed by get returns value
        cache
            .set(&key, value, Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(&key).await.unwrap();
        assert_eq!(
            result,
            Some(value.to_vec()),
            "{prefix}: set then get should return value"
        );

        // Invariant 3: del removes value
        cache.del(&key).await.unwrap();
        let result = cache.get(&key).await.unwrap();
        assert_eq!(result, None, "{prefix}: del should remove value");

        // Invariant 4: set_nx_px on new key succeeds
        let key2 = format!("{prefix}nx_key").into_bytes();
        let was_set = cache
            .set_nx_px(&key2, value, Duration::from_secs(60))
            .await
            .unwrap();
        assert!(was_set, "{prefix}: set_nx_px on new key should succeed");

        // Invariant 5: set_nx_px on existing key fails
        let was_set_again = cache
            .set_nx_px(&key2, b"other", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(
            !was_set_again,
            "{prefix}: set_nx_px on existing key should fail"
        );

        // Invariant 6: value unchanged after failed set_nx_px
        let result = cache.get(&key2).await.unwrap();
        assert_eq!(
            result,
            Some(value.to_vec()),
            "{prefix}: value should be unchanged after failed set_nx_px"
        );
    }

    #[tokio::test]
    async fn test_local_cache_invariants() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));
        test_cache_invariants(&cache, "local:").await;
    }
}

// ============================================================================
// Deduplication Pattern Tests
// ============================================================================

// ============================================================================
// Additional Coverage Tests (10 new tests for 80% coverage)
// ============================================================================

mod additional_coverage_tests {
    use super::*;
    use futures::future::join_all;

    // ==================== Test 1: test_contains_sync_direct ====================

    /// Direct sync check returns correct value for existing and non-existing keys
    #[tokio::test]
    async fn test_contains_sync_direct() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));

        // Test on completely fresh cache - no key exists
        assert!(
            !cache.contains_sync(b"never_set_key"),
            "contains_sync should return false for never-set key"
        );

        // Set a key and verify contains_sync returns true
        cache
            .set(b"existing_key", b"value", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(
            cache.contains_sync(b"existing_key"),
            "contains_sync should return true for existing key"
        );

        // Verify it still returns false for other keys
        assert!(
            !cache.contains_sync(b"other_key"),
            "contains_sync should return false for different key"
        );

        // Test with empty key
        cache
            .set(b"", b"empty_key_value", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(
            cache.contains_sync(b""),
            "contains_sync should work with empty key"
        );
    }

    // ==================== Test 2: test_ttl_zero_immediate_expiry ====================

    /// TTL=0 expires immediately - entry should not be retrievable
    #[tokio::test]
    async fn test_ttl_zero_immediate_expiry() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));
        let key = b"zero_ttl_immediate";

        // Set with Duration::ZERO
        cache.set(key, b"value", Duration::ZERO).await.unwrap();

        // Give minimal time for Moka to process expiry
        tokio::time::sleep(Duration::from_millis(5)).await;

        // Value should be expired
        let result = cache.get(key).await.unwrap();
        assert!(result.is_none(), "TTL=0 should cause immediate expiration");

        // contains_sync should also return false
        assert!(
            !cache.contains_sync(key),
            "contains_sync should return false after zero TTL expiry"
        );
    }

    // ==================== Test 3: test_ttl_max_value ====================

    /// Very long TTL values work correctly without causing issues
    #[tokio::test]
    async fn test_ttl_max_value() {
        let cache = LocalCache::new(1000, Duration::from_secs(86400 * 365 * 10)); // 10 years default
        let key = b"max_ttl_key";

        // Set with 1 year TTL - a very long but reasonable value
        // This tests that we handle long-lived cache entries properly
        let one_year = Duration::from_secs(86400 * 365);
        cache.set(key, b"long_lived_value", one_year).await.unwrap();

        // Value should exist
        let result = cache.get(key).await.unwrap();
        assert_eq!(
            result,
            Some(b"long_lived_value".to_vec()),
            "large TTL should not cause issues"
        );

        // Also test set_nx_px with large TTL
        let key2 = b"max_ttl_nx_key";
        let was_set = cache.set_nx_px(key2, b"value", one_year).await.unwrap();
        assert!(was_set, "set_nx_px with large TTL should succeed");

        let result2 = cache.get(key2).await.unwrap();
        assert!(result2.is_some(), "value should be retrievable");

        // Test with maximum safe TTL (100 years - still within Instant bounds)
        let key3 = b"century_ttl_key";
        let hundred_years = Duration::from_secs(86400 * 365 * 100);
        cache
            .set(key3, b"century_value", hundred_years)
            .await
            .unwrap();

        let result3 = cache.get(key3).await.unwrap();
        assert_eq!(
            result3,
            Some(b"century_value".to_vec()),
            "100 year TTL should work"
        );
    }

    // ==================== Test 4: test_very_long_key ====================

    /// Key >1KB works correctly
    #[tokio::test]
    async fn test_very_long_key() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));

        // Create a key >1KB (1500 bytes)
        let long_key: Vec<u8> = (0..1500).map(|i| (i % 256) as u8).collect();
        let value = b"value_for_long_key";

        // Set and retrieve
        cache
            .set(&long_key, value, Duration::from_secs(60))
            .await
            .unwrap();

        let result = cache.get(&long_key).await.unwrap();
        assert_eq!(
            result,
            Some(value.to_vec()),
            "long key (>1KB) should work correctly"
        );

        // Test contains_sync with long key
        assert!(
            cache.contains_sync(&long_key),
            "contains_sync should work with long key"
        );

        // Test set_nx_px with long key
        let long_key2: Vec<u8> = (0..2000).map(|i| ((i + 100) % 256) as u8).collect();
        let was_set = cache
            .set_nx_px(&long_key2, b"nx_value", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(was_set, "set_nx_px should work with long key");

        // Delete long key
        cache.del(&long_key).await.unwrap();
        assert!(
            !cache.contains_sync(&long_key),
            "delete should work with long key"
        );
    }

    // ==================== Test 5: test_concurrent_ttl_expiry ====================

    /// Race condition handling with multiple keys expiring concurrently
    #[tokio::test]
    async fn test_concurrent_ttl_expiry() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));
        let num_keys = 100;
        let short_ttl = Duration::from_millis(50);

        // Set many keys with short TTL concurrently
        let handles: Vec<_> = (0..num_keys)
            .map(|i| {
                let cache = cache.clone();
                tokio::spawn(async move {
                    let key = format!("expiring_key_{i}");
                    let value = format!("value_{i}");
                    cache
                        .set(key.as_bytes(), value.as_bytes(), short_ttl)
                        .await
                        .unwrap();
                })
            })
            .collect();

        join_all(handles).await;

        // Verify all keys exist initially
        let mut initial_count = 0;
        for i in 0..num_keys {
            let key = format!("expiring_key_{i}");
            if cache.get(key.as_bytes()).await.unwrap().is_some() {
                initial_count += 1;
            }
        }
        assert!(initial_count > 0, "some keys should exist before expiry");

        // Wait for TTL expiration
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Concurrent reads after expiry
        let read_handles: Vec<_> = (0..num_keys)
            .map(|i| {
                let cache = cache.clone();
                tokio::spawn(async move {
                    let key = format!("expiring_key_{i}");
                    cache.get(key.as_bytes()).await.unwrap()
                })
            })
            .collect();

        let results: Vec<Option<Vec<u8>>> = join_all(read_handles)
            .await
            .into_iter()
            .map(|r| r.unwrap())
            .collect();

        // All should be expired
        let remaining = results.iter().filter(|r| r.is_some()).count();
        assert_eq!(
            remaining, 0,
            "all keys should be expired after TTL, found {remaining} remaining"
        );
    }

    // ==================== Test 6: test_cache_clone_drop ====================

    /// Behavior after clone drop - shared state persists
    #[tokio::test]
    async fn test_cache_clone_drop() {
        let cache1 = LocalCache::new(1000, Duration::from_secs(60));

        // Set value using cache1
        cache1
            .set(b"shared_key", b"shared_value", Duration::from_secs(60))
            .await
            .unwrap();

        // Create clone and drop it
        {
            let cache2 = cache1.clone();

            // Verify clone can read
            let result = cache2.get(b"shared_key").await.unwrap();
            assert_eq!(result, Some(b"shared_value".to_vec()));

            // Set another key via clone
            cache2
                .set(b"clone_key", b"clone_value", Duration::from_secs(60))
                .await
                .unwrap();

            // cache2 is dropped here
        }

        // Original cache should still work and see all data
        let result1 = cache1.get(b"shared_key").await.unwrap();
        assert_eq!(
            result1,
            Some(b"shared_value".to_vec()),
            "original key should persist after clone drop"
        );

        let result2 = cache1.get(b"clone_key").await.unwrap();
        assert_eq!(
            result2,
            Some(b"clone_value".to_vec()),
            "key set by dropped clone should persist"
        );

        // Create another clone and verify it sees all data
        let cache3 = cache1.clone();
        assert!(
            cache3.contains_sync(b"shared_key"),
            "new clone should see shared_key"
        );
        assert!(
            cache3.contains_sync(b"clone_key"),
            "new clone should see clone_key"
        );
    }

    // ==================== Test 7: test_set_nx_px_ttl_race ====================

    /// Concurrent `set_nx_px` with TTL - verify atomic behavior
    #[tokio::test]
    async fn test_set_nx_px_ttl_race() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));
        let key = b"race_nx_ttl_key";
        let num_workers = 50;
        let short_ttl = Duration::from_millis(30);

        // First wave: concurrent set_nx_px
        let handles1: Vec<_> = (0..num_workers)
            .map(|i| {
                let cache = cache.clone();
                tokio::spawn(async move {
                    let value = format!("value_{i}");
                    cache
                        .set_nx_px(key, value.as_bytes(), short_ttl)
                        .await
                        .unwrap()
                })
            })
            .collect();

        let results1: Vec<bool> = join_all(handles1)
            .await
            .into_iter()
            .map(|r| r.unwrap())
            .collect();

        // Exactly one winner
        let winners1 = results1.iter().filter(|&&x| x).count();
        assert_eq!(
            winners1, 1,
            "exactly one set_nx_px should succeed in first wave"
        );

        // Wait for TTL to expire
        tokio::time::sleep(Duration::from_millis(60)).await;

        // Second wave: should be able to set again
        let handles2: Vec<_> = (0..num_workers)
            .map(|i| {
                let cache = cache.clone();
                tokio::spawn(async move {
                    let value = format!("new_value_{i}");
                    cache
                        .set_nx_px(key, value.as_bytes(), short_ttl)
                        .await
                        .unwrap()
                })
            })
            .collect();

        let results2: Vec<bool> = join_all(handles2)
            .await
            .into_iter()
            .map(|r| r.unwrap())
            .collect();

        // Again exactly one winner
        let winners2 = results2.iter().filter(|&&x| x).count();
        assert_eq!(
            winners2, 1,
            "exactly one set_nx_px should succeed in second wave after TTL expiry"
        );
    }

    // ==================== Test 8: test_local_cache_moka_eviction ====================

    /// Capacity eviction when cache is full
    #[tokio::test]
    async fn test_local_cache_moka_eviction() {
        // Create very small capacity cache
        let cache = LocalCache::new(10, Duration::from_secs(60));

        // Insert many more items than capacity
        for i in 0..100 {
            let key = format!("evict_key_{i:03}");
            let value = format!("value_{i:03}");
            cache
                .set(key.as_bytes(), value.as_bytes(), Duration::from_secs(60))
                .await
                .unwrap();
        }

        // Give Moka time to process evictions asynchronously
        tokio::time::sleep(Duration::from_millis(200)).await;

        // Count remaining entries
        let mut count = 0;
        for i in 0..100 {
            let key = format!("evict_key_{i:03}");
            if cache.get(key.as_bytes()).await.unwrap().is_some() {
                count += 1;
            }
        }

        // With capacity 10 and 100 inserts, significant eviction should occur
        // Moka may keep slightly more due to async eviction, but should be well under 100
        assert!(
            count <= 50,
            "cache should evict entries when over capacity, found {count} entries (expected <= 50)"
        );

        // Most recent entries should more likely be present (LRU-like behavior)
        // Check that at least some recent entries exist
        let mut recent_count = 0;
        for i in 90..100 {
            let key = format!("evict_key_{i:03}");
            if cache.get(key.as_bytes()).await.unwrap().is_some() {
                recent_count += 1;
            }
        }
        // Recent entries should have higher likelihood of being present
        // (though Moka uses TinyLFU, not pure LRU, so this is a soft check)
        assert!(
            recent_count > 0 || count > 0,
            "cache should have some entries present"
        );
    }

    // ==================== Test 9: test_redis_namespace_special_chars ====================

    /// Special characters in namespace are handled correctly
    #[test]
    fn test_redis_namespace_special_chars() {
        use std::borrow::Cow;

        // Replicate the namespaced logic for testing
        fn namespaced<'a>(prefix: &str, key: &[u8], stack: &'a mut [u8; 96]) -> Cow<'a, [u8]> {
            let needed = prefix.len() + key.len();
            if needed <= stack.len() {
                let mut offset = 0;
                stack[..prefix.len()].copy_from_slice(prefix.as_bytes());
                offset += prefix.len();
                stack[offset..offset + key.len()].copy_from_slice(key);
                return Cow::Borrowed(&stack[..needed]);
            }
            let mut buf = Vec::with_capacity(needed);
            buf.extend_from_slice(prefix.as_bytes());
            buf.extend_from_slice(key);
            Cow::Owned(buf)
        }

        // Test with colons (common in Redis)
        let mut stack = [0u8; 96];
        let result = namespaced("app:env:cache:", b"user:123", &mut stack);
        assert_eq!(result.as_ref(), b"app:env:cache:user:123");

        // Test with special characters
        let mut stack = [0u8; 96];
        let result = namespaced("ns-with-dashes_and_underscores:", b"key", &mut stack);
        assert_eq!(result.as_ref(), b"ns-with-dashes_and_underscores:key");

        // Test with unicode in prefix
        let mut stack = [0u8; 96];
        let result = namespaced("cache:", b"key", &mut stack);
        assert_eq!(result.as_ref(), b"cache:key");

        // Test with dots and slashes
        let mut stack = [0u8; 96];
        let result = namespaced("v1.0/api:", b"endpoint/resource", &mut stack);
        assert_eq!(result.as_ref(), b"v1.0/api:endpoint/resource");

        // Test with curly braces (Redis cluster hash tags)
        let mut stack = [0u8; 96];
        let result = namespaced("{hashtag}:", b"key", &mut stack);
        assert_eq!(result.as_ref(), b"{hashtag}:key");

        // Test with spaces (unusual but valid)
        let mut stack = [0u8; 96];
        let result = namespaced("my app: ", b"my key", &mut stack);
        assert_eq!(result.as_ref(), b"my app: my key");
    }

    // ==================== Test 10: test_redis_namespace_binary ====================

    /// Binary key data handling in namespace concatenation
    #[test]
    fn test_redis_namespace_binary() {
        use std::borrow::Cow;

        // Replicate the namespaced logic for testing
        fn namespaced<'a>(prefix: &str, key: &[u8], stack: &'a mut [u8; 96]) -> Cow<'a, [u8]> {
            let needed = prefix.len() + key.len();
            if needed <= stack.len() {
                let mut offset = 0;
                stack[..prefix.len()].copy_from_slice(prefix.as_bytes());
                offset += prefix.len();
                stack[offset..offset + key.len()].copy_from_slice(key);
                return Cow::Borrowed(&stack[..needed]);
            }
            let mut buf = Vec::with_capacity(needed);
            buf.extend_from_slice(prefix.as_bytes());
            buf.extend_from_slice(key);
            Cow::Owned(buf)
        }

        // Test with null bytes in key
        let mut stack = [0u8; 96];
        let binary_key: &[u8] = &[0x00, 0x01, 0x02, 0x00, 0xFF];
        let result = namespaced("bin:", binary_key, &mut stack);
        assert_eq!(&result[..4], b"bin:");
        assert_eq!(&result[4..], binary_key);
        assert_eq!(result.len(), 4 + binary_key.len());

        // Test with high bytes (0x80-0xFF)
        let mut stack = [0u8; 96];
        let high_bytes: &[u8] = &[0x80, 0x90, 0xA0, 0xB0, 0xC0, 0xD0, 0xE0, 0xF0, 0xFF];
        let result = namespaced("high:", high_bytes, &mut stack);
        assert_eq!(&result[..5], b"high:");
        assert_eq!(&result[5..], high_bytes);

        // Test with all zero bytes
        let mut stack = [0u8; 96];
        let zeros: &[u8] = &[0x00, 0x00, 0x00, 0x00];
        let result = namespaced("zeros:", zeros, &mut stack);
        assert_eq!(&result[..6], b"zeros:");
        assert_eq!(&result[6..], zeros);

        // Test with mixed ASCII and binary
        let mut stack = [0u8; 96];
        let mixed: &[u8] = &[b'a', 0x00, b'b', 0xFF, b'c'];
        let result = namespaced("mix:", mixed, &mut stack);
        assert_eq!(&result[..4], b"mix:");
        assert_eq!(&result[4..], mixed);

        // Test binary key that exceeds stack buffer (forces heap allocation)
        let mut stack = [0u8; 96];
        let long_binary: Vec<u8> = (0..100).map(|i| i as u8).collect();
        let result = namespaced("long:", &long_binary, &mut stack);
        assert!(
            matches!(result, Cow::Owned(_)),
            "should use heap for long binary key"
        );
        assert_eq!(&result[..5], b"long:");
        assert_eq!(&result[5..], long_binary.as_slice());

        // Verify binary data is preserved exactly
        let mut stack = [0u8; 96];
        let exact_binary: &[u8] = &[0xDE, 0xAD, 0xBE, 0xEF, 0xCA, 0xFE, 0xBA, 0xBE];
        let result = namespaced("magic:", exact_binary, &mut stack);
        let expected: Vec<u8> = b"magic:"
            .iter()
            .chain(exact_binary.iter())
            .copied()
            .collect();
        assert_eq!(result.as_ref(), expected.as_slice());
    }
}

mod deduplication_tests {
    use super::*;
    use futures::future::join_all;

    /// Simulates the deduplication pattern used in mx-notifier
    #[tokio::test]
    async fn test_deduplication_pattern() {
        let cache = LocalCache::new(10_000, Duration::from_secs(300));

        // Simulate processing events with deduplication
        let event_ids: Vec<Vec<u8>> = (0..100)
            .map(|i| format!("event_{i}").into_bytes())
            .collect();

        let mut processed = Vec::new();

        for event_id in &event_ids {
            // Check if already processed using set_nx_px
            let is_new = cache
                .set_nx_px(event_id, b"1", Duration::from_secs(300))
                .await
                .unwrap();

            if is_new {
                processed.push(event_id.clone());
            }
        }

        // All events should be processed exactly once
        assert_eq!(processed.len(), 100);

        // Processing same events again should result in no new processing
        let mut reprocessed = Vec::new();
        for event_id in &event_ids {
            let is_new = cache
                .set_nx_px(event_id, b"1", Duration::from_secs(300))
                .await
                .unwrap();

            if is_new {
                reprocessed.push(event_id.clone());
            }
        }

        assert_eq!(reprocessed.len(), 0, "no events should be reprocessed");
    }

    /// Test concurrent deduplication (race condition handling)
    #[tokio::test]
    async fn test_concurrent_deduplication() {
        let cache = LocalCache::new(10_000, Duration::from_secs(300));

        // Same event ID processed by multiple concurrent tasks
        let event_id = b"duplicate_event";
        let num_workers = 50;

        let handles: Vec<_> = (0..num_workers)
            .map(|_| {
                let cache = cache.clone();
                tokio::spawn(async move {
                    cache
                        .set_nx_px(event_id, b"1", Duration::from_secs(300))
                        .await
                        .unwrap()
                })
            })
            .collect();

        let results: Vec<bool> = join_all(handles)
            .await
            .into_iter()
            .map(|r| r.unwrap())
            .collect();

        // Exactly one worker should successfully process
        let success_count = results.iter().filter(|&&x| x).count();
        assert_eq!(success_count, 1, "exactly one worker should win the race");
    }

    /// Test deduplication with TTL expiration
    #[tokio::test]
    async fn test_deduplication_with_expiration() {
        let cache = LocalCache::new(10_000, Duration::from_secs(60));
        let event_id = b"expiring_event";

        // Process with short TTL
        let first = cache
            .set_nx_px(event_id, b"1", Duration::from_millis(50))
            .await
            .unwrap();
        assert!(first, "first processing should succeed");

        // Wait for TTL to expire
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Should be able to process again after expiration
        let second = cache
            .set_nx_px(event_id, b"1", Duration::from_millis(50))
            .await
            .unwrap();
        assert!(second, "should be able to reprocess after expiration");
    }
}

// ============================================================================
// Mock Cache Tests - Testing Cache trait implementations
// ============================================================================

mod mock_cache_tests {
    use super::*;
    use crate::mock::MockCache;

    #[tokio::test]
    async fn test_mock_cache_trait_compliance() {
        let cache = MockCache::new();

        // Test all Cache trait methods
        cache
            .set(b"key1", b"value1", Duration::from_secs(60))
            .await
            .unwrap();

        let result = cache.get(b"key1").await.unwrap();
        assert_eq!(result, Some(b"value1".to_vec()));

        let was_set = cache
            .set_nx_px(b"key2", b"value2", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(was_set);

        let was_set = cache
            .set_nx_px(b"key2", b"other", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(!was_set);

        cache.del(b"key1").await.unwrap();
        let result = cache.get(b"key1").await.unwrap();
        assert_eq!(result, None);
    }

    #[tokio::test]
    async fn test_mock_cache_error_injection() {
        let cache = MockCache::new();

        // Set value before enabling error mode
        cache
            .set(b"key1", b"value1", Duration::from_secs(60))
            .await
            .unwrap();

        // Enable error mode
        cache.enable_error_mode("Simulated Redis failure");

        // All operations should fail
        let get_err = cache.get(b"key1").await;
        assert!(get_err.is_err());
        assert!(
            get_err
                .unwrap_err()
                .to_string()
                .contains("Simulated Redis failure")
        );

        let set_err = cache.set(b"key2", b"value2", Duration::from_secs(60)).await;
        assert!(set_err.is_err());

        let set_nx_err = cache
            .set_nx_px(b"key3", b"value3", Duration::from_secs(60))
            .await;
        assert!(set_nx_err.is_err());

        let del_err = cache.del(b"key1").await;
        assert!(del_err.is_err());

        // Disable error mode
        cache.disable_error_mode();

        // Operations should succeed again
        let result = cache.get(b"key1").await.unwrap();
        assert_eq!(result, Some(b"value1".to_vec()));
    }

    #[tokio::test]
    async fn test_mock_cache_operation_counting() {
        let cache = MockCache::new();

        cache
            .set(b"k1", b"v1", Duration::from_secs(60))
            .await
            .unwrap();
        cache
            .set(b"k2", b"v2", Duration::from_secs(60))
            .await
            .unwrap();
        cache.get(b"k1").await.unwrap();
        cache.get(b"k2").await.unwrap();
        cache.get(b"k3").await.unwrap();
        cache
            .set_nx_px(b"k4", b"v4", Duration::from_secs(60))
            .await
            .unwrap();
        cache
            .set_nx_px(b"k4", b"v5", Duration::from_secs(60))
            .await
            .unwrap();
        cache.del(b"k1").await.unwrap();

        let counts = cache.operation_counts();
        assert_eq!(counts.sets, 2, "should count 2 set operations");
        assert_eq!(counts.gets, 3, "should count 3 get operations");
        assert_eq!(counts.set_nx_px, 2, "should count 2 set_nx_px operations");
        assert_eq!(counts.deletes, 1, "should count 1 delete operation");

        // Reset and verify
        cache.reset_counts();
        let counts = cache.operation_counts();
        assert_eq!(counts.sets, 0);
        assert_eq!(counts.gets, 0);
        assert_eq!(counts.set_nx_px, 0);
        assert_eq!(counts.deletes, 0);
    }

    #[tokio::test]
    async fn test_mock_cache_ttl_behavior() {
        let cache = MockCache::new();

        // Set with short TTL
        cache
            .set(b"expiring", b"value", Duration::from_millis(50))
            .await
            .unwrap();

        // Should exist immediately
        let result = cache.get(b"expiring").await.unwrap();
        assert_eq!(result, Some(b"value".to_vec()));

        // Wait for expiration
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Should be expired
        let result = cache.get(b"expiring").await.unwrap();
        assert_eq!(result, None);
    }

    #[tokio::test]
    async fn test_mock_cache_force_expire() {
        let cache = MockCache::new();

        cache
            .set(b"key1", b"value1", Duration::from_secs(60))
            .await
            .unwrap();

        // Value exists
        assert!(cache.get(b"key1").await.unwrap().is_some());

        // Force expire
        cache.force_expire(b"key1");

        // Value should be gone
        assert!(cache.get(b"key1").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn test_mock_cache_with_preloaded_data() {
        let cache = MockCache::with_data([
            (b"preload1".as_slice(), b"value1".as_slice()),
            (b"preload2", b"value2"),
            (b"preload3", b"value3"),
        ]);

        assert_eq!(cache.len(), 3);

        let v1 = cache.get(b"preload1").await.unwrap();
        assert_eq!(v1, Some(b"value1".to_vec()));

        let v2 = cache.get(b"preload2").await.unwrap();
        assert_eq!(v2, Some(b"value2".to_vec()));

        let v3 = cache.get(b"preload3").await.unwrap();
        assert_eq!(v3, Some(b"value3".to_vec()));
    }

    #[tokio::test]
    async fn test_mock_cache_len_is_empty_clear() {
        let cache = MockCache::new();

        assert!(cache.is_empty());
        assert_eq!(cache.len(), 0);

        cache
            .set(b"k1", b"v1", Duration::from_secs(60))
            .await
            .unwrap();
        cache
            .set(b"k2", b"v2", Duration::from_secs(60))
            .await
            .unwrap();

        assert!(!cache.is_empty());
        assert_eq!(cache.len(), 2);

        cache.clear();

        assert!(cache.is_empty());
        assert_eq!(cache.len(), 0);
    }

    #[tokio::test]
    async fn test_mock_cache_set_nx_px_on_expired_key() {
        let cache = MockCache::new();

        // Set with short TTL
        cache
            .set_nx_px(b"key", b"value1", Duration::from_millis(50))
            .await
            .unwrap();

        // Wait for expiration
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Should be able to set again since key expired
        let was_set = cache
            .set_nx_px(b"key", b"value2", Duration::from_secs(60))
            .await
            .unwrap();
        assert!(was_set, "should succeed on expired key");

        let result = cache.get(b"key").await.unwrap();
        assert_eq!(result, Some(b"value2".to_vec()));
    }
}

// ============================================================================
// Mock Set Tests
// ============================================================================

mod mock_set_tests {
    use crate::mock::MockSet;

    #[tokio::test]
    async fn test_mock_set_add_and_contains() {
        let set = MockSet::new();

        assert!(set.is_empty());
        assert!(!set.contains("item1"));

        set.add_item("item1").await.unwrap();
        set.add_item("item2").await.unwrap();

        assert_eq!(set.len(), 2);
        assert!(set.contains("item1"));
        assert!(set.contains("item2"));
        assert!(!set.contains("item3"));
    }

    #[tokio::test]
    async fn test_mock_set_remove() {
        let set = MockSet::new();

        set.add_item("item1").await.unwrap();
        set.add_item("item2").await.unwrap();

        assert_eq!(set.len(), 2);

        set.remove_item("item1").await.unwrap();

        assert_eq!(set.len(), 1);
        assert!(!set.contains("item1"));
        assert!(set.contains("item2"));
    }

    #[tokio::test]
    async fn test_mock_set_batch_operations() {
        let set = MockSet::new();

        set.add_items(&[
            "a".to_owned(),
            "b".to_owned(),
            "c".to_owned(),
            "d".to_owned(),
        ])
        .await
        .unwrap();

        assert_eq!(set.len(), 4);

        set.remove_items(&["a".to_owned(), "c".to_owned()])
            .await
            .unwrap();

        assert_eq!(set.len(), 2);
        assert!(!set.contains("a"));
        assert!(set.contains("b"));
        assert!(!set.contains("c"));
        assert!(set.contains("d"));
    }

    #[tokio::test]
    async fn test_mock_set_load_items() {
        let set = MockSet::new();

        set.add_items(&["x".to_owned(), "y".to_owned(), "z".to_owned()])
            .await
            .unwrap();

        let items = set.load_items().await.unwrap();
        assert_eq!(items.len(), 3);
        assert!(items.contains(&"x".to_owned()));
        assert!(items.contains(&"y".to_owned()));
        assert!(items.contains(&"z".to_owned()));
    }

    #[tokio::test]
    async fn test_mock_set_trim_to() {
        let set = MockSet::new();

        for i in 0..20 {
            set.add_item(&format!("item{i}")).await.unwrap();
        }

        assert_eq!(set.len(), 20);

        set.trim_to(10).await.unwrap();
        assert_eq!(set.len(), 10);

        set.trim_to(5).await.unwrap();
        assert_eq!(set.len(), 5);

        // trim_to(0) should do nothing
        set.trim_to(0).await.unwrap();
        assert_eq!(set.len(), 5);
    }

    #[tokio::test]
    async fn test_mock_set_error_mode() {
        let set = MockSet::new();

        set.add_item("item1").await.unwrap();

        set.enable_error_mode("Set operation failed");

        assert!(set.add_item("item2").await.is_err());
        assert!(set.remove_item("item1").await.is_err());
        assert!(set.load_items().await.is_err());
        assert!(set.trim_to(1).await.is_err());
        assert!(set.add_items(&["x".to_owned()]).await.is_err());
        assert!(set.remove_items(&["item1".to_owned()]).await.is_err());

        set.disable_error_mode();

        // Operations should work again
        let items = set.load_items().await.unwrap();
        assert!(items.contains(&"item1".to_owned()));
    }

    #[tokio::test]
    async fn test_mock_set_empty_operations() {
        let set = MockSet::new();

        // Empty batch operations should not error
        set.add_items(&[]).await.unwrap();
        set.remove_items(&[]).await.unwrap();

        assert!(set.is_empty());
    }

    #[tokio::test]
    async fn test_mock_set_clear() {
        let set = MockSet::new();

        set.add_items(&["a".to_owned(), "b".to_owned(), "c".to_owned()])
            .await
            .unwrap();

        assert_eq!(set.len(), 3);

        set.clear();

        assert!(set.is_empty());
        assert_eq!(set.len(), 0);
    }

    #[tokio::test]
    async fn test_mock_set_duplicate_add() {
        let set = MockSet::new();

        set.add_item("dup").await.unwrap();
        set.add_item("dup").await.unwrap();
        set.add_item("dup").await.unwrap();

        // Set should only have one entry (deduplication)
        assert_eq!(set.len(), 1);
    }
}

// ============================================================================
// Cache Trait Generic Tests with Mock
// ============================================================================

mod cache_trait_tests_with_mock {
    use super::*;
    use crate::mock::MockCache;

    /// Generic test function that can test any Cache implementation
    async fn verify_cache_contract<C: Cache>(cache: &C, test_name: &str) {
        let key1 = format!("{test_name}_key1").into_bytes();
        let key2 = format!("{test_name}_key2").into_bytes();
        let value1 = b"test_value_1";
        let value2 = b"test_value_2";

        // 1. Get on empty cache returns None
        let result = cache.get(&key1).await.unwrap();
        assert_eq!(result, None, "{test_name}: empty cache get should be None");

        // 2. Set then get returns value
        cache
            .set(&key1, value1, Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(&key1).await.unwrap();
        assert_eq!(
            result,
            Some(value1.to_vec()),
            "{test_name}: set-get roundtrip"
        );

        // 3. Set overwrites
        cache
            .set(&key1, value2, Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(&key1).await.unwrap();
        assert_eq!(
            result,
            Some(value2.to_vec()),
            "{test_name}: set should overwrite"
        );

        // 4. Delete removes key
        cache.del(&key1).await.unwrap();
        let result = cache.get(&key1).await.unwrap();
        assert_eq!(result, None, "{test_name}: del should remove key");

        // 5. set_nx_px on new key succeeds
        let was_set = cache
            .set_nx_px(&key2, value1, Duration::from_secs(60))
            .await
            .unwrap();
        assert!(was_set, "{test_name}: set_nx_px on new key should succeed");

        // 6. set_nx_px on existing key fails
        let was_set = cache
            .set_nx_px(&key2, value2, Duration::from_secs(60))
            .await
            .unwrap();
        assert!(
            !was_set,
            "{test_name}: set_nx_px on existing key should fail"
        );

        // 7. Value unchanged after failed set_nx_px
        let result = cache.get(&key2).await.unwrap();
        assert_eq!(
            result,
            Some(value1.to_vec()),
            "{test_name}: value should be unchanged after failed set_nx_px"
        );
    }

    #[tokio::test]
    async fn test_local_cache_contract() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));
        verify_cache_contract(&cache, "local").await;
    }

    #[tokio::test]
    async fn test_mock_cache_contract() {
        let cache = MockCache::new();
        verify_cache_contract(&cache, "mock").await;
    }

    #[tokio::test]
    async fn test_both_caches_same_behavior() {
        let local = LocalCache::new(1000, Duration::from_secs(60));
        let mock = MockCache::new();

        // Test identical sequences
        let test_keys: Vec<&[u8]> = vec![b"k1", b"k2", b"k3"];
        let test_values: Vec<&[u8]> = vec![b"v1", b"v2", b"v3"];

        for (key, value) in test_keys.iter().zip(test_values.iter()) {
            local
                .set(key, value, Duration::from_secs(60))
                .await
                .unwrap();
            mock.set(key, value, Duration::from_secs(60)).await.unwrap();

            let local_result = local.get(key).await.unwrap();
            let mock_result = mock.get(key).await.unwrap();

            assert_eq!(
                local_result, mock_result,
                "Local and Mock should behave identically"
            );
        }
    }
}

// ============================================================================
// LocalCache Debug Implementation Test
// ============================================================================

mod local_cache_debug_tests {
    use super::*;

    #[test]
    fn test_local_cache_debug() {
        let cache = LocalCache::new(100, Duration::from_secs(60));
        let debug_str = format!("{cache:?}");

        assert!(debug_str.contains("LocalCache"));
        assert!(debug_str.contains("inner"));
    }

    #[tokio::test]
    async fn test_local_cache_clone_is_debug() {
        let cache1 = LocalCache::new(100, Duration::from_secs(60));
        cache1
            .set(b"key", b"value", Duration::from_secs(60))
            .await
            .unwrap();

        let cache2 = cache1.clone();
        let debug1 = format!("{cache1:?}");
        let debug2 = format!("{cache2:?}");

        // Both should be debuggable
        assert!(debug1.contains("LocalCache"));
        assert!(debug2.contains("LocalCache"));
    }
}

// ============================================================================
// PerEntryExpiry Tests - Testing internal expiry behavior
// ============================================================================

mod per_entry_expiry_tests {
    use super::*;

    /// Test that TTL update changes expiration time
    #[tokio::test]
    async fn test_ttl_update_on_overwrite() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));
        let key = b"ttl_update_key";

        // Set with short TTL
        cache
            .set(key, b"value1", Duration::from_millis(50))
            .await
            .unwrap();

        // Immediately overwrite with longer TTL
        cache
            .set(key, b"value2", Duration::from_secs(60))
            .await
            .unwrap();

        // Wait past original TTL
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Should still exist because TTL was updated
        let result = cache.get(key).await.unwrap();
        assert_eq!(
            result,
            Some(b"value2".to_vec()),
            "TTL should be updated on overwrite"
        );
    }

    /// Test multiple TTL updates
    #[tokio::test]
    async fn test_multiple_ttl_updates() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));
        let key = b"multi_ttl_key";

        // Chain of sets with different TTLs
        cache
            .set(key, b"v1", Duration::from_millis(10))
            .await
            .unwrap();
        cache
            .set(key, b"v2", Duration::from_millis(20))
            .await
            .unwrap();
        cache
            .set(key, b"v3", Duration::from_millis(30))
            .await
            .unwrap();
        cache
            .set(key, b"v4", Duration::from_secs(60))
            .await
            .unwrap();

        // Wait past all short TTLs
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Should still exist with latest TTL
        let result = cache.get(key).await.unwrap();
        assert_eq!(result, Some(b"v4".to_vec()));
    }

    /// Test that reads don't extend TTL
    #[tokio::test]
    async fn test_read_does_not_extend_ttl() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));
        let key = b"no_extend_key";

        cache
            .set(key, b"value", Duration::from_millis(100))
            .await
            .unwrap();

        // Read multiple times
        for _ in 0..10 {
            let _ = cache.get(key).await.unwrap();
            tokio::time::sleep(Duration::from_millis(10)).await;
        }

        // Should expire after ~100ms regardless of reads
        tokio::time::sleep(Duration::from_millis(50)).await;

        let result = cache.get(key).await.unwrap();
        assert_eq!(result, None, "reads should not extend TTL");
    }

    /// Test expiry behavior with very short TTLs
    #[tokio::test]
    async fn test_very_short_ttl_variations() {
        let cache = LocalCache::new(1000, Duration::from_secs(60));

        // Test 1ms TTL
        cache
            .set(b"1ms", b"value", Duration::from_millis(1))
            .await
            .unwrap();

        // Test 5ms TTL
        cache
            .set(b"5ms", b"value", Duration::from_millis(5))
            .await
            .unwrap();

        // Test 10ms TTL
        cache
            .set(b"10ms", b"value", Duration::from_millis(10))
            .await
            .unwrap();

        // Wait 20ms - all should be expired
        tokio::time::sleep(Duration::from_millis(20)).await;

        assert!(cache.get(b"1ms").await.unwrap().is_none());
        assert!(cache.get(b"5ms").await.unwrap().is_none());
        assert!(cache.get(b"10ms").await.unwrap().is_none());
    }
}

// ============================================================================
// Minimum Capacity Tests
// ============================================================================

mod minimum_capacity_tests {
    use super::*;

    #[tokio::test]
    async fn test_capacity_zero_becomes_one() {
        // LocalCache::new uses max(1, capacity), so capacity 0 becomes 1
        let cache = LocalCache::new(0, Duration::from_secs(60));

        // Should still work
        cache
            .set(b"key", b"value", Duration::from_secs(60))
            .await
            .unwrap();
        let result = cache.get(b"key").await.unwrap();
        assert_eq!(result, Some(b"value".to_vec()));
    }

    #[tokio::test]
    async fn test_capacity_one() {
        let cache = LocalCache::new(1, Duration::from_secs(60));

        cache
            .set(b"k1", b"v1", Duration::from_secs(60))
            .await
            .unwrap();

        // Moka's eviction is async, so we might briefly have both
        // But eventually one should be evicted
        cache
            .set(b"k2", b"v2", Duration::from_secs(60))
            .await
            .unwrap();

        // Give time for eviction
        tokio::time::sleep(Duration::from_millis(100)).await;

        // At least k2 should exist
        let k2_result = cache.get(b"k2").await.unwrap();
        assert_eq!(k2_result, Some(b"v2".to_vec()));
    }
}

// ============================================================================
// Default TTL Boundary Tests
// ============================================================================

mod default_ttl_tests {
    use super::*;

    #[tokio::test]
    async fn test_per_entry_ttl_shorter_than_default() {
        // Cache with 10 second default TTL
        let cache = LocalCache::new(1000, Duration::from_secs(10));

        // Set with 50ms TTL (shorter than default)
        cache
            .set(b"short", b"value", Duration::from_millis(50))
            .await
            .unwrap();

        // Should exist immediately
        assert!(cache.get(b"short").await.unwrap().is_some());

        // Wait for per-entry TTL to expire
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Should be expired (per-entry TTL takes precedence)
        assert!(cache.get(b"short").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn test_per_entry_ttl_longer_than_default() {
        // Cache with very short default TTL (but Moka also respects per-entry expiry)
        let cache = LocalCache::new(1000, Duration::from_millis(50));

        // Set with longer per-entry TTL
        cache
            .set(b"long", b"value", Duration::from_secs(60))
            .await
            .unwrap();

        // Wait past default TTL
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Per-entry TTL should still be honored
        // Note: Moka's behavior here depends on configuration
        // The time_to_live sets an upper bound
        let result = cache.get(b"long").await.unwrap();

        // This might be None or Some depending on how Moka handles the conflict
        // The important thing is it doesn't crash
        if result.is_none() {
            // Default TTL took precedence - this is acceptable behavior
        } else {
            // Per-entry TTL took precedence - also acceptable
            assert_eq!(result, Some(b"value".to_vec()));
        }
    }
}

// ============================================================================
// Error Message Tests
// ============================================================================

mod error_message_tests {
    use crate::Cache;
    use crate::mock::MockCache;
    use std::time::Duration;

    #[tokio::test]
    async fn test_error_message_preservation() {
        let cache = MockCache::new();

        let custom_error = "Custom error: connection refused (host: localhost:6379)";
        cache.enable_error_mode(custom_error);

        let err = cache.get(b"key").await.unwrap_err();
        let err_msg = err.to_string();

        assert!(err_msg.contains("connection refused"));
        assert!(err_msg.contains("localhost:6379"));
    }

    #[tokio::test]
    async fn test_multiple_error_mode_switches() {
        let cache = MockCache::new();

        // Start with no error
        cache
            .set(b"k1", b"v1", Duration::from_secs(60))
            .await
            .unwrap();

        // Enable error
        cache.enable_error_mode("error 1");
        assert!(cache.get(b"k1").await.is_err());

        // Disable
        cache.disable_error_mode();
        assert!(cache.get(b"k1").await.is_ok());

        // Enable different error
        cache.enable_error_mode("error 2");
        let err = cache.get(b"k1").await.unwrap_err();
        assert!(err.to_string().contains("error 2"));

        // Disable again
        cache.disable_error_mode();
        assert_eq!(cache.get(b"k1").await.unwrap(), Some(b"v1".to_vec()));
    }
}