phytrace_sdk/transport/

mock.rs

//! Mock transport for testing.
//!
//! Provides a configurable mock transport that records sent events
//! and can simulate various success/failure scenarios.

#![expect(
    clippy::unwrap_used,
    reason = "mock transport for testing uses unwrap for clarity"
)]

use async_trait::async_trait;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use super::traits::{BatchResult, SendResult, Transport, TransportStats};
use crate::error::PhyTraceResult;
use crate::models::event::UdmEvent;

/// Configuration for mock transport behavior.
#[derive(Debug, Clone)]
pub struct MockConfig {
    /// Simulate connection state.
    pub connected: bool,

    /// Simulate latency (milliseconds).
    pub latency_ms: u64,

    /// Failure rate (0.0 to 1.0).
    pub failure_rate: f64,

    /// Status code to return on failure.
    pub failure_status: u16,

    /// Error message on failure.
    pub failure_message: String,
}

impl Default for MockConfig {
    fn default() -> Self {
        Self {
            connected: true,
            latency_ms: 10,
            failure_rate: 0.0,
            failure_status: 500,
            failure_message: "Mock failure".to_string(),
        }
    }
}

/// Mock transport for testing.
#[derive(Debug)]
pub struct MockTransport {
    config: Arc<Mutex<MockConfig>>,
    connected: AtomicBool,
    events: Arc<Mutex<Vec<UdmEvent>>>,
    stats: Arc<MockStats>,
}

#[derive(Debug, Default)]
struct MockStats {
    events_sent: AtomicU64,
    events_succeeded: AtomicU64,
    events_failed: AtomicU64,
    bytes_sent: AtomicU64,
    total_latency_ms: AtomicU64,
}

impl MockTransport {
    /// Create a new mock transport with default config.
    pub fn new() -> Self {
        Self::with_config(MockConfig::default())
    }

    /// Create a mock transport with custom config.
    pub fn with_config(config: MockConfig) -> Self {
        let connected = config.connected;
        Self {
            config: Arc::new(Mutex::new(config)),
            connected: AtomicBool::new(connected),
            events: Arc::new(Mutex::new(Vec::new())),
            stats: Arc::new(MockStats::default()),
        }
    }

    /// Get all sent events.
    pub fn sent_events(&self) -> Vec<UdmEvent> {
        self.events.lock().unwrap().clone()
    }

    /// Get the last sent event.
    pub fn last_event(&self) -> Option<UdmEvent> {
        self.events.lock().unwrap().last().cloned()
    }

    /// Get count of sent events.
    pub fn event_count(&self) -> usize {
        self.events.lock().unwrap().len()
    }

    /// Clear sent events.
    pub fn clear_events(&self) {
        self.events.lock().unwrap().clear();
    }

    /// Update the mock configuration.
    pub fn set_config(&self, config: MockConfig) {
        *self.config.lock().unwrap() = config;
    }

    /// Set failure rate.
    pub fn set_failure_rate(&self, rate: f64) {
        self.config.lock().unwrap().failure_rate = rate;
    }

    /// Set latency.
    pub fn set_latency(&self, latency_ms: u64) {
        self.config.lock().unwrap().latency_ms = latency_ms;
    }

    /// Simulate a failure decision.
    fn should_fail(&self) -> bool {
        let config = self.config.lock().unwrap();
        if config.failure_rate <= 0.0 {
            return false;
        }
        if config.failure_rate >= 1.0 {
            return true;
        }
        rand::random::<f64>() < config.failure_rate
    }

    /// Get simulated latency.
    fn get_latency(&self) -> u64 {
        self.config.lock().unwrap().latency_ms
    }

    /// Get failure info.
    fn failure_info(&self) -> (u16, String) {
        let config = self.config.lock().unwrap();
        (config.failure_status, config.failure_message.clone())
    }
}

impl Default for MockTransport {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl Transport for MockTransport {
    async fn send(&self, event: &UdmEvent) -> PhyTraceResult<SendResult> {
        self.stats.events_sent.fetch_add(1, Ordering::Relaxed);

        // Simulate latency
        let latency = self.get_latency();
        if latency > 0 {
            tokio::time::sleep(Duration::from_millis(latency)).await;
        }

        // Record the event
        self.events.lock().unwrap().push(event.clone());

        // Simulate success/failure
        if self.should_fail() {
            self.stats.events_failed.fetch_add(1, Ordering::Relaxed);
            let (status, message) = self.failure_info();
            Ok(SendResult::failure(Some(status), message))
        } else {
            self.stats.events_succeeded.fetch_add(1, Ordering::Relaxed);
            self.stats
                .total_latency_ms
                .fetch_add(latency, Ordering::Relaxed);

            let bytes = serde_json::to_string(event)
                .map(|s| s.len() as u64)
                .unwrap_or(0);
            self.stats.bytes_sent.fetch_add(bytes, Ordering::Relaxed);

            Ok(SendResult::success(latency))
        }
    }

