spb/doc-api

1"""Batch API tests. List is cheap and always runs; the end-to-end test creates a 1-request batch (≈ $0.000004) and is2gated by RUN_BATCH_TESTS=true (marker run_batch_tests). Batches themselves cannot be deleted; files are."""3from __future__ import annotations4import json, time, uuid5import pytest678def test_list_batches_shape(openai):9    st, page, _ = openai("GET", "/v1/batches?limit=2", note="test_batch list")10    assert st == 200 and page["object"] == "list" and "has_more" in page11    for b in page["data"]:12        assert b["object"] == "batch" and b["status"] in {"validating", "failed", "in_progress", "finalizing", "completed", "expired", "cancelling", "cancelled"}131415def test_retrieve_unknown_batch_404(openai):16    st, body, _ = openai("GET", "/v1/batches/batch_atlasdoesnotexist", note="test_batch bogus id")17    assert st == 404 and body["error"]["type"] == "invalid_request_error"181920@pytest.mark.run_batch_tests21def test_batch_end_to_end(openai, models):22    line = {"custom_id": "atlas-platform-agent-test-1", "method": "POST", "url": "/v1/responses",23            "body": {"model": models["openai"], "input": "Reply with OK.", "max_output_tokens": 16}}24    b = f"----atlas{uuid.uuid4().hex}"25    body = (f"--{b}\r\nContent-Disposition: form-data; name=\"purpose\"\r\n\r\nbatch\r\n--{b}\r\n"26            f"Content-Disposition: form-data; name=\"file\"; filename=\"in.jsonl\"\r\nContent-Type: application/jsonl\r\n\r\n").encode() \27        + (json.dumps(line) + "\n").encode() + f"\r\n--{b}--\r\n".encode()28    st, f, _ = openai("POST", "/v1/files", data=body, content_type=f"multipart/form-data; boundary={b}", note="test_batch upload")29    assert st == 200 and f["purpose"] == "batch" and f["expires_at"] == f["created_at"] + 30 * 8640030    files = [f["id"]]31    try:32        st, batch, _ = openai("POST", "/v1/batches", {"input_file_id": f["id"], "endpoint": "/v1/responses", "completion_window": "24h",33                              "metadata": {"owner": "atlas-platform-agent-test"}, "output_expires_after": {"anchor": "created_at", "seconds": 3600}},34                              note="test_batch create", est_cost_usd=0.00001)35        assert st == 200 and batch["object"] == "batch" and batch["status"] == "validating"36        assert batch["expires_at"] == batch["created_at"] + 8640037        t0 = time.time()38        while batch["status"] not in ("completed", "failed", "expired", "cancelled") and time.time() - t0 < 180:39            time.sleep(15)40            st, batch, _ = openai("GET", f"/v1/batches/{batch['id']}", note="test_batch poll")41        if batch["status"] not in ("completed", "failed", "expired", "cancelled"):42            st, c, _ = openai("POST", f"/v1/batches/{batch['id']}/cancel", note="test_batch cancel")43            assert st == 200 and c["status"] in ("cancelling", "cancelled")44            pytest.skip("batch did not finish within 3 minutes; cancelled")45        assert batch["status"] == "completed", batch46        assert batch["request_counts"] == {"total": 1, "completed": 1, "failed": 0}47        assert batch["usage"]["total_tokens"] > 048        files.append(batch["output_file_id"])49        st, out, _ = openai("GET", f"/v1/files/{batch['output_file_id']}/content", content_type=None, note="test_batch output")50        row = json.loads(out.decode().splitlines()[0])51        assert row["custom_id"] == line["custom_id"] and row["error"] is None52        assert row["response"]["status_code"] == 200 and row["response"]["body"]["object"] == "response"53        st, meta, _ = openai("GET", f"/v1/files/{batch['output_file_id']}", note="test_batch output meta")54        assert meta["purpose"] == "batch_output" and meta["expires_at"] == meta["created_at"] + 360055    finally:56        for fid in files:57            openai("DELETE", f"/v1/files/{fid}", note="test_batch cleanup")58