audacity-sdk 0.4.0

//! Integration tests against a local mock HTTP server.
//!
//! Tests spin up a real tokio TCP listener, serve canned HTTP responses,
//! and exercise the SDK end-to-end.

use std::sync::{
    atomic::{AtomicUsize, Ordering},
    Arc,
};

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

use audacity_sdk::{
    CachePointBlock, Client, Config, ContentBlock, ConversationRole, ConverseStreamOutput, Error,
    ImageBlock, ImageFormat, ImageSource, MediaResolution, Message, ToolResultBlock,
    ToolResultContent, VideoBlock, VideoFormat, VideoSource,
};

// ── 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) {
    let (mut sock, _) = listener.accept().await.unwrap();
    read_request(&mut sock).await;
    sock.write_all(response.as_bytes()).await.unwrap();
}

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()
}

fn user_message(text: &str) -> Message {
    Message::builder()
        .role(ConversationRole::User)
        .content(ContentBlock::Text(text.into()))
        .build()
        .unwrap()
}

// ── conformance §5.1 — converse happy path ────────────────────────────────────

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

    let body = r#"{"choices":[{"message":{"role":"assistant","content":"Hello world"},"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":3,"total_tokens":8}}"#;
    let response = http_response(200, "OK", "", body);

    let (result, _) = tokio::join!(
        async {
            let client = client_for(&base_url);
            client
                .converse()
                .model_id("gpt-5.4-mini")
                .messages(user_message("Hi"))
                .send()
                .await
        },
        serve_once(&listener, response),
    );

    let out = result.unwrap();
    let msg = out.output().unwrap().as_message().unwrap();
    assert_eq!(msg.content[0].as_text().unwrap(), "Hello world");
    assert_eq!(*out.stop_reason(), audacity_sdk::StopReason::EndTurn);
    assert_eq!(out.usage().input_tokens, 5);
    assert_eq!(out.usage().output_tokens, 3);
    assert_eq!(out.usage().total_tokens, 8);
    assert!(out.metrics().latency_ms < 5000);
}

// ── conformance §5.7 — defensive unwrap rule ─────────────────────────────────

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

    let body = r#"{"data":{"choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{}}}"#;
    let response = http_response(200, "OK", "", body);

    let (result, _) = tokio::join!(
        async {
            client_for(&base_url)
                .converse()
                .model_id("m")
                .messages(user_message("Q"))
                .send()
                .await
        },
        serve_once(&listener, response),
    );

    assert!(result.unwrap().output().is_some());
}

// ── conformance §5.5 — error shape A ─────────────────────────────────────────

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

    let body = r#"{"error":{"message":"Invalid API key","type":"authentication_error","code":"invalid_api_key"}}"#;
    let response = http_response(401, "Unauthorized", "", body);

    let (result, _) = tokio::join!(
        async {
            client_for(&base_url)
                .converse()
                .model_id("m")
                .messages(user_message("Q"))
                .send()
                .await
        },
        serve_once(&listener, response),
    );

    let err = result.unwrap_err();
    // §4: access-denied errors are not retryable — checked on the real enum.
    assert!(!err.is_retryable());
    match err {
        Error::AccessDenied(d) => {
            assert_eq!(d.error_code.as_deref(), Some("invalid_api_key"));
        }
        other => panic!("expected AccessDenied, got {other:?}"),
    }
}

// ── conformance §5.5 — error shape B ─────────────────────────────────────────

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

    let body = r#"{"success":false,"error":{"code":"MODEL_NOT_ALLOWED","message":"Not allowed","request_id":"req-abc","details":{}}}"#;
    let response = http_response(403, "Forbidden", "", body);

    let (result, _) = tokio::join!(
        async {
            client_for(&base_url)
                .converse()
                .model_id("m")
                .messages(user_message("Q"))
                .send()
                .await
        },
        serve_once(&listener, response),
    );

    assert!(matches!(result, Err(Error::AccessDenied(_))));
    if let Err(Error::AccessDenied(d)) = result {
        assert_eq!(d.request_id.as_deref(), Some("req-abc"));
    }
}

// ── conformance §5.6 — 429 with Retry-After retries then succeeds ─────────────

#[tokio::test]
async fn retry_after_429_then_success() {
    converse_with_429_then_ok(0).await;
}

/// Serve a 429 (with the given `Retry-After`) then a 200, drive one `converse`
/// with one retry allowed, assert it succeeds, and return the elapsed time.
async fn converse_with_429_then_ok(retry_after_secs: u64) -> std::time::Duration {
    let (listener, base_url) = bind().await;

    let throttle_body = r#"{"error":{"message":"rate limited","code":"rate_limit_exceeded"}}"#;
    let throttle_resp = http_response(
        429,
        "Too Many Requests",
        &format!("Retry-After: {retry_after_secs}\r\n"),
        throttle_body,
    );
    let ok_body = r#"{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}],"usage":{}}"#;
    let ok_resp = http_response(200, "OK", "", ok_body);

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

    let serve_two = async {
        // first request → 429
        serve_once(&listener, throttle_resp).await;
        // second request → 200
        serve_once(&listener, ok_resp).await;
    };

    let started = std::time::Instant::now();
    let (result, _) = tokio::join!(
        client
            .converse()
            .model_id("m")
            .messages(user_message("Q"))
            .send(),
        serve_two,
    );

    assert!(result.is_ok(), "expected Ok but got: {:?}", result.err());
    started.elapsed()
}

// ── conformance §5.6 — 401 does NOT retry ────────────────────────────────────

#[tokio::test]
async fn no_retry_on_401() {
    let (listener, base_url) = bind().await;
    let attempt_count = Arc::new(AtomicUsize::new(0));
    let attempt_count2 = attempt_count.clone();

    let body = r#"{"error":{"message":"unauth","code":"invalid_api_key"}}"#;

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

    // Serve only ONE response — if the SDK retries it would hang.
    let serve_task = async move {
        serve_once(&listener, http_response(401, "Unauthorized", "", body)).await;
        attempt_count2.fetch_add(1, Ordering::SeqCst);
    };

    let (result, _) = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        tokio::join!(
            client
                .converse()
                .model_id("m")
                .messages(user_message("Q"))
                .send(),
            serve_task,
        )
    })
    .await
    .unwrap();

    assert!(matches!(result, Err(Error::AccessDenied(_))));
    assert_eq!(attempt_count.load(Ordering::SeqCst), 1);
}

