audacity-sdk 0.5.1

//! Integration tests for the spec §9 pass-through surfaces (conformance
//! items 18–19) against a local mock HTTP server — no live APIs.

use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

use audacity_sdk::{Client, Config, Error};

// ── helpers ───────────────────────────────────────────────────────────────────

/// Bind a random port and return (listener, base_url).
async fn bind() -> (TcpListener, String) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    (listener, format!("http://127.0.0.1:{port}"))
}

/// Build an HTTP/1.1 response string.
fn http_response(status: u16, status_text: &str, headers_extra: &str, body: &str) -> String {
    format!(
        "HTTP/1.1 {status} {status_text}\r\nContent-Type: application/json\r\nContent-Length: {len}\r\n{headers_extra}\r\n{body}",
        len = body.len()
    )
}

fn http_sse_response(body: &str) -> String {
    format!(
        "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n{len:x}\r\n{body}\r\n0\r\n\r\n",
        len = body.len()
    )
}

/// Read one full HTTP request (headers + Content-Length body) from the socket.
async fn read_request(sock: &mut tokio::net::TcpStream) -> String {
    let mut buf: Vec<u8> = Vec::new();
    let mut tmp = [0u8; 4096];
    loop {
        let n = sock.read(&mut tmp).await.unwrap();
        if n == 0 {
            break;
        }
        buf.extend_from_slice(&tmp[..n]);
        if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
            let headers = String::from_utf8_lossy(&buf[..pos]).to_lowercase();
            let content_length = headers
                .lines()
                .find_map(|l| l.strip_prefix("content-length:"))
                .and_then(|v| v.trim().parse::<usize>().ok())
                .unwrap_or(0);
            if buf.len() >= pos + 4 + content_length {
                break;
            }
        }
    }
    String::from_utf8_lossy(&buf).to_string()
}

/// Serve a single connection with the given raw HTTP bytes.
async fn serve_once(listener: &TcpListener, response: String) {
    serve_and_capture(listener, response).await;
}

/// Serve a single connection and return the raw captured request.
async fn serve_and_capture(listener: &TcpListener, response: String) -> String {
    let (mut sock, _) = listener.accept().await.unwrap();
    let req = read_request(&mut sock).await;
    sock.write_all(response.as_bytes()).await.unwrap();
    req
}

/// Parse the JSON body out of a captured request.
fn request_body(req: &str) -> Value {
    let body = req.split("\r\n\r\n").nth(1).expect("request has a body");
    serde_json::from_str(body).expect("request body is JSON")
}

fn client_for(base_url: &str) -> Client {
    Client::new(
        &Config::builder()
            .api_key("audacity_api_test")
            .base_url(base_url)
            .max_retries(0)
            .build()
            .unwrap(),
    )
    .unwrap()
}

/// Drain a raw event stream, returning yielded events and the terminal error
/// (if any).
async fn drain_stream(mut stream: audacity_sdk::RawEventStream) -> (Vec<Value>, Option<Error>) {
    let mut events = Vec::new();
    loop {
        match stream.recv().await {
            Ok(Some(event)) => events.push(event),
            Ok(None) => return (events, None),
            Err(e) => return (events, Some(e)),
        }
    }
}

// ── conformance §5 item 18 — OpenAI pass-through, non-streaming ──────────────

