announcement/tests/integration.rs

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<u8>,
    }

    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::<i32>::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<u8>,
    }

    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::<i32>::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::<Vec<u8>>::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);
    });
}