// ── conformance §5.6 — BUDGET_EXCEEDED does NOT retry ────────────────────────

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

    let body = r#"{"success":false,"error":{"code":"BUDGET_EXCEEDED","message":"over budget"}}"#;

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

    let serve_task = serve_once(&listener, http_response(429, "Too Many Requests", "", body));

    let (result, _) = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        tokio::join!(
            client
                .converse()
                .model_id("m")
                .messages(user_message("Q"))
                .send(),
            serve_task,
        )
    })
    .await
    .unwrap();

    assert!(matches!(result, Err(Error::ServiceQuotaExceeded(_))));
}

// ── conformance §5.9 — missing API key ───────────────────────────────────────

#[test]
fn missing_api_key_fails_fast() {
    // Unset env var just in case. Safe despite the parallel test binary:
    // every other test passes api_key explicitly, so nothing else reads env.
    std::env::remove_var("AUDACITY_API_KEY");
    let result = Config::builder().build();
    assert!(matches!(result, Err(Error::MissingApiKey)));
}

// ── conformance §5.10 — User-Agent header ────────────────────────────────────

#[tokio::test]
async fn user_agent_header_sent() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    let base_url = format!("http://127.0.0.1:{port}");

    let body = r#"{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}],"usage":{}}"#;
    let response = http_response(200, "OK", "", body);

    let capture_ua = async {
        let (mut sock, _) = listener.accept().await.unwrap();
        let req_str = read_request(&mut sock).await;
        sock.write_all(response.as_bytes()).await.unwrap();
        req_str
    };

    let (result, req_str) = tokio::join!(
        async {
            client_for(&base_url)
                .converse()
                .model_id("m")
                .messages(user_message("Q"))
                .send()
                .await
        },
        capture_ua,
    );

    result.unwrap();
    let expected_ua = concat!("audacity-sdk-rust/", env!("CARGO_PKG_VERSION"));
    assert!(
        req_str.contains(expected_ua),
        "User-Agent {expected_ua} not found in: {req_str}"
    );
}

// ── conformance §5.2 — streaming happy path ──────────────────────────────────

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

    let sse_body = concat!(
        "data: {\"choices\":[{\"delta\":{\"role\":\"assistant\"},\"index\":0}]}\n\n",
        "data: {\"choices\":[{\"delta\":{\"content\":\"Hello\"},\"index\":0}]}\n\n",
        "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\",\"index\":0}]}\n\n",
        "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":2,\"total_tokens\":3}}\n\n",
        "data: [DONE]\n\n",
    );
    let response = http_sse_response(sse_body);

    let serve_task = serve_once(&listener, response);
    let stream_task = async {
        let mut out = client_for(&base_url)
            .converse_stream()
            .model_id("gpt-5.4-mini")
            .messages(user_message("Hi"))
            .send()
            .await?;

        let mut events: Vec<String> = Vec::new();
        while let Some(event) = out.stream.recv().await? {
            match &event {
                ConverseStreamOutput::MessageStart(_) => events.push("MessageStart".into()),
                ConverseStreamOutput::ContentBlockDelta(_) => {
                    events.push("ContentBlockDelta".into())
                }
                ConverseStreamOutput::ContentBlockStop(_) => events.push("ContentBlockStop".into()),
                ConverseStreamOutput::MessageStop(_) => events.push("MessageStop".into()),
                ConverseStreamOutput::Metadata(_) => events.push("Metadata".into()),
                _ => {}
            }
        }
        Ok::<_, Error>(events)
    };

    let (events, _) = tokio::join!(stream_task, serve_task);
    let events = events.unwrap();

    // Verify ordering: MessageStart → ContentBlockDelta → ContentBlockStop → MessageStop → Metadata
    assert_eq!(
        events,
        vec![
            "MessageStart",
            "ContentBlockDelta",
            "ContentBlockStop",
            "MessageStop",
            "Metadata",
        ]
    );
}

// ── conformance §5.3 — tool round-trip (streaming) ───────────────────────────

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

    let sse_body = concat!(
        "data: {\"choices\":[{\"delta\":{\"role\":\"assistant\"}}]}\n\n",
        "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"get_weather\",\"arguments\":\"\"}}]}}]}\n\n",
        "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"city\\\":\\\"NYC\\\"}\"}}]}}]}\n\n",
        "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n",
        "data: {\"choices\":[],\"usage\":{}}\n\n",
        "data: [DONE]\n\n",
    );
    let response = http_sse_response(sse_body);

    let stream_task = async {
        let mut out = client_for(&base_url)
            .converse_stream()
            .model_id("m")
            .messages(user_message("weather?"))
            .send()
            .await?;

        let mut has_start = false;
        let mut has_delta = false;
        let mut stop_reason = None;

        while let Some(event) = out.stream.recv().await? {
            match event {
                ConverseStreamOutput::ContentBlockStart(_) => has_start = true,
                ConverseStreamOutput::ContentBlockDelta(_) => has_delta = true,
                ConverseStreamOutput::MessageStop(e) => stop_reason = Some(e.stop_reason),
                _ => {}
            }
        }
        Ok::<_, Error>((has_start, has_delta, stop_reason))
    };

    let (result, _) = tokio::join!(stream_task, serve_once(&listener, response));
    let (has_start, has_delta, stop_reason) = result.unwrap();

    assert!(has_start, "expected ContentBlockStart for tool call");
    assert!(has_delta, "expected ContentBlockDelta for tool arguments");
    assert_eq!(stop_reason, Some(audacity_sdk::StopReason::ToolUse));
}

// ── chunked-transfer helpers for staged SSE writes ───────────────────────────

const SSE_HEADERS: &str =
    "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n";

fn sse_chunk(data: &str) -> Vec<u8> {
    format!("{:x}\r\n{data}\r\n", data.len()).into_bytes()
}

