focus/crates/focus-providers/tests/common/mod.rs

//! 测试共享的 mock transport：录制请求并回放录制的 SSE 流。
//! Shared test mock transport: records requests and replays recorded SSE
//! streams.
//!
//! 回放时把流按固定小尺寸切块，刻意让 SSE 事件跨 chunk 边界，
//! 以验证 provider 的增量解析。
//! Replays are split into small fixed-size chunks to deliberately straddle
//! SSE event boundaries, exercising the providers' incremental parsing.

use focus_transport::{ChunkStream, HttpRequest, HttpResponse, Transport, TransportError};
use std::collections::{HashMap, VecDeque};
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::time::Duration;

/// 录制请求并回放响应体的 mock transport。
/// A mock transport that records requests and replays response bodies.
#[derive(Debug, Clone, Default)]
pub struct MockTransport {
    /// 待回放的响应体（每个消费一个）；流式接口按小块回放。
    /// Bodies to replay (one consumed per request); the streaming interface
    /// replays them in small chunks.
    streams: Arc<Mutex<VecDeque<Vec<u8>>>>,
    /// 待返回的传输层错误（每个消费一个，优先于 streams）。
    /// Transport errors to return (one consumed per request; take precedence).
    errors: Arc<Mutex<VecDeque<TransportError>>>,
    /// 已发出的请求记录。
    /// Requests sent so far.
    requests: Arc<Mutex<Vec<HttpRequest>>>,
}

impl MockTransport {
    /// 创建一个空的 mock transport。
    /// Create an empty mock transport.
    pub fn new() -> Self {
        Self::default()
    }

    /// 加入一条待回放的响应体。
    /// Queue a response body to replay.
    pub fn push_body(&mut self, body: impl Into<Vec<u8>>) {
        self.streams.lock().unwrap().push_back(body.into());
    }

    /// 加入一个待返回的传输错误。
    /// Queue a transport error to return.
    #[allow(dead_code)]
    pub fn push_error(&mut self, err: TransportError) {
        self.errors.lock().unwrap().push_back(err);
    }

    /// 最近一次请求的克隆。
    /// Clone of the most recent request.
    #[allow(dead_code)]
    pub fn last_request(&self) -> HttpRequest {
        self.requests
            .lock()
            .unwrap()
            .last()
            .cloned()
            .expect("no request recorded")
    }

    /// 已记录的请求数量。
    /// Number of recorded requests.
    #[allow(dead_code)]
    pub fn request_count(&self) -> usize {
        self.requests.lock().unwrap().len()
    }
}

impl Transport for MockTransport {
    fn request<'a>(
        &'a self,
        req: &'a HttpRequest,
    ) -> Pin<
        Box<
            dyn std::future::Future<Output = focus_transport::TransportResult<HttpResponse>>
                + Send
                + 'a,
        >,
    > {
        Box::pin(async move {
            self.requests.lock().unwrap().push(req.clone());
            if let Some(err) = self.errors.lock().unwrap().pop_front() {
                return Err(err);
            }
            let body = self.streams.lock().unwrap().pop_front().unwrap_or_default();
            Ok(HttpResponse {
                status: 200,
                headers: HashMap::new(),
                body,
            })
        })
    }

    fn stream_request<'a>(
        &'a self,
        req: &'a HttpRequest,
        _timeout: Duration,
    ) -> Pin<
        Box<
            dyn std::future::Future<
                    Output = focus_transport::TransportResult<Box<dyn ChunkStream + Send>>,
                > + Send
                + 'a,
        >,
    > {
        Box::pin(async move {
            self.requests.lock().unwrap().push(req.clone());
            if let Some(err) = self.errors.lock().unwrap().pop_front() {
                return Err(err);
            }
            let body = self.streams.lock().unwrap().pop_front().unwrap_or_default();
            Ok(Box::new(ReplayChunkStream {
                data: body,
                pos: 0,
                chunk_size: 7,
            }) as Box<dyn ChunkStream + Send>)
        })
    }
}

/// 按固定小尺寸逐块回放的 chunk 流。
/// A chunk stream that replays its data in small fixed-size chunks.
struct ReplayChunkStream {
    data: Vec<u8>,
    pos: usize,
    chunk_size: usize,
}

impl ChunkStream for ReplayChunkStream {
    fn next_chunk<'a>(
        &'a mut self,
    ) -> Pin<
        Box<
            dyn std::future::Future<Output = focus_transport::TransportResult<Option<Vec<u8>>>>
                + Send
                + 'a,
        >,
    > {
        Box::pin(async move {
            if self.pos >= self.data.len() {
                Ok(None)
            } else {
                let end = (self.pos + self.chunk_size).min(self.data.len());
                let out = self.data[self.pos..end].to_vec();
                self.pos = end;
                Ok(Some(out))
            }
        })
    }
}

/// 把一个 provider 的全部流事件收集成 Vec。
/// Collect all stream events from a provider into a Vec.
#[allow(dead_code)]
pub fn collect(
    provider: &dyn focus_core::provider::StreamProvider,
    request: &focus_core::provider::ProviderRequest,
) -> Vec<focus_core::provider::StreamEvent> {
    let mut iter = provider
        .stream(request)
        .expect("provider.stream must not return Err for a well-formed request");
    let mut out = Vec::new();
    while let Some(ev) = iter.next_event() {
        out.push(ev);
    }
    out
}