666 lines
20 KiB
Rust
666 lines
20 KiB
Rust
//! Closed Detection Tests
|
|
//!
|
|
//! Tests for the closed detection API that notifies listeners when
|
|
//! an announcer is dropped or explicitly closed without announcing.
|
|
//!
|
|
//! NOTE: All async listen() calls use short timeouts (100ms) to prevent
|
|
//! hanging if Drop impl is mutated. This ensures fast failure detection
|
|
//! in mutation testing.
|
|
|
|
use announcement::{Announcement, ListenError};
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
/// Timeout for listen operations in tests (short to catch Drop impl mutations)
|
|
const LISTEN_TIMEOUT: Duration = Duration::from_millis(100);
|
|
|
|
// =============================================================================
|
|
// Basic Closed Detection Tests
|
|
// =============================================================================
|
|
|
|
/// Test: Close test: dropped announcer notifies listener
|
|
///
|
|
/// Verifies: Verifies dropped announcer notifies listener behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_dropped_announcer_notifies_listener() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Drop announcer without announcing
|
|
drop(announcer);
|
|
|
|
// Listener should receive Closed error (with timeout to prevent hanging in mutation tests)
|
|
let result = tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout when announcer is dropped");
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
|
|
/// Test: Close test: explicit close notifies listener
|
|
///
|
|
/// Verifies: Verifies explicit close notifies listener behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_explicit_close_notifies_listener() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Explicitly close
|
|
announcer.close();
|
|
|
|
// Listener should receive Closed error (with timeout to prevent hanging)
|
|
let result = tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout when announcer is closed");
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
|
|
/// Test: Close test: multiple listeners all notified on drop
|
|
///
|
|
/// Verifies: Verifies multiple listeners all notified on drop behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_multiple_listeners_all_notified_on_drop() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
let l1 = announcement.listener();
|
|
let l2 = announcement.listener();
|
|
let l3 = announcement.listener();
|
|
|
|
// Drop announcer
|
|
drop(announcer);
|
|
|
|
// All listeners should receive Closed error
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, l1.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, l2.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, l3.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
}
|
|
|
|
/// Test: Close test: multiple listeners all notified on close
|
|
///
|
|
/// Verifies: Verifies multiple listeners all notified on close behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_multiple_listeners_all_notified_on_close() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
let l1 = announcement.listener();
|
|
let l2 = announcement.listener();
|
|
let l3 = announcement.listener();
|
|
|
|
// Explicit close
|
|
announcer.close();
|
|
|
|
// All listeners should receive Closed error
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, l1.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, l2.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, l3.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
}
|
|
|
|
/// Test: Close test: is closed after drop
|
|
///
|
|
/// Verifies: Verifies is closed after drop behavior
|
|
/// Category: close
|
|
#[test]
|
|
fn test_is_closed_after_drop() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
assert!(!announcement.is_closed());
|
|
assert!(!listener.is_closed());
|
|
|
|
drop(announcer);
|
|
|
|
assert!(announcement.is_closed());
|
|
assert!(listener.is_closed());
|
|
}
|
|
|
|
/// Test: Close test: is closed after explicit close
|
|
///
|
|
/// Verifies: Verifies is closed after explicit close behavior
|
|
/// Category: close
|
|
#[test]
|
|
fn test_is_closed_after_explicit_close() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
assert!(!announcement.is_closed());
|
|
assert!(!listener.is_closed());
|
|
|
|
announcer.close();
|
|
|
|
assert!(announcement.is_closed());
|
|
assert!(listener.is_closed());
|
|
}
|
|
|
|
// =============================================================================
|
|
// Concurrent Closed Detection Tests
|
|
// =============================================================================
|
|
|
|
/// Test: Close test: concurrent listeners waiting when closed
|
|
///
|
|
/// Verifies: Verifies concurrent listeners waiting when closed behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_concurrent_listeners_waiting_when_closed() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
// Spawn multiple waiting listeners
|
|
let mut handles = vec![];
|
|
for _ in 0..10 {
|
|
let listener = announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout")
|
|
}));
|
|
}
|
|
|
|
// Give them time to start waiting
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
|
|
// Drop announcer - all should wake with Closed error
|
|
drop(announcer);
|
|
|
|
// Verify all received Closed error
|
|
for handle in handles {
|
|
assert_eq!(handle.await.unwrap(), Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
/// Test: Close test: race drop during listen setup
|
|
///
|
|
/// Verifies: Verifies race drop during listen setup behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_race_drop_during_listen_setup() {
|
|
// Test the race condition where announcer is dropped while listener
|
|
// is setting up (between checking and registering event listener)
|
|
for _ in 0..100 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
let handle = tokio::spawn(async move {
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout")
|
|
});
|
|
|
|
// Drop immediately (race with listen setup)
|
|
drop(announcer);
|
|
|
|
// Should still receive Closed error
|
|
assert_eq!(handle.await.unwrap(), Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
/// Test: Close test: mixed waiting and polling listeners
|
|
///
|
|
/// Verifies: Verifies mixed waiting and polling listeners behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_mixed_waiting_and_polling_listeners() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
// Some listeners waiting
|
|
let mut async_handles = vec![];
|
|
for _ in 0..5 {
|
|
let listener = announcement.listener();
|
|
async_handles.push(tokio::spawn(async move {
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout")
|
|
}));
|
|
}
|
|
|
|
// Some listeners will poll after close
|
|
let poll_listeners: Vec<_> = (0..5).map(|_| announcement.listener()).collect();
|
|
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
|
|
// Close announcer
|
|
announcer.close();
|
|
|
|
// Verify waiting listeners get Closed
|
|
for handle in async_handles {
|
|
assert_eq!(handle.await.unwrap(), Err(ListenError::Closed));
|
|
}
|
|
|
|
// Verify polling listeners also get Closed
|
|
for listener in poll_listeners {
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Blocking Mode Closed Detection Tests
|
|
// =============================================================================
|
|
|
|
/// Test: Close test: listen blocking returns closed
|
|
///
|
|
/// Verifies: Verifies listen blocking returns closed behavior
|
|
/// Category: close
|
|
#[cfg(feature = "std")]
|
|
#[test]
|
|
fn test_listen_blocking_returns_closed() {
|
|
use std::thread;
|
|
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
let handle = thread::spawn(move || listener.listen_blocking());
|
|
|
|
thread::sleep(Duration::from_millis(10));
|
|
drop(announcer);
|
|
|
|
assert_eq!(handle.join().unwrap(), Err(ListenError::Closed));
|
|
}
|
|
|
|
/// Test: Close test: listen timeout blocking returns closed not timeout
|
|
///
|
|
/// Verifies: Verifies listen timeout blocking returns closed not timeout behavior
|
|
/// Category: close
|
|
#[cfg(feature = "std")]
|
|
#[test]
|
|
fn test_listen_timeout_blocking_returns_closed_not_timeout() {
|
|
use std::thread;
|
|
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
let handle = thread::spawn(move || listener.listen_timeout_blocking(Duration::from_secs(10)));
|
|
|
|
thread::sleep(Duration::from_millis(10));
|
|
drop(announcer);
|
|
|
|
// Should return Closed, not Timeout (even though timeout is 10 seconds)
|
|
assert_eq!(handle.join().unwrap(), Err(ListenError::Closed));
|
|
}
|
|
|
|
/// Test: Close test: multiple blocking threads notified
|
|
///
|
|
/// Verifies: Verifies multiple blocking threads notified behavior
|
|
/// Category: close
|
|
#[cfg(feature = "std")]
|
|
#[test]
|
|
fn test_multiple_blocking_threads_notified() {
|
|
use std::thread;
|
|
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
let mut handles = vec![];
|
|
for _ in 0..5 {
|
|
let listener = announcement.listener();
|
|
handles.push(thread::spawn(move || listener.listen_blocking()));
|
|
}
|
|
|
|
thread::sleep(Duration::from_millis(10));
|
|
drop(announcer);
|
|
|
|
for handle in handles {
|
|
assert_eq!(handle.join().unwrap(), Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Edge Cases and Ordering Tests
|
|
// =============================================================================
|
|
|
|
/// Test: Close test: listener created after close
|
|
///
|
|
/// Verifies: Verifies listener created after close behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_listener_created_after_close() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
// Close first
|
|
announcer.close();
|
|
|
|
// Create listener after close
|
|
let listener = announcement.listener();
|
|
|
|
// Should immediately return Closed (fast path)
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
}
|
|
|
|
/// Test: Close test: announce takes precedence over close
|
|
///
|
|
/// Verifies: Verifies announce takes precedence over close behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_announce_takes_precedence_over_close() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Announce a value
|
|
announcer.announce(42).unwrap();
|
|
|
|
// Listener should get the value, not Closed
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Ok(42)
|
|
);
|
|
|
|
// is_closed should be false because announcement succeeded
|
|
assert!(!announcement.is_closed());
|
|
}
|
|
|
|
/// Test: Close test: drop immediately after creation
|
|
///
|
|
/// Verifies: Verifies drop immediately after creation behavior
|
|
/// Category: close
|
|
#[test]
|
|
fn test_drop_immediately_after_creation() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Drop immediately
|
|
drop(announcer);
|
|
|
|
// Should be closed
|
|
assert!(announcement.is_closed());
|
|
assert!(listener.is_closed());
|
|
}
|
|
|
|
/// Test: Close test: try listen with closed announcer
|
|
///
|
|
/// Verifies: Verifies try listen with closed announcer behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_try_listen_with_closed_announcer() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Drop announcer
|
|
drop(announcer);
|
|
|
|
// try_listen returns None (doesn't distinguish between not-announced and closed)
|
|
assert_eq!(listener.try_listen(), None);
|
|
|
|
// But can check is_closed to distinguish
|
|
assert!(listener.is_closed());
|
|
}
|
|
|
|
/// Test: Close test: cloned announcement sees closed state
|
|
///
|
|
/// Verifies: Verifies cloned announcement sees closed state behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_cloned_announcement_sees_closed_state() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let announcement2 = announcement.clone();
|
|
|
|
let l1 = announcement.listener();
|
|
let l2 = announcement2.listener();
|
|
|
|
drop(announcer);
|
|
|
|
// Both announcements and their listeners see closed state
|
|
assert!(announcement.is_closed());
|
|
assert!(announcement2.is_closed());
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, l1.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, l2.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
}
|
|
|
|
// =============================================================================
|
|
// Multiple Runtime Tests for Closed Detection
|
|
// =============================================================================
|
|
|
|
/// Test: Close test: tokio closed detection
|
|
///
|
|
/// Verifies: Verifies tokio closed detection behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_tokio_closed_detection() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
drop(announcer);
|
|
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
#[cfg_attr(
|
|
miri,
|
|
ignore = "Miri doesn't support timerfd_create syscall used by smol"
|
|
)]
|
|
/// Test: Close test: smol closed detection
|
|
///
|
|
/// Verifies: Verifies smol closed detection behavior
|
|
/// Category: close
|
|
fn test_smol_closed_detection() {
|
|
smol::block_on(async {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
drop(announcer);
|
|
|
|
assert_eq!(
|
|
smol::future::or(
|
|
async {
|
|
smol::Timer::after(LISTEN_TIMEOUT).await;
|
|
panic!("listen should not timeout")
|
|
},
|
|
listener.listen()
|
|
)
|
|
.await,
|
|
Err(ListenError::Closed)
|
|
);
|
|
});
|
|
}
|
|
|
|
// =============================================================================
|
|
// Stress Tests for Closed Detection
|
|
// =============================================================================
|
|
|
|
/// Test: Close test: many listeners closed
|
|
///
|
|
/// Verifies: Verifies many listeners closed behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_many_listeners_closed() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
let mut handles = vec![];
|
|
for _ in 0..100 {
|
|
let listener = announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout")
|
|
}));
|
|
}
|
|
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
drop(announcer);
|
|
|
|
for handle in handles {
|
|
assert_eq!(handle.await.unwrap(), Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
/// Test: Close test: repeated close scenarios
|
|
///
|
|
/// Verifies: Verifies repeated close scenarios behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_repeated_close_scenarios() {
|
|
// Run the close scenario many times to catch any race conditions
|
|
for _ in 0..1000 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
let handle = tokio::spawn(async move {
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout")
|
|
});
|
|
|
|
drop(announcer);
|
|
|
|
assert_eq!(handle.await.unwrap(), Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Integration with Other Features
|
|
// =============================================================================
|
|
|
|
/// Test: Close test: closed with arc value type
|
|
///
|
|
/// Verifies: Verifies closed with arc value type behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_closed_with_arc_value_type() {
|
|
let (announcer, announcement) = Announcement::<Arc<String>>::new();
|
|
let listener = announcement.listener();
|
|
|
|
drop(announcer);
|
|
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
}
|
|
|
|
/// Test: Close test: is announced vs is closed
|
|
///
|
|
/// Verifies: Verifies is announced vs is closed behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_is_announced_vs_is_closed() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
// Initially: not announced, not closed
|
|
assert!(!announcement.is_announced());
|
|
assert!(!announcement.is_closed());
|
|
|
|
// After drop: not announced, is closed
|
|
drop(announcer);
|
|
assert!(!announcement.is_announced());
|
|
assert!(announcement.is_closed());
|
|
}
|
|
|
|
/// Test: Close test: announced then never closed
|
|
///
|
|
/// Verifies: Verifies announced then never closed behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_announced_then_never_closed() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
announcer.announce(42).unwrap();
|
|
|
|
// After announce: is announced, not closed
|
|
assert!(announcement.is_announced());
|
|
assert!(!announcement.is_closed());
|
|
}
|
|
|
|
// =============================================================================
|
|
// Error Handling Tests
|
|
// =============================================================================
|
|
|
|
/// Test: Close test: closed error display
|
|
///
|
|
/// Verifies: Verifies closed error display behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_closed_error_display() {
|
|
let err = ListenError::Closed;
|
|
assert_eq!(err.to_string(), "announcer was dropped without announcing");
|
|
}
|
|
|
|
/// Test: Close test: listen multiple times after closed
|
|
///
|
|
/// Verifies: Verifies listen multiple times after closed behavior
|
|
/// Category: close
|
|
#[tokio::test]
|
|
async fn test_listen_multiple_times_after_closed() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
drop(announcer);
|
|
|
|
// Multiple listen calls should all return Closed
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
assert_eq!(
|
|
tokio::time::timeout(LISTEN_TIMEOUT, listener.listen())
|
|
.await
|
|
.expect("listen should not timeout"),
|
|
Err(ListenError::Closed)
|
|
);
|
|
}
|