const SSE_TERMINAL: &[u8] = b"0\r\n\r\n";

// ── spec §3 — connection drop before [DONE] → ModelStreamError ───────────────

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

    let serve_task = async {
        let (mut sock, _) = listener.accept().await.unwrap();
        read_request(&mut sock).await;
        sock.write_all(SSE_HEADERS.as_bytes()).await.unwrap();
        sock.write_all(&sse_chunk(
            "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n",
        ))
        .await
        .unwrap();
        sock.flush().await.unwrap();
        // Drop the socket without the chunked terminator or [DONE].
        drop(sock);
    };

    let stream_task = async {
        let mut out = client_for(&base_url)
            .converse_stream()
            .model_id("m")
            .messages(user_message("Q"))
            .send()
            .await?;

        let mut saw_delta = false;
        loop {
            match out.stream.recv().await {
                Ok(Some(ConverseStreamOutput::ContentBlockDelta(_))) => saw_delta = true,
                Ok(Some(_)) => {}
                Ok(None) => return Ok::<_, Error>((saw_delta, None)),
                Err(e) => return Ok((saw_delta, Some(e))),
            }
        }
    };

    let (result, _) = tokio::join!(stream_task, serve_task);
    let (saw_delta, err) = result.unwrap();
    assert!(saw_delta, "expected the pre-drop delta to arrive");
    let err = err.expect("expected a stream error after the connection drop");
    assert!(
        matches!(err, Error::ModelStreamError(_)),
        "expected ModelStreamError, got {err:?}"
    );
    assert!(!err.is_retryable());
}

// ── spec §1/§4 — in-stream error chunk maps per the §4 code table ────────────

#[tokio::test]
async fn 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-inline-1\"}}\n\n",
    );
    let response = http_sse_response(sse_body);

    let stream_task = async {
        let mut out = client_for(&base_url)
            .converse_stream()
            .model_id("m")
            .messages(user_message("Q"))
            .send()
            .await?;

        loop {
            match out.stream.recv().await {
                Ok(Some(_)) => {}
                Ok(None) => return Ok::<_, Error>(None),
                Err(e) => return Ok(Some(e)),
            }
        }
    };

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

// ── spec §4 — Retry-After lower-bounds the backoff sleep ─────────────────────

#[tokio::test]
async fn retry_after_lower_bounds_backoff() {
    let elapsed = converse_with_429_then_ok(1).await;
    assert!(
        elapsed >= std::time::Duration::from_secs(1),
        "backoff must sleep at least Retry-After (1s), only slept {elapsed:?}"
    );
}

// ── spec §1 — the timeout must NOT bound the SSE body read ───────────────────

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

    let client = Client::new(
        &Config::builder()
            .api_key("audacity_api_test")
            .base_url(&base_url)
            .timeout(std::time::Duration::from_secs(1))
            .max_retries(0)
            .build()
            .unwrap(),
    )
    .unwrap();

    let serve_task = async {
        let (mut sock, _) = listener.accept().await.unwrap();
        read_request(&mut sock).await;
        sock.write_all(SSE_HEADERS.as_bytes()).await.unwrap();
        sock.write_all(&sse_chunk(
            "data: {\"choices\":[{\"delta\":{\"content\":\"first\"}}]}\n\n",
        ))
        .await
        .unwrap();
        sock.flush().await.unwrap();
        // Stall well past the 1s configured timeout mid-body.
        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        sock.write_all(&sse_chunk(concat!(
            "data: {\"choices\":[{\"delta\":{\"content\":\"second\"},\"finish_reason\":null}]}\n\n",
            "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
            "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":2,\"total_tokens\":3}}\n\n",
            "data: [DONE]\n\n",
        )))
        .await
        .unwrap();
        sock.write_all(SSE_TERMINAL).await.unwrap();
    };

    let stream_task = async {
        let mut out = client
            .converse_stream()
            .model_id("m")
            .messages(user_message("Q"))
            .send()
            .await?;

        let mut deltas = 0;
        let mut saw_stop = false;
        let mut saw_metadata = false;
        while let Some(event) = out.stream.recv().await? {
            match event {
                ConverseStreamOutput::ContentBlockDelta(_) => deltas += 1,
                ConverseStreamOutput::MessageStop(_) => saw_stop = true,
                ConverseStreamOutput::Metadata(_) => saw_metadata = true,
                _ => {}
            }
        }
        Ok::<_, Error>((deltas, saw_stop, saw_metadata))
    };

    let (result, _) = tokio::join!(stream_task, serve_task);
    let (deltas, saw_stop, saw_metadata) =
        result.expect("stream must survive past the configured timeout");
    assert_eq!(deltas, 2);
    assert!(saw_stop);
    assert!(saw_metadata);
}

// ── conformance §5.13 — video blocks ─────────────────────────────────────────

use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use base64::Engine as _;

fn message(role: ConversationRole, blocks: Vec<ContentBlock>) -> Message {
    Message {
        role,
        content: blocks,
    }
}

fn video_block(format: VideoFormat, bytes: &[u8]) -> ContentBlock {
    ContentBlock::Video(VideoBlock {
        format,
        source: VideoSource::Bytes(bytes.to_vec()),
    })
}

fn video_uri_block(format: VideoFormat, uri: &str) -> ContentBlock {
    ContentBlock::Video(VideoBlock {
        format,
        source: VideoSource::Uri(uri.into()),
    })
}

/// Drive one `converse` call against a capture server and return the JSON
/// request body the SDK put on the wire.
async fn capture_request_body(messages: Vec<Message>) -> serde_json::Value {
    let (listener, base_url) = bind().await;

    let ok_body = r#"{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}],"usage":{}}"#;
    let response = http_response(200, "OK", "", ok_body);

    let capture = async {
        let (mut sock, _) = listener.accept().await.unwrap();
        let req_str = read_request(&mut sock).await;
        sock.write_all(response.as_bytes()).await.unwrap();
        req_str
    };

    let send = async {
        let mut builder = client_for(&base_url).converse().model_id("gemini-2.5-flash");
        for msg in messages {
            builder = builder.messages(msg);
        }
        builder.send().await
    };

    let (result, req_str) = tokio::join!(send, capture);
    result.unwrap();

    let body_start = req_str.find("\r\n\r\n").unwrap() + 4;
    serde_json::from_str(&req_str[body_start..]).unwrap()
}