#[tokio::test]
async fn openai_create_verbatim_request_and_raw_response() {
    let (listener, base_url) = bind().await;

    // Raw OpenAI-shaped response including a field the SDK doesn't know.
    let response_body = r#"{"id":"chatcmpl-1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"Hello"},"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":3},"future_response_field":{"nested":true}}"#;
    let response = http_response(200, "OK", "", response_body);

    let params = json!({
        "model": "gpt-5.4-mini",
        "messages": [{"role": "user", "content": "Hi"}],
        "max_tokens": 100,
        "future_request_field": {"knob": [1, 2, 3]},
    });

    let (result, req) = tokio::join!(
        async {
            client_for(&base_url)
                .chat_completions()
                .create(params)
                .await
        },
        serve_and_capture(&listener, response),
    );

    // Path + auth header.
    assert!(req.starts_with("POST /v1/chat/completions HTTP/1.1"));
    assert!(req
        .to_lowercase()
        .contains("authorization: bearer audacity_api_test"));

    // Body verbatim — unknown fields preserved, nothing added or renamed.
    let body = request_body(&req);
    assert_eq!(
        body,
        json!({
            "model": "gpt-5.4-mini",
            "messages": [{"role": "user", "content": "Hi"}],
            "max_tokens": 100,
            "future_request_field": {"knob": [1, 2, 3]},
        })
    );

    // Raw response returned untranslated.
    let out = result.unwrap();
    assert_eq!(out["choices"][0]["message"]["content"], "Hello");
    assert_eq!(out["future_response_field"]["nested"], true);
    assert_eq!(out, serde_json::from_str::<Value>(response_body).unwrap());
}

#[tokio::test]
async fn openai_create_missing_required_fields_is_validation() {
    // No server needed — validation fails before any network I/O.
    let client = client_for("http://127.0.0.1:9");
    let err = client
        .chat_completions()
        .create(json!({"messages": []}))
        .await
        .unwrap_err();
    assert!(matches!(err, Error::Validation(_)), "got {err:?}");

    let err = client
        .chat_completions()
        .create(json!({"model": "m"}))
        .await
        .unwrap_err();
    assert!(matches!(err, Error::Validation(_)), "got {err:?}");
}

#[tokio::test]
async fn openai_create_rejects_stream_true() {
    let client = client_for("http://127.0.0.1:9");
    let err = client
        .chat_completions()
        .create(json!({"model": "m", "messages": [], "stream": true}))
        .await
        .unwrap_err();
    match err {
        Error::Validation(d) => assert!(d.message.contains("create_stream")),
        other => panic!("expected Validation, got {other:?}"),
    }
}

// ── conformance §5 item 18 — §4 retry policy on the pass-through route ───────

