xz-provider 0.5.0

//! Wiremock SSE test utilities and [`SseLineBuffer`] integration tests.
//!
//! # SSE Test Utilities
//!
//! - [`sse_response`] — builds a simple SSE response with all events at once
//! - [`chunked_sse_response`] — builds an SSE response for byte-level fragmentation testing
//! - [`delayed_sse_response`] — builds an SSE response with a delay for cancellation testing
//!
//! # SseLineBuffer Edge-Case Tests
//!
//! Tests verify that [`SseLineBuffer`] correctly reassembles lines under:
//! - 1-byte chunks (worst case fragmentation)
//! - Multi-byte UTF-8 characters split across chunks
//! - `\r\n` and `\n` line endings mixed
//! - `\r\n` split across chunk boundaries
//! - Empty chunks
//! - Events without `data:` prefix
//! - `clear()` and `remaining()` API methods

use std::time::Duration;

use wiremock::{Request, Respond, ResponseTemplate};

use crate::providers::sse::SseLineBuffer;

// ---------------------------------------------------------------------------
// Wiremock SSE test utilities
// ---------------------------------------------------------------------------

/// Build the SSE body string from event payloads.
///
/// Each event payload is wrapped in `data: <payload>\n\n` SSE framing.
fn build_sse_body(events: &[&str]) -> String {
    let mut body = String::new();
    for event in events {
        body.push_str("data: ");
        body.push_str(event);
        body.push_str("\n\n");
    }
    body
}

/// Create a standard SSE response with all events delivered as a single body.
///
/// Each event is automatically wrapped in `data: <payload>\n\n` framing.
///
/// # Example
///
/// ```ignore
/// use xz_provider::test_sse::sse_response;
///
/// let template = sse_response(&[
///     r#"{"delta":"Hello"}"#,
///     r#"{"delta":"World"}"#,
/// ]);
/// ```
pub fn sse_response(events: &[&str]) -> ResponseTemplate {
    ResponseTemplate::new(200).set_body_string(build_sse_body(events))
}

/// Create an SSE response for byte-level fragmentation testing.
///
/// The response body is the full SSE text, but the `chunk_size` parameter
/// indicates the fragmentation granularity. When testing, split the body
/// into chunks of `chunk_size` bytes and feed them sequentially to
/// [`SseLineBuffer::feed()`].
///
/// # Wiremock limitation
///
/// Wiremock does not support streaming responses — the full body is delivered
/// as a single HTTP response. For byte-level fragmentation testing, combine
/// this with manual [`SseLineBuffer::feed()`] calls.
pub fn chunked_sse_response(events: &[&str], chunk_size: usize) -> impl Respond {
    ChunkedSseRespond { body: build_sse_body(events).into_bytes(), _chunk_size: chunk_size }
}

/// Build an SSE response with a delay before delivery.
///
/// The entire response is delayed by `delay_ms` milliseconds. Use this to test
/// cancellation: cancel the stream before the delay expires and verify no
/// events are received.
pub fn delayed_sse_response(events: &[&str], delay_ms: u64) -> ResponseTemplate {
    let body = build_sse_body(events);
    ResponseTemplate::new(200).set_body_string(body).set_delay(Duration::from_millis(delay_ms))
}

// ---------------------------------------------------------------------------
// ChunkedSseRespond — custom Respond impl for chunked SSE
// ---------------------------------------------------------------------------

#[derive(Debug)]
struct ChunkedSseRespond {
    /// Full SSE body bytes (built from events).
    body: Vec<u8>,
    /// The fragmentation granularity (used for documentation/test setup).
    _chunk_size: usize,
}

impl Respond for ChunkedSseRespond {
    fn respond(&self, _request: &Request) -> ResponseTemplate {
        let body_str = String::from_utf8_lossy(&self.body).into_owned();
        ResponseTemplate::new(200).set_body_string(body_str)
    }
}