#[tokio::test]
async fn video_bytes_to_file_part() {
    let body = capture_request_body(vec![message(
        ConversationRole::User,
        vec![
            ContentBlock::Text("What happens in this clip?".into()),
            video_block(VideoFormat::Mp4, b"vidbytes"),
        ],
    )])
    .await;

    let parts = body["messages"][0]["content"].as_array().unwrap();
    assert_eq!(
        parts[0],
        serde_json::json!({ "type": "text", "text": "What happens in this clip?" })
    );
    assert_eq!(parts[1]["type"], "file");
    assert_eq!(parts[1]["file"]["format"], "video/mp4");
    let file_data = parts[1]["file"]["file_data"].as_str().unwrap();
    let b64 = file_data.strip_prefix("data:video/mp4;base64,").unwrap();
    assert_eq!(BASE64_STANDARD.decode(b64).unwrap(), b"vidbytes");
}

#[tokio::test]
async fn video_block_order_preserved() {
    let body = capture_request_body(vec![message(
        ConversationRole::User,
        vec![
            ContentBlock::Text("before".into()),
            video_block(VideoFormat::Webm, b"v"),
            ContentBlock::Text("after".into()),
        ],
    )])
    .await;

    let parts = body["messages"][0]["content"].as_array().unwrap();
    let types: Vec<&str> = parts.iter().map(|p| p["type"].as_str().unwrap()).collect();
    assert_eq!(types, ["text", "file", "text"]);
    assert_eq!(parts[0]["text"], "before");
    assert_eq!(parts[2]["text"], "after");
}

#[tokio::test]
async fn text_only_turn_still_plain_string() {
    let body = capture_request_body(vec![message(
        ConversationRole::User,
        vec![
            ContentBlock::Text("a".into()),
            ContentBlock::Text("b".into()),
        ],
    )])
    .await;

    assert_eq!(body["messages"][0]["content"], "a\nb");
}

#[tokio::test]
async fn video_in_assistant_turn_ignored() {
    let body = capture_request_body(vec![
        message(
            ConversationRole::User,
            vec![ContentBlock::Text("Hi".into())],
        ),
        message(
            ConversationRole::Assistant,
            vec![
                ContentBlock::Text("Sure".into()),
                video_block(VideoFormat::Mp4, b"x"),
            ],
        ),
        message(
            ConversationRole::User,
            vec![ContentBlock::Text("Go on".into())],
        ),
    ])
    .await;

    let msgs = body["messages"].as_array().unwrap();
    assert_eq!(msgs.len(), 3);
    assert_eq!(msgs[1]["role"], "assistant");
    // assistant content stays a plain string; the video block is dropped
    assert_eq!(msgs[1]["content"], "Sure");
    assert!(!body.to_string().contains("file_data"));
}

#[tokio::test]
async fn video_emitted_after_tool_result_messages() {
    let body = capture_request_body(vec![message(
        ConversationRole::User,
        vec![
            ContentBlock::ToolResult(ToolResultBlock {
                tool_use_id: "call_1".into(),
                content: vec![ToolResultContent::Text("42".into())],
                status: None,
            }),
            ContentBlock::Text("And this clip?".into()),
            video_block(VideoFormat::Mp4, b"v"),
        ],
    )])
    .await;

    let msgs = body["messages"].as_array().unwrap();
    assert_eq!(msgs[0]["role"], "tool");
    assert_eq!(msgs[0]["tool_call_id"], "call_1");
    assert_eq!(msgs[1]["role"], "user");
    let types: Vec<&str> = msgs[1]["content"]
        .as_array()
        .unwrap()
        .iter()
        .map(|p| p["type"].as_str().unwrap())
        .collect();
    assert_eq!(types, ["text", "file"]);
}

#[tokio::test]
async fn mixed_image_and_video_turn() {
    let body = capture_request_body(vec![message(
        ConversationRole::User,
        vec![
            ContentBlock::Image(ImageBlock {
                format: ImageFormat::Png,
                source: ImageSource::Bytes(b"img".to_vec()),
            }),
            video_block(VideoFormat::Mp4, b"vid"),
            ContentBlock::Text("Compare these".into()),
        ],
    )])
    .await;

    let parts = body["messages"][0]["content"].as_array().unwrap();
    let types: Vec<&str> = parts.iter().map(|p| p["type"].as_str().unwrap()).collect();
    assert_eq!(types, ["image_url", "file", "text"]);
    assert!(parts[0]["image_url"]["url"]
        .as_str()
        .unwrap()
        .starts_with("data:image/png;base64,"));
    assert_eq!(parts[1]["file"]["format"], "video/mp4");
}

#[tokio::test]
async fn video_format_mime_table() {
    let cases = [
        (VideoFormat::Mp4, "video/mp4"),
        (VideoFormat::Mov, "video/mov"),
        (VideoFormat::ThreeGp, "video/3gpp"),
        (VideoFormat::Mkv, "video/x-matroska"),
    ];
    for (format, mime) in cases {
        let body = capture_request_body(vec![message(
            ConversationRole::User,
            vec![video_block(format, b"v")],
        )])
        .await;
        let part = &body["messages"][0]["content"][0];
        assert_eq!(part["file"]["format"], mime, "format {format:?}");
        assert!(
            part["file"]["file_data"]
                .as_str()
                .unwrap()
                .starts_with(&format!("data:{mime};base64,")),
            "data URL media type for {format:?}"
        );
    }
}

// ── conformance §5.14 — video uri source mapping ─────────────────────────────

#[tokio::test]
async fn video_uri_to_file_id_part() {
    let uri = "audacity://files/11111111-2222-3333-4444-555555555555";
    let body = capture_request_body(vec![message(
        ConversationRole::User,
        vec![
            ContentBlock::Text("What happens in this clip?".into()),
            video_uri_block(VideoFormat::Mp4, uri),
        ],
    )])
    .await;

    let parts = body["messages"][0]["content"].as_array().unwrap();
    assert_eq!(parts[1]["type"], "file");
    // URI passed through verbatim; MIME from the §3 format table
    assert_eq!(parts[1]["file"]["file_id"], uri);
    assert_eq!(parts[1]["file"]["format"], "video/mp4");
    // no base64 payload for uri sources
    assert!(parts[1]["file"].get("file_data").is_none());
    assert!(!body.to_string().contains("file_data"));
}