#[tokio::test]
async fn openai_create_429_retries_then_succeeds() {
    let (listener, base_url) = bind().await;

    let throttle = http_response(
        429,
        "Too Many Requests",
        "Retry-After: 0\r\n",
        r#"{"error":{"message":"rate limited","code":"rate_limit_exceeded"}}"#,
    );
    let ok = http_response(200, "OK", "", r#"{"id":"chatcmpl-2","choices":[]}"#);

    let client = Client::new(
        &Config::builder()
            .api_key("audacity_api_test")
            .base_url(&base_url)
            .max_retries(1)
            .build()
            .unwrap(),
    )
    .unwrap();

    let serve_two = async {
        serve_once(&listener, throttle).await;
        serve_once(&listener, ok).await;
    };
    let (result, _) = tokio::join!(
        async {
            client
                .chat_completions()
                .create(json!({"model": "m", "messages": [{"role": "user", "content": "Q"}]}))
                .await
        },
        serve_two,
    );

    assert_eq!(result.unwrap()["id"], "chatcmpl-2");
}

#[tokio::test]
async fn openai_create_401_is_access_denied_and_not_retried() {
    let (listener, base_url) = bind().await;
    let response = http_response(
        401,
        "Unauthorized",
        "",
        r#"{"error":{"message":"Invalid API key","type":"authentication_error","code":"invalid_api_key"}}"#,
    );

    // max_retries(0) client; a retry attempt would hang on a second accept,
    // so completing at all proves no retry happened.
    let (result, _) = tokio::join!(
        async {
            client_for(&base_url)
                .chat_completions()
                .create(json!({"model": "m", "messages": []}))
                .await
        },
        serve_once(&listener, response),
    );

    let err = result.unwrap_err();
    assert!(matches!(err, Error::AccessDenied(_)), "got {err:?}");
    assert!(!err.is_retryable());
}

// ── conformance §5 item 18 — OpenAI pass-through, streaming ──────────────────

#[tokio::test]
async fn openai_stream_yields_raw_chunks_and_ends_on_done() {
    let (listener, base_url) = bind().await;

    let sse_body = concat!(
        "data: {\"id\":\"c1\",\"choices\":[{\"delta\":{\"role\":\"assistant\"},\"index\":0}],\"future_chunk_field\":1}\n\n",
        "data: {\"id\":\"c1\",\"choices\":[{\"delta\":{\"content\":\"Hello\"},\"index\":0}]}\n\n",
        "data: {\"id\":\"c1\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\",\"index\":0}]}\n\n",
        "data: [DONE]\n\n",
    );
    let response = http_sse_response(sse_body);

    let stream_task = async {
        let stream = client_for(&base_url)
            .chat_completions()
            .create_stream(json!({
                "model": "gpt-5.4-mini",
                "messages": [{"role": "user", "content": "Hi"}],
                "future_request_field": "x",
            }))
            .await?;
        Ok::<_, Error>(drain_stream(stream).await)
    };

    let (result, req) = tokio::join!(stream_task, serve_and_capture(&listener, response));
    let (events, err) = result.unwrap();
    assert!(err.is_none(), "healthy stream must end cleanly: {err:?}");

    // Request: stream flag set, unknown fields preserved.
    let body = request_body(&req);
    assert_eq!(body["stream"], true);
    assert_eq!(body["future_request_field"], "x");

    // Raw chunks in order; [DONE] not yielded; unknown chunk fields intact.
    assert_eq!(events.len(), 3);
    assert_eq!(events[0]["future_chunk_field"], 1);
    assert_eq!(events[1]["choices"][0]["delta"]["content"], "Hello");
    assert_eq!(events[2]["choices"][0]["finish_reason"], "stop");
}

#[tokio::test]
async fn openai_stream_eof_without_done_is_model_stream_error() {
    let (listener, base_url) = bind().await;

    // Well-formed chunked body that simply never sends [DONE].
    let sse_body = "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n";
    let response = http_sse_response(sse_body);

    let stream_task = async {
        let stream = client_for(&base_url)
            .chat_completions()
            .create_stream(json!({"model": "m", "messages": []}))
            .await?;
        Ok::<_, Error>(drain_stream(stream).await)
    };

    let (result, _) = tokio::join!(stream_task, serve_once(&listener, response));
    let (events, err) = result.unwrap();
    assert_eq!(events.len(), 1, "the pre-EOF chunk must still be yielded");
    let err = err.expect("EOF without [DONE] must error");
    assert!(matches!(err, Error::ModelStreamError(_)), "got {err:?}");
    assert!(!err.is_retryable());
}

#[tokio::test]
async fn openai_stream_inline_error_chunk_maps_per_code_table() {
    let (listener, base_url) = bind().await;

    let sse_body = concat!(
        "data: {\"choices\":[{\"delta\":{\"role\":\"assistant\"}}]}\n\n",
        "data: {\"success\":false,\"error\":{\"code\":\"RATE_LIMIT_EXCEEDED\",\"message\":\"slow down\",\"request_id\":\"req-raw-1\"}}\n\n",
    );
    let response = http_sse_response(sse_body);

    let stream_task = async {
        let stream = client_for(&base_url)
            .chat_completions()
            .create_stream(json!({"model": "m", "messages": []}))
            .await?;
        Ok::<_, Error>(drain_stream(stream).await)
    };

    let (result, _) = tokio::join!(stream_task, serve_once(&listener, response));
    let (events, err) = result.unwrap();
    assert_eq!(events.len(), 1);
    match err.expect("in-stream error must abort the stream") {
        Error::Throttling(d) => {
            assert_eq!(d.error_code.as_deref(), Some("RATE_LIMIT_EXCEEDED"));
            assert_eq!(d.request_id.as_deref(), Some("req-raw-1"));
        }
        other => panic!("expected Throttling per the §4 code table, got {other:?}"),
    }
}

// ── conformance §5 item 19 — Anthropic pass-through, non-streaming ───────────

#[tokio::test]
async fn anthropic_create_verbatim_request_and_raw_response() {
    let (listener, base_url) = bind().await;

    let response_body = r#"{"id":"msg_1","type":"message","role":"assistant","content":[{"type":"text","text":"Hello"}],"model":"claude-sonnet-4-5","stop_reason":"end_turn","usage":{"input_tokens":5,"output_tokens":3},"future_response_field":true}"#;
    let response = http_response(200, "OK", "", response_body);

    let params = json!({
        "model": "claude-sonnet-4-5",
        "max_tokens": 1024,
        "messages": [{"role": "user", "content": "Hi"}],
        "system": "be brief",
        "future_request_field": {"beta": true},
    });

    let (result, req) = tokio::join!(
        async { client_for(&base_url).messages().create(params).await },
        serve_and_capture(&listener, response),
    );

    // Path, auth header, and the anthropic-version header (spec §9).
    assert!(req.starts_with("POST /v1/messages HTTP/1.1"));
    let headers_lower = req.to_lowercase();
    assert!(headers_lower.contains("authorization: bearer audacity_api_test"));
    assert!(headers_lower.contains("anthropic-version: 2023-06-01"));

    // Body verbatim — unknown fields preserved.
    let body = request_body(&req);
    assert_eq!(
        body,
        json!({
            "model": "claude-sonnet-4-5",
            "max_tokens": 1024,
            "messages": [{"role": "user", "content": "Hi"}],
            "system": "be brief",
            "future_request_field": {"beta": true},
        })
    );

    // Raw Anthropic-shaped response returned untranslated.
    let out = result.unwrap();
    assert_eq!(out["content"][0]["text"], "Hello");
    assert_eq!(out, serde_json::from_str::<Value>(response_body).unwrap());
}

#[tokio::test]
async fn anthropic_count_tokens_posts_dedicated_path() {
    let (listener, base_url) = bind().await;
    let response = http_response(200, "OK", "", r#"{"input_tokens":42}"#);

    let (result, req) = tokio::join!(
        async {
            client_for(&base_url)
                .messages()
                .count_tokens(json!({
                    "model": "claude-sonnet-4-5",
                    "messages": [{"role": "user", "content": "Hi"}],
                }))
                .await
        },
        serve_and_capture(&listener, response),
    );

    assert!(req.starts_with("POST /v1/messages/count_tokens HTTP/1.1"));
    assert!(req.to_lowercase().contains("anthropic-version: 2023-06-01"));
    assert_eq!(result.unwrap(), json!({"input_tokens": 42}));
}

#[tokio::test]
async fn anthropic_error_envelope_maps_through_taxonomy() {
    // The gateway's Anthropic-shaped error envelope must hit the §4 code
    // table via error.type (rate_limit_error → Throttling).
    let (listener, base_url) = bind().await;
    let response = http_response(
        429,
        "Too Many Requests",
        "",
        r#"{"type":"error","error":{"type":"rate_limit_error","message":"slow down"}}"#,
    );

    let (result, _) = tokio::join!(
        async {
            client_for(&base_url)
                .messages()
                .create(json!({"model": "m", "messages": []}))
                .await
        },
        serve_once(&listener, response),
    );

    match result.unwrap_err() {
        Error::Throttling(d) => assert_eq!(d.message, "slow down"),
        other => panic!("expected Throttling, got {other:?}"),
    }
}

#[tokio::test]
async fn anthropic_unmapped_error_type_falls_back_to_status() {
    // 402 billing_error is not in the code table — HTTP-status fallback must
    // classify it as ServiceQuotaExceeded.
    let (listener, base_url) = bind().await;
    let response = http_response(
        402,
        "Payment Required",
        "",
        r#"{"type":"error","error":{"type":"billing_error","message":"cap reached"}}"#,
    );

    let (result, _) = tokio::join!(
        async {
            client_for(&base_url)
                .messages()
                .create(json!({"model": "m", "messages": []}))
                .await
        },
        serve_once(&listener, response),
    );

    let err = result.unwrap_err();
    assert!(matches!(err, Error::ServiceQuotaExceeded(_)), "got {err:?}");
    assert!(!err.is_retryable());
}

// ── conformance §5 item 19 — Anthropic pass-through, streaming ───────────────

#[tokio::test]
async fn anthropic_stream_yields_raw_events_until_message_stop() {
    let (listener, base_url) = bind().await;

    // `event:` lines must be ignored (the payload's `type` carries the info).
    let sse_body = concat!(
        "event: message_start\n",
        "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"role\":\"assistant\",\"content\":[]}}\n\n",
        "event: content_block_start\n",
        "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
        "event: content_block_delta\n",
        "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"Hello\"},\"future_event_field\":7}\n\n",
        "event: content_block_stop\n",
        "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
        "event: message_stop\n",
        "data: {\"type\":\"message_stop\"}\n\n",
    );
    let response = http_sse_response(sse_body);

    let stream_task = async {
        let stream = client_for(&base_url)
            .messages()
            .create_stream(json!({
                "model": "claude-sonnet-4-5",
                "max_tokens": 100,
                "messages": [{"role": "user", "content": "Hi"}],
            }))
            .await?;
        Ok::<_, Error>(drain_stream(stream).await)
    };

    let (result, req) = tokio::join!(stream_task, serve_and_capture(&listener, response));
    let (events, err) = result.unwrap();
    assert!(err.is_none(), "healthy stream must end cleanly: {err:?}");

    let body = request_body(&req);
    assert_eq!(body["stream"], true);

    let types: Vec<&str> = events.iter().map(|e| e["type"].as_str().unwrap()).collect();
    assert_eq!(
        types,
        vec![
            "message_start",
            "content_block_start",
            "content_block_delta",
            "content_block_stop",
            "message_stop",
        ]
    );
    // Events are raw — unknown fields intact.
    assert_eq!(events[2]["future_event_field"], 7);
    assert_eq!(events[2]["delta"]["text"], "Hello");
}

#[tokio::test]
async fn anthropic_stream_eof_before_message_stop_is_model_stream_error() {
    let (listener, base_url) = bind().await;

    let sse_body = concat!(
        "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\"}}\n\n",
        "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}\n\n",
    );
    let response = http_sse_response(sse_body);

    let stream_task = async {
        let stream = client_for(&base_url)
            .messages()
            .create_stream(json!({"model": "m", "messages": []}))
            .await?;
        Ok::<_, Error>(drain_stream(stream).await)
    };

    let (result, _) = tokio::join!(stream_task, serve_once(&listener, response));
    let (events, err) = result.unwrap();
    assert_eq!(events.len(), 2, "pre-EOF events must still be yielded");
    let err = err.expect("EOF before message_stop must error");
    assert!(matches!(err, Error::ModelStreamError(_)), "got {err:?}");
}

#[tokio::test]
async fn anthropic_stream_error_event_maps_per_code_table() {
    let (listener, base_url) = bind().await;

    let sse_body = concat!(
        "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\"}}\n\n",
        "event: error\n",
        "data: {\"type\":\"error\",\"error\":{\"type\":\"rate_limit_error\",\"message\":\"overloaded\"}}\n\n",
    );
    let response = http_sse_response(sse_body);

    let stream_task = async {
        let stream = client_for(&base_url)
            .messages()
            .create_stream(json!({"model": "m", "messages": []}))
            .await?;
        Ok::<_, Error>(drain_stream(stream).await)
    };

    let (result, _) = tokio::join!(stream_task, serve_once(&listener, response));
    let (events, err) = result.unwrap();
    assert_eq!(
        events.len(),
        1,
        "message_start must be yielded before the error"
    );
    match err.expect("error event must abort the stream") {
        Error::Throttling(d) => assert_eq!(d.message, "overloaded"),
        other => panic!("expected Throttling, got {other:?}"),
    }
}

#[tokio::test]
async fn anthropic_create_missing_required_fields_is_validation() {
    let client = client_for("http://127.0.0.1:9");
    for params in [json!({"messages": []}), json!({"model": "m"})] {
        let err = client.messages().create(params.clone()).await.unwrap_err();
        assert!(
            matches!(err, Error::Validation(_)),
            "params {params}: got {err:?}"
        );
        let err = client.messages().count_tokens(params).await.unwrap_err();
        assert!(matches!(err, Error::Validation(_)), "got {err:?}");
    }
}