// ---------------------------------------------------------------------------
// SseLineBuffer edge-case tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod sse_line_buffer_tests {
    use super::*;
    use futures::StreamExt;

    /// Verify `remaining()` returns partial buffer content.
    #[test]
    fn test_remaining_returns_partial_data() {
        let mut buf = SseLineBuffer::new();
        buf.feed(b"partial-line");
        assert_eq!(buf.remaining(), "partial-line");
    }

    /// Verify `remaining()` returns empty string when buffer has no data.
    #[test]
    fn test_remaining_empty_when_no_data() {
        let buf = SseLineBuffer::new();
        assert_eq!(buf.remaining(), "");
    }

    /// Verify `remaining()` returns empty string after complete lines consumed.
    #[test]
    fn test_remaining_empty_after_complete_line() {
        let mut buf = SseLineBuffer::new();
        buf.feed(b"complete\n");
        assert_eq!(buf.remaining(), "");
    }

    /// Verify `remaining()` returns only the incomplete trailing data.
    #[test]
    fn test_remaining_only_incomplete_trailing() {
        let mut buf = SseLineBuffer::new();
        buf.feed(b"line1\npartial");
        assert_eq!(buf.remaining(), "partial");
    }

    /// Verify `clear()` resets the buffer to empty.
    #[test]
    fn test_clear_removes_all_data() {
        let mut buf = SseLineBuffer::new();
        buf.feed(b"some-data\nmore-data\n");
        buf.feed(b"still-pending");
        buf.clear();
        assert_eq!(buf.remaining(), "");
    }

    /// Verify `clear()` allows buffer reuse with fresh state.
    #[test]
    fn test_clear_then_reuse() {
        let mut buf = SseLineBuffer::new();
        buf.feed(b"old-data\n");
        buf.clear();
        let lines = buf.feed(b"new-data\n");
        assert_eq!(lines, vec!["new-data"]);
    }

    /// Verify `\r\n` split across chunk boundaries: `\r` in one chunk, `\n` in next.
    #[test]
    fn test_crlf_split_across_chunks() {
        let mut buf = SseLineBuffer::new();
        let lines1 = buf.feed(b"hello\r");
        assert!(lines1.is_empty(), "CR without LF should not yield a line");

        let lines2 = buf.feed(b"\nworld\n");
        assert_eq!(lines2, vec!["hello", "world"]);
    }

    /// Verify multiple `\r\n` pairs split across chunks.
    #[test]
    fn test_multiple_crlf_splits_across_chunks() {
        let mut buf = SseLineBuffer::new();
        let lines1 = buf.feed(b"a\r\nb\r");
        assert_eq!(lines1, vec!["a"], "a\r\n should produce 'a'");

        let lines2 = buf.feed(b"\nc\r\nd\n");
        assert_eq!(lines2, vec!["b", "c", "d"]);
    }

    /// Verify carriage return at chunk boundary without trailing newline.
    /// `\r` without following `\n` is treated as regular content, not a line ending.
    #[test]
    fn test_carriage_return_at_end_no_lf() {
        let mut buf = SseLineBuffer::new();
        let lines1 = buf.feed(b"data: hello\r");
        assert!(lines1.is_empty());

        let lines2 = buf.feed(b"world\n");
        // The `\r` is kept as content since it's not followed by `\n`
        assert_eq!(lines2, vec!["data: hello\rworld"]);
    }

    /// Verify that comment lines (starting with `:`) are returned as-is.
    /// SseLineBuffer does not interpret SSE protocol — it returns all lines.
    #[test]
    fn test_comment_lines_are_preserved() {
        let mut buf = SseLineBuffer::new();
        let lines = buf.feed(b": this is a comment\n: another comment\n");
        assert_eq!(lines, vec![": this is a comment", ": another comment"]);
    }

    /// Verify that lines without `data:` prefix are returned as-is.
    /// Higher-level protocol code is responsible for filtering.
    #[test]
    fn test_non_data_lines_are_preserved() {
        let mut buf = SseLineBuffer::new();
        let lines = buf.feed(b"event: ping\ndata: {\"msg\":\"hello\"}\n\n");
        assert_eq!(lines, vec!["event: ping", "data: {\"msg\":\"hello\"}", ""]);
    }

    /// Verify that 1-byte chunks work correctly for SSE-like data.
    #[test]
    fn test_one_byte_chunks_sse_data() {
        let mut buf = SseLineBuffer::new();
        let input = b"data: A\n\ndata: B\n\n";
        let mut all_lines = Vec::new();
        for &byte in input {
            let lines = buf.feed(&[byte]);
            all_lines.extend(lines);
        }
        assert_eq!(all_lines, vec!["data: A", "", "data: B", ""]);
    }

    /// Verify 1-byte chunks with event lines mixed in.
    #[test]
    fn test_one_byte_chunks_with_event_type() {
        let mut buf = SseLineBuffer::new();
        let input = b"event: msg\ndata: hello\n\n";
        let mut all_lines = Vec::new();
        for &byte in input {
            let lines = buf.feed(&[byte]);
            all_lines.extend(lines);
        }
        assert_eq!(all_lines, vec!["event: msg", "data: hello", ""]);
    }

    /// Verify that a large payload crossing many byte chunks is correctly reassembled.
    #[test]
    fn test_large_data_fragmented() {
        let mut buf = SseLineBuffer::new();
        let large_line = "data: ".to_owned() + &"x".repeat(1000) + "\n";
        let input = large_line.repeat(5);
        let bytes = input.as_bytes();

        let mut all_lines = Vec::new();
        for chunk in bytes.chunks(7) {
            let lines = buf.feed(chunk);
            all_lines.extend(lines);
        }

        assert_eq!(all_lines.len(), 5);
        for line in &all_lines {
            assert!(line.starts_with("data: "));
            assert_eq!(line.len(), 1006); // "data: " (6) + 1000 'x's
        }
    }

    /// Verify that a multi-line SSE response via wiremock works end-to-end.
    #[tokio::test]
    async fn test_sse_response_template_utility() {
        let mock_server = wiremock::MockServer::start().await;
        let events = &[r#"{"delta":"Hello"}"#, r#"{"delta":"World"}"#];
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/sse"))
            .respond_with(sse_response(events))
            .mount(&mock_server)
            .await;

        let client = reqwest::Client::new();
        let resp = client.post(format!("{}/sse", mock_server.uri())).send().await.unwrap();

        assert!(resp.status().is_success());

        let stream = crate::http::create_sse_stream(resp, None);
        let lines: Vec<String> =
            futures::StreamExt::filter_map(stream, |r| futures::future::ready(r.ok()))
                .collect()
                .await;

        let expected = vec!["data: {\"delta\":\"Hello\"}", "", "data: {\"delta\":\"World\"}", ""];
        assert_eq!(lines, expected);
    }

    /// Verify that delayed SSE response works with cancellation.
    #[tokio::test]
    async fn test_delayed_sse_response_cancellation() {
        let mock_server = wiremock::MockServer::start().await;
        let events = &[r#"{"delta":"Hello"}"#, r#"{"delta":"World"}"#];
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/delayed"))
            .respond_with(delayed_sse_response(events, 5000))
            .mount(&mock_server)
            .await;

        let client = reqwest::Client::new();
        // Use a short timeout so the request itself times out before the mock's delay
        let resp = client
            .post(format!("{}/delayed", mock_server.uri()))
            .timeout(Duration::from_millis(100))
            .send()
            .await;

        // The request should time out because the mock delays 5000ms but
        // the client timeout is only 100ms
        assert!(resp.is_err(), "expected timeout error");
    }

    /// Verify that the chunked SSE response utility can be used end-to-end.
    #[tokio::test]
    async fn test_chunked_sse_response_utility() {
        let mock_server = wiremock::MockServer::start().await;
        let events = &[r#"{"delta":"Hello"}"#, r#"{"delta":"World"}"#];
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/chunked"))
            .respond_with(chunked_sse_response(events, 5))
            .mount(&mock_server)
            .await;

        let client = reqwest::Client::new();
        let resp = client.post(format!("{}/chunked", mock_server.uri())).send().await.unwrap();

        assert!(resp.status().is_success());

        // The body should be the full SSE text even though the response
        // was created with chunked_sse_response
        let full_body = resp.text().await.unwrap();
        assert!(full_body.contains("Hello"));
        assert!(full_body.contains("World"));
        assert!(full_body.starts_with("data: "));
    }

    /// Verify that SseLineBuffer handles events with multiple data fields.
    #[test]
    fn test_multiple_data_fields() {
        let mut buf = SseLineBuffer::new();
        let lines = buf.feed(b"data: line1\ndata: line2\ndata: line3\n");
        assert_eq!(lines, vec!["data: line1", "data: line2", "data: line3"]);
    }

    /// Verify that an empty string fed does not change buffer state.
    #[test]
    fn test_empty_feed_after_partial() {
        let mut buf = SseLineBuffer::new();
        buf.feed(b"partial");
        let lines = buf.feed(b"");
        assert!(lines.is_empty());
        assert_eq!(buf.remaining(), "partial");
    }

    /// Verify that multiple empty feeds followed by a terminating newline work.
    #[test]
    fn test_multiple_empty_feeds_then_complete() {
        let mut buf = SseLineBuffer::new();
        buf.feed(b"");
        buf.feed(b"");
        buf.feed(b"hello\n");
        assert_eq!(buf.remaining(), "");
        let lines = buf.feed(b"");
        assert!(lines.is_empty());
    }
}

// ---------------------------------------------------------------------------
// create_sse_stream integration tests using wiremock SSE utilities
// ---------------------------------------------------------------------------

#[cfg(test)]
mod sse_integration_tests {
    use super::*;
    use futures::StreamExt;

    /// Full integration: wiremock SSE response → create_sse_stream → line collection.
    #[tokio::test]
    async fn test_create_sse_stream_with_multiple_events() {
        let mock_server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/multi"))
            .respond_with(sse_response(&[
                r#"{"type":"event1"}"#,
                r#"{"type":"event2"}"#,
                r#"{"type":"event3"}"#,
            ]))
            .mount(&mock_server)
            .await;

        let client = reqwest::Client::new();
        let resp = client.post(format!("{}/multi", mock_server.uri())).send().await.unwrap();

        let stream = crate::http::create_sse_stream(resp, None);
        let lines: Vec<String> =
            stream.filter_map(|r| futures::future::ready(r.ok())).collect().await;

        assert_eq!(lines.len(), 6); // 3 data lines + 3 empty separators
        assert_eq!(lines[0], "data: {\"type\":\"event1\"}");
        assert_eq!(lines[2], "data: {\"type\":\"event2\"}");
        assert_eq!(lines[4], "data: {\"type\":\"event3\"}");
    }

    /// Verify that a single event SSE response works correctly.
    #[tokio::test]
    async fn test_create_sse_stream_single_event() {
        let mock_server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/single"))
            .respond_with(sse_response(&[r#"{"done":true}"#]))
            .mount(&mock_server)
            .await;

        let client = reqwest::Client::new();
        let resp = client.post(format!("{}/single", mock_server.uri())).send().await.unwrap();

        let stream = crate::http::create_sse_stream(resp, None);
        let lines: Vec<String> =
            stream.filter_map(|r| futures::future::ready(r.ok())).collect().await;

        assert_eq!(lines, vec!["data: {\"done\":true}", ""]);
    }

    /// Verify that create_sse_stream with an immediate cancellation token
    /// yields very few or zero lines.
    #[tokio::test]
    async fn test_create_sse_stream_cancellation_immediate() {
        let mock_server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/cancel-immediate"))
            .respond_with(sse_response(&[
                r#"{"msg":"one"}"#,
                r#"{"msg":"two"}"#,
                r#"{"msg":"three"}"#,
            ]))
            .mount(&mock_server)
            .await;

        let client = reqwest::Client::new();
        let resp =
            client.post(format!("{}/cancel-immediate", mock_server.uri())).send().await.unwrap();

        let cancel = crate::CancellationToken::new();
        cancel.cancel(); // cancel before reading

        let stream = crate::http::create_sse_stream(resp, Some(cancel));
        let lines: Vec<String> =
            stream.filter_map(|r| futures::future::ready(r.ok())).collect().await;

        // With immediate cancellation, expect 0 or very few lines
        assert!(
            lines.len() < 3,
            "expected fewer than 3 lines with immediate cancellation, got {}",
            lines.len()
        );
    }
}