#[tokio::test]
async fn video_uri_in_assistant_turn_ignored() {
    let body = capture_request_body(vec![
        message(
            ConversationRole::User,
            vec![ContentBlock::Text("Hi".into())],
        ),
        message(
            ConversationRole::Assistant,
            vec![
                ContentBlock::Text("Sure".into()),
                video_uri_block(VideoFormat::Mp4, "audacity://files/abc"),
            ],
        ),
        message(
            ConversationRole::User,
            vec![ContentBlock::Text("Go on".into())],
        ),
    ])
    .await;

    let msgs = body["messages"].as_array().unwrap();
    assert_eq!(msgs.len(), 3);
    assert_eq!(msgs[1]["role"], "assistant");
    // assistant content stays a plain string; the video block is dropped
    assert_eq!(msgs[1]["content"], "Sure");
    assert!(!body.to_string().contains("file_id"));
}

#[tokio::test]
async fn mixed_bytes_and_uri_video_turn() {
    let uri = "audacity://files/deadbeef";
    let body = capture_request_body(vec![message(
        ConversationRole::User,
        vec![
            video_block(VideoFormat::Mp4, b"inline"),
            video_uri_block(VideoFormat::Webm, uri),
            ContentBlock::Text("Compare these".into()),
        ],
    )])
    .await;

    let parts = body["messages"][0]["content"].as_array().unwrap();
    let types: Vec<&str> = parts.iter().map(|p| p["type"].as_str().unwrap()).collect();
    assert_eq!(types, ["file", "file", "text"]);
    // bytes source → file_data data URL, no file_id
    assert!(parts[0]["file"]["file_data"]
        .as_str()
        .unwrap()
        .starts_with("data:video/mp4;base64,"));
    assert!(parts[0]["file"].get("file_id").is_none());
    // uri source → verbatim file_id, no file_data
    assert_eq!(parts[1]["file"]["file_id"], uri);
    assert_eq!(parts[1]["file"]["format"], "video/webm");
    assert!(parts[1]["file"].get("file_data").is_none());
}

#[tokio::test]
async fn cache_point_after_video_marks_it() {
    let body = capture_request_body(vec![message(
        ConversationRole::User,
        vec![
            video_block(VideoFormat::Mp4, b"v"),
            ContentBlock::CachePoint(CachePointBlock::new()),
            ContentBlock::Text("What is this?".into()),
        ],
    )])
    .await;

    let parts = body["messages"][0]["content"].as_array().unwrap();
    assert_eq!(parts[0]["type"], "file");
    assert_eq!(
        parts[0]["cache_control"],
        serde_json::json!({ "type": "ephemeral" })
    );
    assert!(parts[1].get("cache_control").is_none());
}

// ── conformance §5.15 — mediaResolution request field ────────────────────────

#[tokio::test]
async fn media_resolution_serialized_on_wire() {
    let (listener, base_url) = bind().await;
    let ok_body = r#"{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}],"usage":{}}"#;
    let response = http_response(200, "OK", "", ok_body);

    let capture = async {
        let (mut sock, _) = listener.accept().await.unwrap();
        let req_str = read_request(&mut sock).await;
        sock.write_all(response.as_bytes()).await.unwrap();
        req_str
    };

    let (result, req_str) = tokio::join!(
        async {
            client_for(&base_url)
                .converse()
                .model_id("gemini-2.5-flash")
                .media_resolution(MediaResolution::Low)
                .messages(user_message("Q"))
                .send()
                .await
        },
        capture,
    );
    result.unwrap();

    let body: serde_json::Value = serde_json::from_str(request_body(&req_str)).unwrap();
    assert_eq!(body["media_resolution"], "low");
}

// ── spec §6 — file upload helper (resumable protocol) ────────────────────────

/// The §6 upload ticket body, with `upload_url` pointing back at the mock server.
fn upload_ticket_body(base_url: &str) -> String {
    format!(
        r#"{{"file_id":"file-123","upload_url":"{base_url}/upload/file-123","uri":"audacity://files/file-123","expires_at":"2026-07-07T12:00:00Z"}}"#
    )
}

/// A response that closes its connection — every request in the multi-request
/// resumable flow must arrive on a fresh socket (the mock serves one request
/// per socket).
fn closed_response(status: u16, status_text: &str, headers_extra: &str, body: &str) -> String {
    http_response(
        status,
        status_text,
        &format!("Connection: close\r\n{headers_extra}"),
        body,
    )
}

/// Spec §6 step 2a success: the session URI arrives in the Location header.
fn resumable_init_response(session_url: &str) -> String {
    closed_response(201, "Created", &format!("Location: {session_url}\r\n"), "")
}

/// `308 Resume Incomplete` confirming bytes `0..=confirmed_end` are persisted.
fn resume_incomplete_response(confirmed_end: usize) -> String {
    closed_response(
        308,
        "Resume Incomplete",
        &format!("Range: bytes=0-{confirmed_end}\r\n"),
        "",
    )
}

/// Serve a fixed sequence of responses, one connection per request, returning
/// the captured requests.
async fn serve_upload_sequence(listener: &TcpListener, responses: Vec<String>) -> Vec<String> {
    let mut reqs = Vec::new();
    for response in responses {
        let (mut sock, _) = listener.accept().await.unwrap();
        reqs.push(read_request(&mut sock).await);
        sock.write_all(response.as_bytes()).await.unwrap();
    }
    reqs
}

fn request_body(req: &str) -> &str {
    &req[req.find("\r\n\r\n").unwrap() + 4..]
}

/// ASCII payload so captured requests round-trip losslessly through String.
fn ascii_payload(len: usize) -> Vec<u8> {
    (0..len).map(|i| b'a' + (i % 26) as u8).collect()
}