    async fn send_batch(&self, events: &[UdmEvent]) -> PhyTraceResult<BatchResult> {
        if events.is_empty() {
            return Ok(BatchResult::all_success(0, 0));
        }

        let total = events.len();
        self.stats
            .events_sent
            .fetch_add(total as u64, Ordering::Relaxed);

        // Simulate latency
        let latency = self.get_latency();
        if latency > 0 {
            tokio::time::sleep(Duration::from_millis(latency)).await;
        }

        // Record events
        {
            let mut stored = self.events.lock().unwrap();
            stored.extend(events.iter().cloned());
        }

        // Simulate success/failure
        if self.should_fail() {
            self.stats
                .events_failed
                .fetch_add(total as u64, Ordering::Relaxed);
            let (_, message) = self.failure_info();
            Ok(BatchResult::all_failed(total, &message))
        } else {
            self.stats
                .events_succeeded
                .fetch_add(total as u64, Ordering::Relaxed);
            self.stats
                .total_latency_ms
                .fetch_add(latency, Ordering::Relaxed);

            let bytes = serde_json::to_string(events)
                .map(|s| s.len() as u64)
                .unwrap_or(0);
            self.stats.bytes_sent.fetch_add(bytes, Ordering::Relaxed);

            Ok(BatchResult::all_success(total, latency))
        }
    }

    async fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Relaxed)
    }

    async fn connect(&mut self) -> PhyTraceResult<()> {
        self.connected.store(true, Ordering::Relaxed);
        Ok(())
    }

    async fn disconnect(&mut self) -> PhyTraceResult<()> {
        self.connected.store(false, Ordering::Relaxed);
        Ok(())
    }

    fn stats(&self) -> TransportStats {
        TransportStats {
            events_sent: self.stats.events_sent.load(Ordering::Relaxed),
            events_succeeded: self.stats.events_succeeded.load(Ordering::Relaxed),
            events_failed: self.stats.events_failed.load(Ordering::Relaxed),
            bytes_sent: self.stats.bytes_sent.load(Ordering::Relaxed),
            total_latency_ms: self.stats.total_latency_ms.load(Ordering::Relaxed),
            ..Default::default()
        }
    }

    fn name(&self) -> &str {
        "mock"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::models::domains::IdentityDomain;
    use crate::models::enums::SourceType;

    fn test_event() -> UdmEvent {
        UdmEvent::new(SourceType::Amr).with_identity(IdentityDomain {
            source_id: Some("test-robot".to_string()),
            ..Default::default()
        })
    }

    #[tokio::test]
    async fn test_mock_send() {
        let transport = MockTransport::new();
        let event = test_event();

        let result = transport.send(&event).await.unwrap();
        assert!(result.success);
        assert_eq!(transport.event_count(), 1);
    }

    #[tokio::test]
    async fn test_mock_batch() {
        let transport = MockTransport::new();
        let events = vec![test_event(), test_event(), test_event()];

        let result = transport.send_batch(&events).await.unwrap();
        assert!(result.all_succeeded());
        assert_eq!(transport.event_count(), 3);
    }

    #[tokio::test]
    async fn test_mock_failure() {
        let transport = MockTransport::with_config(MockConfig {
            failure_rate: 1.0, // Always fail
            failure_status: 503,
            failure_message: "Service unavailable".to_string(),
            ..Default::default()
        });

        let result = transport.send(&test_event()).await.unwrap();
        assert!(!result.success);
        assert_eq!(result.status_code, Some(503));
    }

    #[tokio::test]
    async fn test_mock_stats() {
        let transport = MockTransport::new();

        transport.send(&test_event()).await.unwrap();
        transport.send(&test_event()).await.unwrap();

        let stats = transport.stats();
        assert_eq!(stats.events_sent, 2);
        assert_eq!(stats.events_succeeded, 2);
    }

    #[tokio::test]
    async fn test_mock_get_events() {
        let transport = MockTransport::new();
        let event = test_event();

        transport.send(&event).await.unwrap();

        let sent = transport.sent_events();
        assert_eq!(sent.len(), 1);
        assert_eq!(
            sent[0].identity.as_ref().unwrap().source_id,
            Some("test-robot".to_string())
        );
    }
}