use announcement::Announcement; #[cfg(feature = "std")] use announcement::ListenError; use std::sync::Arc; use std::time::Duration; // ============================================================================= // Basic async tests // ============================================================================= /// Test: Basic async listen operation /// /// Verifies: Listener can wait asynchronously and receive announced value /// Runtime: tokio /// Category: integration /// Coverage: Async listen path, task spawning, value reception #[tokio::test] async fn test_async_listen() { let (announcer, announcement) = Announcement::new(); let listener = announcement.listener(); tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(10)).await; announcer.announce(42).unwrap(); }); let result = listener.listen().await.unwrap(); assert_eq!(result, 42); } /// Test: Multiple async listeners receive same broadcast value /// /// Verifies: Broadcast semantics work correctly across multiple async tasks /// Runtime: tokio /// Category: integration /// Coverage: Concurrent async listeners, task spawning, value cloning #[tokio::test] async fn test_multiple_async_listeners() { let (announcer, announcement) = Announcement::new(); let l1 = announcement.listener(); let l2 = announcement.listener(); let l3 = announcement.listener(); let h1 = tokio::spawn(async move { l1.listen().await.unwrap() }); let h2 = tokio::spawn(async move { l2.listen().await.unwrap() }); let h3 = tokio::spawn(async move { l3.listen().await.unwrap() }); tokio::time::sleep(Duration::from_millis(10)).await; announcer.announce(42).unwrap(); assert_eq!(h1.await.unwrap(), 42); assert_eq!(h2.await.unwrap(), 42); assert_eq!(h3.await.unwrap(), 42); } /// Test: Listener created after announce receives value immediately /// /// Verifies: Fast path works - listeners get value immediately if already announced /// Runtime: tokio /// Category: integration /// Coverage: Fast path optimization, post-announce listener creation #[tokio::test] async fn test_listen_after_announce() { let (announcer, announcement) = Announcement::new(); announcer.announce(42).unwrap(); let listener = announcement.listener(); let result = listener.listen().await.unwrap(); // Returns immediately assert_eq!(result, 42); } /// Test: Cloned announcement channels share the same value /// /// Verifies: Cloning announcement preserves broadcast semantics /// Runtime: tokio /// Category: integration /// Coverage: Announcement cloning, shared state, value propagation #[tokio::test] async fn test_clone_announcement() { let (announcer, announcement) = Announcement::new(); let announcement2 = announcement.clone(); let l1 = announcement.listener(); let l2 = announcement2.listener(); announcer.announce(42).unwrap(); assert_eq!(l1.listen().await.unwrap(), 42); assert_eq!(l2.listen().await.unwrap(), 42); } // ============================================================================= // Arc usage test with refcount verification // ============================================================================= /// Test: Arc-wrapped values maintain correct refcount /// /// Verifies: Arc reference counting works correctly with shared values /// Runtime: tokio /// Category: memory /// Coverage: Arc refcount tracking, value sharing, memory efficiency #[tokio::test] async fn test_with_arc() { #[derive(Clone, Debug, PartialEq)] struct Data { value: Vec, } let (announcer, announcement) = Announcement::new(); let data = Arc::new(Data { value: vec![1, 2, 3], }); let l1 = announcement.listener(); let l2 = announcement.listener(); announcer.announce(data.clone()).unwrap(); let r1 = l1.listen().await.unwrap(); let r2 = l2.listen().await.unwrap(); assert_eq!(r1, data); assert_eq!(r2, data); // Verify Arc refcount is 4 (original + announcement + r1 + r2) assert_eq!(Arc::strong_count(&r1), 4); } // ============================================================================= // Blocking tests (requires std feature) // ============================================================================= /// Test: Blocking listen with timeout succeeds when value is announced /// /// Verifies: Blocking listen API works correctly in multi-threaded context /// Category: integration /// Coverage: Blocking listen path, thread spawning, timeout handling #[cfg(feature = "std")] #[test] fn test_blocking_listen() { use std::thread; let (announcer, announcement) = Announcement::new(); let listener = announcement.listener(); let handle = thread::spawn(move || listener.listen_timeout_blocking(Duration::from_millis(100))); thread::sleep(Duration::from_millis(10)); announcer.announce(42).unwrap(); assert_eq!(handle.join().unwrap(), Ok(42)); } /// Test: Blocking listen with longer timeout succeeds /// /// Verifies: Timeout parameter is respected and value is received before timeout /// Category: integration /// Coverage: Timeout configuration, successful receive path #[cfg(feature = "std")] #[test] fn test_blocking_timeout_success() { use std::thread; let (announcer, announcement) = Announcement::new(); let listener = announcement.listener(); let handle = thread::spawn(move || listener.listen_timeout_blocking(Duration::from_secs(1))); thread::sleep(Duration::from_millis(10)); announcer.announce(42).unwrap(); assert_eq!(handle.join().unwrap(), Ok(42)); } /// Test: Blocking listen detects closed channel immediately /// /// Verifies: Closed detection works correctly in blocking mode /// Category: integration /// Coverage: Channel closure detection, error propagation #[cfg(feature = "std")] #[test] fn test_blocking_timeout_failure() { let (_, announcement) = Announcement::::new(); // Note: Announcer dropped immediately via _ pattern let listener = announcement.listener(); // With closed detection, this returns Closed immediately, not Timeout let result = listener.listen_timeout_blocking(Duration::from_millis(50)); assert!(matches!(result, Err(ListenError::Closed))); assert!(announcement.is_closed()); } // ============================================================================= // CRITICAL: Race condition protection test (TOCTOU) // ============================================================================= /// Test: TOCTOU race condition protection with tight timing /// /// Verifies: Double-check pattern prevents race between listen setup and announce /// Runtime: tokio /// Category: race /// Coverage: TOCTOU protection, concurrent announce during listen setup /// Note: Runs 1000 iterations to catch race conditions #[tokio::test] async fn test_race_condition_protection() { // Test the TOCTOU protection in listen() for _ in 0..1000 { let (announcer, announcement) = Announcement::new(); let listener = announcement.listener(); let listen_handle = tokio::spawn(async move { listener.listen().await.unwrap() }); // Announce immediately (race with listen setup) announcer.announce(42).unwrap(); assert_eq!(listen_handle.await.unwrap(), 42); } } // ============================================================================= // Additional edge case tests // ============================================================================= /// Test: Multiple listeners created after announce all receive value /// /// Verifies: Fast path works for multiple post-announce listeners /// Runtime: tokio /// Category: edge /// Coverage: Multiple fast-path retrievals, late listener creation #[tokio::test] async fn test_multiple_listeners_created_after_announce() { let (announcer, announcement) = Announcement::new(); // Announce first announcer.announce(42).unwrap(); // Create listeners after announcement let l1 = announcement.listener(); let l2 = announcement.listener(); let l3 = announcement.listener(); // All should get the value immediately assert_eq!(l1.listen().await.unwrap(), 42); assert_eq!(l2.listen().await.unwrap(), 42); assert_eq!(l3.listen().await.unwrap(), 42); } /// Test: Listener can be called multiple times with same result /// /// Verifies: Listen operation is idempotent and reusable /// Runtime: tokio /// Category: edge /// Coverage: Multiple listen calls on same listener, value stability #[tokio::test] async fn test_listener_multiple_calls() { let (announcer, announcement) = Announcement::new(); let listener = announcement.listener(); announcer.announce(42).unwrap(); // Calling listen multiple times should return the same value assert_eq!(listener.listen().await.unwrap(), 42); assert_eq!(listener.listen().await.unwrap(), 42); assert_eq!(listener.listen().await.unwrap(), 42); } /// Test: High concurrency with 100 simultaneous listeners /// /// Verifies: Broadcast scales correctly with many concurrent listeners /// Runtime: tokio /// Category: stress /// Coverage: High listener count, concurrent wakeup, value distribution #[tokio::test] async fn test_concurrent_listeners() { let (announcer, announcement) = Announcement::new(); // Create many listeners let mut handles = Vec::new(); for i in 0..100 { let listener = announcement.listener(); handles.push(tokio::spawn(async move { (i, listener.listen().await.unwrap()) })); } // Small delay to ensure listeners are waiting tokio::time::sleep(Duration::from_millis(10)).await; // Announce announcer.announce(999).unwrap(); // All should receive the same value for handle in handles { let (_, value) = handle.await.unwrap(); assert_eq!(value, 999); } } /// Test: Large data broadcast with Arc maintains efficient refcounting /// /// Verifies: Large payloads work efficiently via Arc sharing without copying /// Runtime: tokio /// Category: memory /// Coverage: Large data handling, Arc efficiency, memory usage #[tokio::test] async fn test_announcement_with_large_data() { #[derive(Clone, Debug, PartialEq)] struct LargeData { data: Vec, } let (announcer, announcement) = Announcement::new(); let large_data = Arc::new(LargeData { data: vec![42; 1_000_000], // 1MB of data }); let listener = announcement.listener(); announcer.announce(large_data.clone()).unwrap(); let result = listener.listen().await.unwrap(); assert_eq!(result, large_data); // With Arc, we should have 3 references (original + announcement storage + result) assert_eq!(Arc::strong_count(&result), 3); } // ============================================================================= // Memory leak tests // ============================================================================= /// Test: No memory leak when announcer drops without announcing /// /// Verifies: Channel cleanup happens correctly without announcement /// Category: memory /// Coverage: Resource cleanup, drop behavior, leak prevention /// Note: Use valgrind/miri to verify #[test] fn test_no_leak_when_announcer_dropped() { let (announcer, announcement) = Announcement::::new(); let listener = announcement.listener(); drop(announcer); // Drop without announcing drop(listener); drop(announcement); // All memory should be freed // Use valgrind/miri to verify } /// Test: No memory leak when value is announced but no listeners exist /// /// Verifies: Announced values are properly dropped when unused /// Category: memory /// Coverage: Unused value cleanup, memory management #[test] fn test_no_leak_when_no_listeners() { let (announcer, announcement) = Announcement::>::new(); announcer.announce(vec![1; 1000]).unwrap(); drop(announcement); // Value should be dropped } // ============================================================================= // smol tests // ============================================================================= /// Test: Basic async operation with smol runtime /// /// Verifies: Announcement works correctly with smol async runtime /// Runtime: smol /// Category: integration /// Coverage: Smol runtime compatibility, async listen, task spawning #[test] fn test_smol_basic() { smol::block_on(async { let (announcer, announcement) = Announcement::new(); let listener = announcement.listener(); smol::spawn(async move { smol::Timer::after(Duration::from_millis(10)).await; announcer.announce(42).unwrap(); }) .detach(); let result = listener.listen().await.unwrap(); assert_eq!(result, 42); }); } /// Test: Multiple listeners with smol runtime /// /// Verifies: Broadcast semantics work correctly with smol runtime /// Runtime: smol /// Category: integration /// Coverage: Multiple smol tasks, concurrent listeners, value sharing #[test] fn test_smol_multiple_listeners() { smol::block_on(async { let (announcer, announcement) = Announcement::new(); let l1 = announcement.listener(); let l2 = announcement.listener(); let h1 = smol::spawn(async move { l1.listen().await.unwrap() }); let h2 = smol::spawn(async move { l2.listen().await.unwrap() }); announcer.announce(42).unwrap(); assert_eq!(h1.await, 42); assert_eq!(h2.await, 42); }); }