/// Chunk size used by the resumable tests: 512 KiB (2 × the 256 KiB unit).
const TEST_CHUNK: usize = 524_288;

async fn upload(base_url: &str, data: Vec<u8>) -> Result<audacity_sdk::UploadFileOutput, Error> {
    client_for(base_url)
        .upload_file()
        .data(data)
        .content_type("video/mp4")
        .chunk_size(TEST_CHUNK)
        .send()
        .await
}

#[tokio::test]
async fn upload_file_small_file_single_request() {
    let (listener, base_url) = bind().await;
    let video_bytes = b"raw video bytes".to_vec(); // 15 bytes < one chunk

    let session_url = format!("{base_url}/session/abc");
    let responses = vec![
        closed_response(200, "OK", "", &upload_ticket_body(&base_url)),
        resumable_init_response(&session_url),
        closed_response(200, "OK", "", ""),
    ];

    let (result, reqs) = tokio::time::timeout(std::time::Duration::from_secs(10), async {
        tokio::join!(
            upload(&base_url, video_bytes.clone()),
            serve_upload_sequence(&listener, responses),
        )
    })
    .await
    .unwrap();
    let out = result.unwrap();

    // step-1 fields returned as-is
    assert_eq!(out.file_id, "file-123");
    assert_eq!(out.upload_url, format!("{base_url}/upload/file-123"));
    assert_eq!(out.uri, "audacity://files/file-123");
    assert_eq!(out.expires_at, "2026-07-07T12:00:00Z");

    // step-1 POST: right path, auth header present, size computed from data
    let post_req = &reqs[0];
    assert!(post_req.starts_with("POST /v1/files "), "{post_req}");
    assert!(
        post_req.contains("Bearer audacity_api_test"),
        "auth header missing on POST: {post_req}"
    );
    let post_body: serde_json::Value = serde_json::from_str(request_body(post_req)).unwrap();
    assert_eq!(post_body["content_type"], "video/mp4");
    assert_eq!(post_body["size_bytes"], 15);

    // initiation POST: presigned URL, resumable headers, NO auth
    let init_req = &reqs[1];
    assert!(init_req.starts_with("POST /upload/file-123 "), "{init_req}");
    let init_lower = init_req.to_lowercase();
    assert!(init_lower.contains("x-goog-resumable: start"), "{init_req}");
    assert!(init_lower.contains("content-type: video/mp4"), "{init_req}");
    assert!(
        !init_lower.contains("authorization"),
        "auth header must be absent on initiation: {init_req}"
    );

    // a file smaller than one chunk is a single PUT covering all bytes
    assert_eq!(reqs.len(), 3);
    let put_req = &reqs[2];
    assert!(put_req.starts_with("PUT /session/abc "), "{put_req}");
    assert!(
        !put_req.to_lowercase().contains("authorization"),
        "auth header must be absent on PUT: {put_req}"
    );
    assert!(
        put_req.to_lowercase().contains("content-range: bytes 0-14/15"),
        "Content-Range missing on PUT: {put_req}"
    );
    assert_eq!(request_body(put_req).as_bytes(), &video_bytes[..]);
}

#[tokio::test]
async fn upload_file_multi_chunk_happy_path() {
    let (listener, base_url) = bind().await;
    let total = 2 * TEST_CHUNK + 1_000;
    let data = ascii_payload(total);

    let session_url = format!("{base_url}/session/abc");
    let responses = vec![
        closed_response(200, "OK", "", &upload_ticket_body(&base_url)),
        resumable_init_response(&session_url),
        resume_incomplete_response(TEST_CHUNK - 1),
        resume_incomplete_response(2 * TEST_CHUNK - 1),
        closed_response(200, "OK", "", ""),
    ];

    let (result, reqs) = tokio::time::timeout(std::time::Duration::from_secs(15), async {
        tokio::join!(
            upload(&base_url, data.clone()),
            serve_upload_sequence(&listener, responses),
        )
    })
    .await
    .unwrap();
    let out = result.unwrap();
    assert_eq!(out.uri, "audacity://files/file-123");

    // initiation headers (incl. x-goog-resumable: start) and no auth
    let init_lower = reqs[1].to_lowercase();
    assert!(init_lower.contains("x-goog-resumable: start"), "{}", reqs[1]);
    assert!(init_lower.contains("content-type: video/mp4"), "{}", reqs[1]);
    assert!(!init_lower.contains("authorization"), "{}", reqs[1]);

    // each chunk PUT carries the right Content-Range; 308 → next, final → 200
    let expected_ranges = [
        format!("bytes 0-{}/{total}", TEST_CHUNK - 1),
        format!("bytes {TEST_CHUNK}-{}/{total}", 2 * TEST_CHUNK - 1),
        format!("bytes {}-{}/{total}", 2 * TEST_CHUNK, total - 1),
    ];
    assert_eq!(reqs.len(), 2 + expected_ranges.len());
    let mut reassembled: Vec<u8> = Vec::new();
    for (req, range) in reqs[2..].iter().zip(&expected_ranges) {
        assert!(req.starts_with("PUT /session/abc "), "{req}");
        assert!(
            req.to_lowercase().contains(&format!("content-range: {range}")),
            "expected Content-Range {range} in: {req}"
        );
        reassembled.extend_from_slice(request_body(req).as_bytes());
    }
    assert_eq!(reassembled, data, "reassembled chunks must equal the input");
}

#[tokio::test]
async fn upload_file_resumes_after_transient_500_via_offset_query() {
    let (listener, base_url) = bind().await;
    let total = 2 * TEST_CHUNK;
    let data = ascii_payload(total);

    let session_url = format!("{base_url}/session/abc");
    let responses = vec![
        closed_response(200, "OK", "", &upload_ticket_body(&base_url)),
        resumable_init_response(&session_url),
        // chunk 1 confirmed
        resume_incomplete_response(TEST_CHUNK - 1),
        // chunk 2 dies server-side
        closed_response(500, "Internal Server Error", "", "boom"),
        // offset query → only chunk 1 persisted
        resume_incomplete_response(TEST_CHUNK - 1),
        // resumed chunk 2 lands
        closed_response(200, "OK", "", ""),
    ];

    let (result, reqs) = tokio::time::timeout(std::time::Duration::from_secs(30), async {
        tokio::join!(
            upload(&base_url, data.clone()),
            serve_upload_sequence(&listener, responses),
        )
    })
    .await
    .unwrap();
    result.unwrap();

    // the recovery request is a bare offset query…
    let query = &reqs[4];
    assert!(
        query
            .to_lowercase()
            .contains(&format!("content-range: bytes */{total}")),
        "expected offset query Content-Range in: {query}"
    );
    assert_eq!(request_body(query), "", "offset query must have an empty body");

    // …and the upload resumes from the confirmed byte (N+1) to completion
    let resumed = &reqs[5];
    let range = format!("bytes {TEST_CHUNK}-{}/{total}", total - 1);
    assert!(
        resumed
            .to_lowercase()
            .contains(&format!("content-range: {range}")),
        "expected resumption Content-Range {range} in: {resumed}"
    );
    assert_eq!(request_body(resumed).as_bytes(), &data[TEST_CHUNK..]);
}

#[tokio::test]
async fn upload_file_persistent_503_exhausts_recovery_budget() {
    let (listener, base_url) = bind().await;
    let data = b"tiny".to_vec();

    let session_url = format!("{base_url}/session/abc");
    let unavailable = closed_response(503, "Service Unavailable", "", "");
    let mut responses = vec![
        closed_response(200, "OK", "", &upload_ticket_body(&base_url)),
        resumable_init_response(&session_url),
    ];
    // The initial chunk PUT plus 5 offset-query recovery attempts, all 503.
    responses.extend(std::iter::repeat(unavailable).take(6));

    // Generous bound: the 5 jittered backoffs sleep at most 1+2+4+8+16 = 31s.
    let (result, reqs) = tokio::time::timeout(std::time::Duration::from_secs(120), async {
        tokio::join!(
            upload(&base_url, data),
            serve_upload_sequence(&listener, responses),
        )
    })
    .await
    .unwrap();

    let err = result.unwrap_err();
    assert!(
        matches!(err, Error::ServiceUnavailable(_)),
        "expected ServiceUnavailable after exhaustion, got {err:?}"
    );
    // ticket + initiation + failing chunk + 5 recovery attempts — then give up.
    assert_eq!(reqs.len(), 8);
    let session_puts = reqs
        .iter()
        .filter(|r| r.starts_with("PUT /session/abc "))
        .count();
    assert_eq!(
        session_puts, 6,
        "expected the failing chunk plus exactly 5 recovery attempts"
    );
}

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

    let session_url = format!("{base_url}/session/abc");
    let responses = vec![
        closed_response(200, "OK", "", &upload_ticket_body(&base_url)),
        resumable_init_response(&session_url),
        // Storage-provider XML body, not gateway JSON — mapping must fall
        // back to the HTTP status (403 → AccessDenied), with no retry.
        closed_response(403, "Forbidden", "", "<Error>SignatureDoesNotMatch</Error>"),
    ];

    let (result, reqs) = tokio::time::timeout(std::time::Duration::from_secs(10), async {
        tokio::join!(
            upload(&base_url, b"video".to_vec()),
            serve_upload_sequence(&listener, responses),
        )
    })
    .await
    .unwrap();

    let err = result.unwrap_err();
    assert!(
        matches!(err, Error::AccessDenied(_)),
        "expected AccessDenied, got {err:?}"
    );
    assert!(!err.is_retryable());
    // exactly one chunk PUT — a 4xx must never trigger recovery
    assert_eq!(reqs.len(), 3);
}

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

    let body = r#"{"success":false,"error":{"code":"VALIDATION_ERROR","message":"size_bytes exceeds 1 GB","request_id":"req-up-1"}}"#;
    let serve_task = serve_once(&listener, http_response(400, "Bad Request", "", body));

    let (result, _) = tokio::join!(
        async {
            client_for(&base_url)
                .upload_file()
                .data(b"v".to_vec())
                .content_type("video/mp4")
                .send()
                .await
        },
        serve_task,
    );

    let err = result.unwrap_err();
    assert!(!err.is_retryable());
    match err {
        Error::Validation(d) => {
            assert_eq!(d.error_code.as_deref(), Some("VALIDATION_ERROR"));
            assert_eq!(d.request_id.as_deref(), Some("req-up-1"));
        }
        other => panic!("expected Validation, got {other:?}"),
    }
}

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

    let responses = vec![
        closed_response(200, "OK", "", &upload_ticket_body(&base_url)),
        // Serve exactly ONE initiation response — a retry would hang past the
        // timeout. Presigned-URL responses carry storage-provider bodies, not
        // gateway JSON, so mapping falls back to HTTP status (403 → AccessDenied).
        closed_response(403, "Forbidden", "", "<Error>SignatureDoesNotMatch</Error>"),
    ];

    let (result, reqs) = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        tokio::join!(
            upload(&base_url, b"v".to_vec()),
            serve_upload_sequence(&listener, responses),
        )
    })
    .await
    .unwrap();

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

// ── image generation (spec §8) — POST /v1/images/generations ─────────────────

const IMAGE_URL_BODY: &str = r#"{"created":1752000000,"data":[{"url":"https://storage.example.com/img-1.png?sig=abc","revised_prompt":"A fluffy cat"}],"usage":{"input_tokens":12,"output_tokens":1024,"total_tokens":1036}}"#;

/// Drive one `generate_image` call against a capture server; return the
/// result together with the raw request the SDK put on the wire.
async fn generate_image_capture(
    builder: impl FnOnce(audacity_sdk::GenerateImageBuilder) -> audacity_sdk::GenerateImageBuilder,
    response: String,
) -> (Result<audacity_sdk::GenerateImageOutput, Error>, String) {
    let (listener, base_url) = bind().await;

    let capture = async {
        let (mut sock, _) = listener.accept().await.unwrap();
        let req_str = read_request(&mut sock).await;
        sock.write_all(response.as_bytes()).await.unwrap();
        req_str
    };

    let send = async { builder(client_for(&base_url).generate_image()).send().await };

    tokio::join!(send, capture)
}

#[tokio::test]
async fn generate_image_url_happy_path() {
    let (result, req_str) = generate_image_capture(
        |b| b.model("gpt-image-1").prompt("A fluffy cat"),
        http_response(200, "OK", "", IMAGE_URL_BODY),
    )
    .await;

    assert!(req_str.starts_with("POST /v1/images/generations HTTP/1.1"));
    let body_start = req_str.find("\r\n\r\n").unwrap() + 4;
    let body: serde_json::Value = serde_json::from_str(&req_str[body_start..]).unwrap();
    assert_eq!(
        body,
        serde_json::json!({"model": "gpt-image-1", "prompt": "A fluffy cat"})
    );

    let out = result.unwrap();
    assert_eq!(out.created, 1752000000);
    assert_eq!(out.data.len(), 1);
    assert_eq!(
        out.data[0].url.as_deref(),
        Some("https://storage.example.com/img-1.png?sig=abc")
    );
    assert_eq!(out.data[0].b64_json, None);
    assert_eq!(out.data[0].revised_prompt.as_deref(), Some("A fluffy cat"));
    let usage = out.usage.unwrap();
    assert_eq!(usage.input_tokens, Some(12));
    assert_eq!(usage.output_tokens, Some(1024));
    assert_eq!(usage.total_tokens, Some(1036));
}

#[tokio::test]
async fn generate_image_b64_json_response() {
    let body = r#"{"created":1752000001,"data":[{"b64_json":"aGVsbG8="},{"b64_json":"d29ybGQ="}]}"#;
    let (result, _) = generate_image_capture(
        |b| {
            b.model("gpt-image-1")
                .prompt("Two words")
                .n(2)
                .response_format("b64_json")
        },
        http_response(200, "OK", "", body),
    )
    .await;

    let out = result.unwrap();
    assert_eq!(out.data.len(), 2);
    assert_eq!(out.data[0].b64_json.as_deref(), Some("aGVsbG8="));
    assert_eq!(out.data[1].b64_json.as_deref(), Some("d29ybGQ="));
    assert_eq!(out.data[0].url, None);
    assert_eq!(out.usage, None);
}

#[tokio::test]
async fn generate_image_serializes_all_params() {
    let (result, req_str) = generate_image_capture(
        |b| {
            b.model("dall-e-3")
                .prompt("A watercolor fox")
                .n(3)
                .size("1792x1024")
                .quality("hd")
                .response_format("b64_json")
                .user("user-123")
        },
        http_response(200, "OK", "", IMAGE_URL_BODY),
    )
    .await;
    result.unwrap();

    let body_start = req_str.find("\r\n\r\n").unwrap() + 4;
    let body: serde_json::Value = serde_json::from_str(&req_str[body_start..]).unwrap();
    assert_eq!(
        body,
        serde_json::json!({
            "model": "dall-e-3",
            "prompt": "A watercolor fox",
            "n": 3,
            "size": "1792x1024",
            "quality": "hd",
            "response_format": "b64_json",
            "user": "user-123",
        })
    );
}

#[tokio::test]
async fn generate_image_401_maps_to_access_denied() {
    let body = r#"{"error":{"message":"Invalid API key.","type":"authentication_error","param":null,"code":"invalid_api_key"}}"#;
    let (result, _) = generate_image_capture(
        |b| b.model("gpt-image-1").prompt("x"),
        http_response(401, "Unauthorized", "", body),
    )
    .await;

    match result.unwrap_err() {
        Error::AccessDenied(d) => {
            assert_eq!(d.status_code, 401);
            assert_eq!(d.error_code.as_deref(), Some("invalid_api_key"));
        }
        other => panic!("expected AccessDenied, got {other:?}"),
    }
}

#[tokio::test]
async fn generate_image_402_maps_to_service_quota_exceeded() {
    let body = r#"{"error":{"message":"Spend cap exceeded.","type":"usage_cap_error","param":null,"code":"usage_cap_exceeded"}}"#;
    let (result, _) = generate_image_capture(
        |b| b.model("gpt-image-1").prompt("x"),
        http_response(402, "Payment Required", "", body),
    )
    .await;

    match result.unwrap_err() {
        Error::ServiceQuotaExceeded(d) => assert_eq!(d.status_code, 402),
        other => panic!("expected ServiceQuotaExceeded, got {other:?}"),
    }
}

#[tokio::test]
async fn generate_image_429_throttling_with_retry_after() {
    let body = r#"{"error":{"message":"Rate limit exceeded (rpm).","type":"rate_limit_error","param":null,"code":"rate_limit_exceeded"}}"#;
    let (result, _) = generate_image_capture(
        |b| b.model("gpt-image-1").prompt("x"),
        http_response(429, "Too Many Requests", "Retry-After: 7\r\n", body),
    )
    .await;

    match result.unwrap_err() {
        Error::Throttling(d) => assert_eq!(d.retry_after_seconds, Some(7.0)),
        other => panic!("expected Throttling, got {other:?}"),
    }
}

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

    let throttle_body =
        r#"{"error":{"message":"slow down","code":"rate_limit_exceeded"}}"#;
    let throttle_resp = http_response(429, "Too Many Requests", "Retry-After: 0\r\n", throttle_body);
    let ok_resp = http_response(200, "OK", "", IMAGE_URL_BODY);

    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_resp).await;
        serve_once(&listener, ok_resp).await;
    };

    let (result, _) = tokio::join!(
        client
            .generate_image()
            .model("gpt-image-1")
            .prompt("retry me")
            .send(),
        serve_two,
    );

    let out = result.unwrap();
    assert!(out.data[0].url.is_some());
}

#[tokio::test]
async fn generate_image_missing_required_fields() {
    let client = client_for("http://127.0.0.1:1");

    let err = client.generate_image().prompt("x").send().await.unwrap_err();
    assert!(matches!(err, Error::Validation(_)), "got {err:?}");

    let err = client
        .generate_image()
        .model("gpt-image-1")
        .send()
        .await
        .unwrap_err();
    assert!(matches!(err, Error::Validation(_)), "got {err:?}");
}