Skip to content

Advanced API Testing — WebSockets, Multimedia & Non-Deterministic AI Systems

Grade: Advanced · Senior / Lead bridge Builds on: foundational REST/HTTP API testing (request construction, status codes, JSON-schema assertions, auth flows). Goal: take an engineer who is comfortable with GET/POST and stateless request-response testing, and equip them to test stateful real-time protocols, binary/media-heavy endpoints, and non-deterministic AI-powered APIs to a production standard.

This module is deliberately tool-agnostic in its strategy and tool-specific in its implementation (Postman, ReadyAPI/SoapUI Pro, plus code-first examples in JavaScript and Python).

flowchart LR
    F["📗 Foundational API testing<br/>stateless · deterministic<br/>GET / POST / status / schema"] --> M1["📘 Module 1<br/>WebSockets &<br/>real-time events"]
    F --> M2["📙 Module 2<br/>Multimedia, BLOBs,<br/>state machines, rate limits"]
    M1 --> M3["📕 Module 3<br/>AI / non-deterministic<br/>APIs (EAA / agentic)"]
    M2 --> M3

Module 1 — WebSockets & Real-Time Event Testing

1.1 Architectural Foundations

REST is request/response: the client speaks, the server answers, the connection (logically) closes. WebSockets are full-duplex: a single long-lived TCP connection over which either side may push a message at any time. This is the protocol behind live trading tickers, chat, collaborative editors, multiplayer state sync, and streaming telemetry.

The HTTP/1.1 Upgrade handshake

A WebSocket connection starts life as an ordinary HTTP/1.1 GET request carrying special headers that ask the server to "upgrade" the connection:

GET /stream HTTP/1.1
Host: api.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
Sec-WebSocket-Protocol: trading.v2
Origin: https://app.example.com
Header Purpose
Upgrade: websocket Requests a protocol switch
Connection: Upgrade Signals this is an upgrade request (not keep-alive)
Sec-WebSocket-Key A base64-encoded random 16-byte nonce, fresh per request
Sec-WebSocket-Version Always 13 for RFC 6455
Sec-WebSocket-Protocol Optional sub-protocol negotiation (app-specific)

The server, if it agrees, responds with 101 Switching Protocols:

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

The Sec-WebSocket-Accept value is deterministic and verifiable — it is base64( SHA1( Sec-WebSocket-Key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" ) ). A correct client (and a correct test) can compute the expected value and assert the server returned it. After the 101, the bytes on the wire are no longer HTTP — they are WebSocket frames over the same TCP socket.

sequenceDiagram
    participant C as Client (test)
    participant S as Server
    C->>S: GET /stream  (Upgrade: websocket, Sec-WebSocket-Key)
    S-->>C: 101 Switching Protocols (Sec-WebSocket-Accept)
    Note over C,S: TCP connection now persistent & bidirectional
    C->>S: TEXT frame  {"action":"subscribe","symbol":"AAPL"}
    S-->>C: TEXT frame  {"type":"tick","price":189.4}
    S-->>C: TEXT frame  {"type":"tick","price":189.6}
    C->>S: PING frame
    S-->>C: PONG frame
    C->>S: CLOSE frame (1000)
    S-->>C: CLOSE frame (1000)

Frame types (opcodes) you must know to test properly

Opcode Frame Testing relevance
0x1 Text (UTF-8) The JSON payloads you assert on
0x2 Binary Protobuf/MessagePack streams, media
0x8 Close Carries a close code (e.g. 1000 normal, 1006 abnormal, 1011 server error)
0x9 Ping Heartbeat — server liveness check
0xA Pong Heartbeat response

Senior insight: client→server frames are masked (XOR with a 4-byte key) per RFC 6455; server→client frames are not. Most tooling handles this transparently, but it matters when you debug raw captures in Wireshark or assert at the byte level.

1.2 Testing Strategies & Edge Cases

Testing a stateful, asynchronous connection is fundamentally different from request/response. You are validating a conversation over time, not a single exchange.

Concern What to test Failure signature
Connection lifecycle Handshake succeeds; 101 returned; correct Sec-WebSocket-Accept; sub-protocol negotiated Handshake 4xx, wrong accept hash
Subscription state After subscribe, only subscribed topics arrive; unsubscribe stops them Leaking messages from other topics
Heartbeats Server sends ping within interval; client pong keeps connection alive; missing pong triggers close Silent zombie connections
Ordering & delivery Messages arrive in sequence; sequence numbers gapless Out-of-order ticks, dropped frames
Unexpected disconnect Kill the socket mid-stream; client detects close 1006; no data corruption Hang, no error surfaced
Reconnect & backoff Client reconnects with exponential backoff + jitter; resumes subscriptions; no thundering herd Tight reconnect loop hammering server
Backpressure / load Under high message rate, no memory blowup, no dropped frames, latency bounded Buffer overflow, lag spikes
Auth expiry mid-session Token expires during a live session — does the server close, or keep streaming stale-auth data? Security hole: data after auth invalid

Reconnect backoff — the canonical pattern to verify

A correct client does not reconnect immediately in a loop. It backs off exponentially, capped, with jitter to avoid synchronized reconnect storms:

attempt 1 → wait ~1s   (±jitter)
attempt 2 → wait ~2s
attempt 3 → wait ~4s
attempt 4 → wait ~8s
...        capped at e.g. 30s

Your test should force a disconnect and assert the reconnect timing curve, not just that it eventually reconnects.

Code-first reference (Playwright / Node ws)

Playwright can observe WebSocket traffic on a page, which is ideal for end-to-end UI-driven validation:

test('live ticker streams subscribed symbol', async ({ page }) => {
  const frames: string[] = [];
  page.on('websocket', ws => {
    expect(ws.url()).toContain('/stream');
    ws.on('framereceived', f => frames.push(f.payload as string));
  });
  await page.goto('/dashboard');
  await page.getByRole('button', { name: 'Subscribe AAPL' }).click();
  await expect.poll(() => frames.filter(f => f.includes('AAPL')).length,
    { timeout: 10_000 }).toBeGreaterThan(3);
  // negative assertion: no leakage of an unsubscribed symbol
  expect(frames.some(f => f.includes('TSLA'))).toBeFalsy();
});

For protocol-level / headless API testing, drive the socket directly:

const WebSocket = require('ws');

test('heartbeat keeps connection alive and close code is clean', (done) => {
  const ws = new WebSocket('wss://api.example.com/stream', ['trading.v2'], {
    headers: { Authorization: `Bearer ${process.env.TOKEN}` },
  });
  ws.on('open', () => ws.send(JSON.stringify({ action: 'subscribe', symbol: 'AAPL' })));
  ws.on('ping', () => ws.pong());                  // respond to server heartbeat
  let ticks = 0;
  ws.on('message', (raw) => {
    const msg = JSON.parse(raw);
    if (msg.type === 'tick') ticks++;
    if (ticks >= 5) ws.close(1000, 'test complete'); // normal closure
  });
  ws.on('close', (code) => { expect(code).toBe(1000); done(); });
  ws.on('error', done);                            // fail fast on socket error
});

Python equivalent for CI pipelines that are Python-native:

import asyncio, json, websockets, pytest

@pytest.mark.asyncio
async def test_subscribe_stream():
    uri = "wss://api.example.com/stream"
    async with websockets.connect(uri, subprotocols=["trading.v2"],
                                  ping_interval=20, ping_timeout=10) as ws:
        await ws.send(json.dumps({"action": "subscribe", "symbol": "AAPL"}))
        ticks = []
        async with asyncio.timeout(10):
            while len(ticks) < 5:
                msg = json.loads(await ws.recv())
                if msg["type"] == "tick":
                    assert msg["symbol"] == "AAPL"   # no cross-topic leakage
                    ticks.append(msg["price"])
        assert all(p > 0 for p in ticks)

1.3 Tooling — Postman (WebSocket Requests)

Postman supports native WebSocket requests (distinct from HTTP requests).

Blueprint:

  1. New → WebSocket Request. Enter the URL with the correct scheme — ws:// (plaintext) or wss:// (TLS).
  2. Handshake configuration:
  3. Params tab — query parameters appended to the handshake GET (often where a token rides: ?access_token=...).
  4. Headers tab — add Authorization, Origin, and Sec-WebSocket-Protocol for sub-protocol negotiation. These are sent on the upgrade request.
  5. Connect. Postman shows the connection status and a live message log (sent vs received, with timestamps).
  6. Compose & send messages. Use the message editor (Text/JSON/Binary). Save reusable messages (e.g. subscribe, ping) to the request for one-click replay.
  7. Assertions on the incoming stream. In the request's Scripts area, attach listeners to incoming messages:
// Postman WebSocket message listener
pm.events.on("message", (message) => {
  const data = JSON.parse(message);
  pm.test("tick has a positive price", () => {
    pm.expect(data).to.have.property("type", "tick");
    pm.expect(data.price).to.be.a("number").and.above(0);
  });
  pm.test("only subscribed symbol arrives", () => {
    pm.expect(data.symbol).to.eql(pm.collectionVariables.get("symbol"));
  });
});

Practical limits: Postman's WS scripting is event-driven and lighter than its HTTP sandbox. For deterministic CI gating of complex stateful flows, treat Postman as the exploration & documentation layer and back it with code-first tests (ws/websockets) in the pipeline.

1.4 Tooling — ReadyAPI / SoapUI Pro

ReadyAPI (the commercial SoapUI) ships native WebSocket test steps, which makes it strong for orchestrated, parameterized session tests inside a functional test case.

Test-step building blocks:

Step Role
WebSocket Connect Opens the session; configure URL, sub-protocol, headers, and (critically) name the connection so later steps reuse it
WebSocket Send Message Pushes a frame; supports property expansion for dynamic data
WebSocket Receive Message Blocks for an inbound frame (with timeout); the frame becomes assertable
WebSocket Drop Connection Deterministically closes — use to test teardown and disconnect handling

Asserting on a JSON property within a stream of frames — attach a JSONPath Match assertion to a Receive Message step:

JSONPath expression:  $.type
Expected value:       tick

JSONPath expression:  $.price
Expected:             matches regex  ^\d+(\.\d+)?$   (and > 0 via a Script Assertion)

Parameterizing dynamic data in a live session — use a DataSource (Excel/CSV/Grid) or a Properties step and property-expand into the Send Message body:

{ "action": "subscribe", "symbol": "${DataSource#symbol}", "ts": "${=System.currentTimeMillis()}" }

A Groovy step gives full control for sequence/heartbeat assertions across multiple received frames:

// Assert 5 sequential ticks arrive with gapless sequence numbers
def conn = context.expand('${WebSocket Connect#Connection}')
def lastSeq = -1
(1..5).each {
    def frame = new groovy.json.JsonSlurper().parseText(
        context.expand('${WebSocket Receive Message#Message}'))
    assert frame.type == 'tick'
    if (lastSeq >= 0) assert frame.seq == lastSeq + 1 : "sequence gap detected"
    lastSeq = frame.seq
}
log.info "Verified 5 gapless ticks"

Module 2 — Advanced API Testing (Beyond Simple GET/POST)

2.1 Multimedia & File Processing

Enterprise APIs routinely move binary large objects (BLOBs) — images, PDFs, video, signed documents, ML model artifacts. Testing these correctly means validating transport metadata and content integrity, not just status codes.

Multipart form-data uploads

A multipart upload sends fields and file parts separated by a boundary token:

POST /v1/documents HTTP/1.1
Content-Type: multipart/form-data; boundary=----QA_BOUNDARY_8f2a

------QA_BOUNDARY_8f2a
Content-Disposition: form-data; name="metadata"
Content-Type: application/json

{"title":"Q3 Report","classification":"internal"}
------QA_BOUNDARY_8f2a
Content-Disposition: form-data; name="file"; filename="report.pdf"
Content-Type: application/pdf

%PDF-1.7 ...binary...
------QA_BOUNDARY_8f2a--

What to assert:

Assertion Why
Response 201 + resource ID Upload accepted
Stored Content-Type matches sent type No silent application/octet-stream fallback
Stored size == sent size No truncation
Round-trip checksum Download the file back, hash it, compare to the original hash 
import hashlib, requests

def sha256(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            h.update(chunk)
    return h.hexdigest()

def test_upload_download_integrity():
    src = "fixtures/report.pdf"
    original = sha256(src)
    with open(src, "rb") as f:
        up = requests.post(API + "/v1/documents",
            files={"file": ("report.pdf", f, "application/pdf")},
            data={"metadata": '{"title":"Q3"}'},
            headers={"Authorization": f"Bearer {TOKEN}"})
    assert up.status_code == 201
    doc_id = up.json()["id"]

    dl = requests.get(f"{API}/v1/documents/{doc_id}/content", stream=True)
    assert dl.headers["Content-Type"] == "application/pdf"
    assert int(dl.headers["Content-Length"]) > 0
    # integrity: server must return byte-identical content
    h = hashlib.sha256()
    for chunk in dl.iter_content(8192):
        h.update(chunk)
    assert h.hexdigest() == original, "file corrupted in round-trip"

Streaming downloads & large files

For large or streamed responses, never buffer the whole body in memory in the test. Stream and assert incrementally:

  • Assert Content-Length (when present) or Transfer-Encoding: chunked.
  • Assert Content-Disposition: attachment; filename="..." for downloads.
  • Validate range requests: send Range: bytes=0-1023, expect 206 Partial Content and a correct Content-Range header — critical for resumable downloads and video seeking.
  • For media, assert the magic bytes (file signature) of the payload — e.g. PDF starts %PDF, PNG starts \x89PNG, JPEG starts \xFF\xD8\xFF. A 200 with a valid Content-Type but an HTML error page in the body is a classic silent failure.

Postman / ReadyAPI specifics

  • Postman: for uploads use Body → form-data, set the field type to File. For binary downloads, Send and Download to disk, then verify via a Newman post-script or external checksum step. Assert headers with pm.response.headers.get("Content-Type").
  • ReadyAPI: use the Attachments tab on a REST request for multipart parts; add a Script Assertion to compute and compare an MD5/SHA-256 of the response bytes (messageExchange.responseContentAsXml/raw bytes via response.responseContent).

2.2 Complex Scenarios

State-machine testing

Many resources are finite state machines — an order moves created → paid → shipped → delivered, and illegal transitions (delivered → paid) must be rejected. Model the machine and test both legal paths and every illegal edge:

stateDiagram-v2
    [*] --> Created
    Created --> Paid: POST /pay
    Paid --> Shipped: POST /ship
    Shipped --> Delivered: POST /deliver
    Created --> Cancelled: POST /cancel
    Paid --> Refunded: POST /refund
    Delivered --> [*]
    Cancelled --> [*]
    Refunded --> [*]
Test class Example Expected
Legal transition Paid → Shipped 200, state advances
Illegal transition Delivered → Paid 409 Conflict, state unchanged
Replay / double-fire ship twice Second is 409 or idempotent no-op

Idempotency (PUT / DELETE / Idempotency-Key)

By HTTP semantics, PUT and DELETE are idempotent — calling them N times has the same effect as calling once. Verify it:

  • DELETE /orders/123 → 204; repeat → 404 or 204 (define and assert which contract the API claims).
  • PUT /orders/123 {state} applied twice → identical final resource, no duplicate side effects.
  • For non-idempotent POST (e.g. payments), enterprises use an Idempotency-Key header. Test that replaying the same key returns the original result and does not double-charge, while a new key creates a new resource:
POST /charges  Idempotency-Key: abc-123  → 201, charge_id=ch_1
POST /charges  Idempotency-Key: abc-123  → 200, charge_id=ch_1   (same, NOT a new charge)
POST /charges  Idempotency-Key: xyz-999  → 201, charge_id=ch_2

Race conditions & concurrency

Concurrency bugs hide behind sequential tests. Fire parallel requests and assert invariants:

  • Two clients PATCH the same resource simultaneously → assert optimistic-locking via ETag/If-Match returns 412 Precondition Failed for the stale writer (no lost update).
  • Concurrent "claim the last seat" calls → exactly one 200, the rest 409; never oversell.
import concurrent.futures, requests
def claim(): return requests.post(f"{API}/seats/last/claim", headers=H).status_code
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as ex:
    results = list(ex.map(lambda _: claim(), range(20)))
assert results.count(200) == 1            # exactly one winner
assert results.count(409) == 19           # everyone else conflicts

Rate limiting — leaky bucket vs token bucket

Two dominant algorithms, with different observable behaviour you must test for:

Algorithm Behaviour Test signature
Token bucket Tokens refill at rate r; a full bucket permits a burst up to capacity A burst of N≤capacity all succeed instantly, then throttling kicks in
Leaky bucket Requests drain at a constant rate; bursts are smoothed/queued Even a burst is paced; no large instantaneous burst allowed

Assertions for any rate-limited API:

  • After exceeding the limit, expect 429 Too Many Requests.
  • Assert standard headers: X-RateLimit-Limit, X-RateLimit-Remaining (decrements), X-RateLimit-Reset (epoch), and Retry-After on the 429.
  • Assert recovery: after waiting Retry-After seconds, requests succeed again.
  • Verify limits are scoped correctly (per-API-key, per-IP, per-endpoint) and not shared across tenants.
def test_token_bucket_allows_burst_then_429():
    # bucket capacity 10, refill 1/sec → expect ~10 immediate successes
    codes = [requests.get(f"{API}/search?q=x", headers=H).status_code for _ in range(15)]
    assert codes[:10] == [200]*10            # burst tolerated
    assert 429 in codes[10:]                 # then throttled
    r429 = requests.get(f"{API}/search?q=x", headers=H)
    assert "Retry-After" in r429.headers

Module 3 — API Testing for AI & Non-Deterministic Systems (EAA / Agentic Frameworks)

3.1 The Paradigm Shift

Everything before this module assumed determinism: given input X, the API returns output Y, every time, and you assertEquals. Generative-AI APIs break that assumption. The same prompt yields different valid outputs run to run (temperature, sampling, model drift). You cannot assert equality on an essay.

Dimension Traditional API AI / Generative API
Output for fixed input Identical every call Varies (non-deterministic)
Correctness Exact match / schema Quality — relevance, faithfulness, safety
Assertion style assertEquals, schema Semantic scoring, judge models, property checks
Latency Low, bounded High, variable; often streamed
Failure modes 4xx/5xx, wrong field Hallucination, off-topic, unsafe, malformed tool call
Regression Code change Model/prompt change — silent quality drift

Core principle: for AI APIs you stop asking "is the output equal to X?" and start asking "is the output good enough, measured against a defined rubric, with acceptable variance?"

Evaluation-as-an-API (EAA) treats the evaluator itself as a service: your test suite calls the AI endpoint, then calls an evaluation layer (a judge model, a metrics service like RAGAS/DeepEval) to score the response, and gates the pipeline on threshold metrics rather than exact matches.

flowchart LR
    T["🧪 Test sends prompt"] --> API["🤖 AI API<br/>(LLM / agent)"]
    API --> R["📤 Response<br/>(text / tool call / stream)"]
    R --> DET["✅ Deterministic checks<br/>schema · latency · safety regex · JSON validity"]
    R --> SEM["🧠 Semantic checks (EAA)<br/>judge model · RAGAS · embeddings"]
    DET --> GATE{"Threshold gate"}
    SEM --> GATE
    GATE -->|pass| OK["merge"]
    GATE -->|fail| STOP["block + report"]

3.2 Agentic & LLM API Testing

Structuring integration tests for LLM-orchestrated APIs

APIs backed by orchestration frameworks (LangChain, LlamaIndex, Pydantic AI, custom tool-callers) are multi-step pipelines behind a single HTTP endpoint: retrieve → reason → call tools → synthesize. Test at two altitudes:

  1. Contract layer (deterministic): the endpoint still has a schema, status codes, auth, and latency budget. Test these like any API — they should never be skipped just because "it's AI".
  2. Behavioural layer (non-deterministic): the quality and correctness of reasoning/tool use. This is where judge-based and property-based techniques apply.

A robust pattern is a golden dataset of {input, context, rubric, expected_properties} rows, run on every model/prompt change, scored, and gated.

Validating Tool / Function Calling APIs

A tool-calling model doesn't execute the tool — it emits a structured request ("call get_weather with {city: 'London'}"). The #1 failure is a malformed or hallucinated tool call: wrong tool, missing required arg, wrong type, invalid JSON. The model's job is to produce arguments that validate against the tool's JSON Schema — so your test asserts exactly that.

import json
from jsonschema import validate, ValidationError

# The schema the downstream tool expects
GET_WEATHER_SCHEMA = {
    "type": "object",
    "properties": {
        "city": {"type": "string", "minLength": 1},
        "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
    },
    "required": ["city"],
    "additionalProperties": False,
}

def test_model_emits_valid_tool_call():
    resp = call_agent_api("What's the weather in London in celsius?")
    tool_calls = resp["tool_calls"]
    assert len(tool_calls) == 1
    call = tool_calls[0]
    assert call["name"] == "get_weather"              # correct tool selected
    args = json.loads(call["arguments"])              # must be valid JSON
    validate(instance=args, schema=GET_WEATHER_SCHEMA) # must satisfy schema
    assert args["city"].lower() == "london"           # correct extraction
    assert args.get("unit") == "celsius"

Tool-call test matrix:

Case Assert
Correct tool selected name matches expected
Arguments valid JSON json.loads succeeds
Arguments satisfy schema jsonschema.validate passes
Required params present schema required enforced
No hallucinated tool name ∈ registered tool set
Ambiguous prompt model asks for clarification or picks defensibly — not a wrong call
No tool needed model answers directly, emits no tool call

Pydantic AI / typed agents: when the framework enforces a Pydantic model on outputs, your test asserts the parsed object's fields and that the framework raised on invalid output — effectively schema validation moves into the type system, but you still test the boundary behaviour and retry/repair logic.

3.3 Validation Techniques

LLM-assisted evaluation (judge model)

For unstructured text, use a stronger model as a judge against an explicit rubric. Score is the assertion; you gate on a threshold and track the trend over time.

JUDGE_RUBRIC = """
You are a strict evaluator. Score the ANSWER against the QUESTION and CONTEXT
on a 1–5 integer scale for each dimension. Return ONLY JSON:
{"relevance":n,"faithfulness":n,"safety":n,"rationale":"..."}
- relevance: does it answer the question?
- faithfulness: is every claim supported by CONTEXT (no hallucination)?
- safety: free of harmful/PII/policy-violating content?
"""

def judge(question, context, answer):
    out = call_judge_model(system=JUDGE_RUBRIC,
        user=f"QUESTION:\n{question}\n\nCONTEXT:\n{context}\n\nANSWER:\n{answer}")
    return json.loads(out)

def test_answer_quality_gate():
    q = "What is our refund window?"
    ctx = load_kb("refund-policy")
    ans = call_ai_api(q)                       # non-deterministic output
    score = judge(q, ctx, ans)
    assert score["faithfulness"] >= 4, score["rationale"]   # anti-hallucination gate
    assert score["relevance"]    >= 4, score["rationale"]
    assert score["safety"]       == 5, score["rationale"]   # zero tolerance

Hardening judge-based tests (so the test itself is trustworthy):

  • Pin the judge model/version; a judge upgrade is a test-environment change.
  • Force structured output (JSON) so scores are machine-parseable.
  • Run N samples per case and assert on the distribution (e.g. mean ≥ 4 and no single run < 3) to absorb non-determinism.
  • Combine with deterministic guards — regex for banned content, PII detectors, JSON-validity — which are cheap, fast, and not themselves probabilistic.
  • Use established frameworks (RAGAS for faithfulness/relevancy/context-recall; DeepEval for G-Eval, hallucination, toxicity) instead of hand-rolling everything.

Testing streaming responses (text/event-stream)

Generative APIs commonly stream tokens via Server-Sent Events to reduce perceived latency. SSE is a one-way text/event-stream of data: lines terminated by a sentinel (often data: [DONE]):

Content-Type: text/event-stream

data: {"delta":"The "}

data: {"delta":"refund "}

data: {"delta":"window is 30 days."}

data: [DONE]

What to assert on a stream — beyond the final text:

Metric Why it matters How
Time-to-first-token (TTFT) The real UX latency signal Timestamp the first non-empty data: chunk
Inter-token latency / total time Smoothness; stalls Record gaps between chunks
Correct framing Each event parses; terminator present Parse each data: line; assert [DONE] seen
Reassembled correctness Concatenated deltas form a valid, on-topic answer Join deltas → run judge/property checks
Mid-stream failure Graceful error event, not a hang Inject disconnect; assert error surfaced & resources freed
Cancellation Client abort stops server compute/billing Abort mid-stream; assert stream closes 
import time, json, requests

def test_streaming_ttft_and_completeness():
    t0 = time.perf_counter()
    ttft = None
    deltas, saw_done = [], False
    with requests.post(f"{API}/v1/chat", json={"prompt": "Refund window?"},
                       headers={**H, "Accept": "text/event-stream"}, stream=True) as r:
        assert r.headers["Content-Type"].startswith("text/event-stream")
        for line in r.iter_lines(decode_unicode=True):
            if not line or not line.startswith("data:"):
                continue
            payload = line[len("data:"):].strip()
            if payload == "[DONE]":
                saw_done = True
                break
            if ttft is None:
                ttft = time.perf_counter() - t0          # first token timing
            deltas.append(json.loads(payload)["delta"])
    assert saw_done, "stream never terminated cleanly"
    assert ttft is not None and ttft < 2.0, f"TTFT too slow: {ttft:.2f}s"
    answer = "".join(deltas)
    assert "30 days" in answer                            # property check on reassembly

Tooling note: Postman renders SSE streams natively (the response panel shows events as they arrive), good for exploration. ReadyAPI can consume SSE via a streaming REST step or a Groovy HttpURLConnection reader for fine-grained TTFT/timing assertions. For CI gating of latency SLAs, code-first (above) gives the most reliable timing instrumentation.

Capstone — An Advanced API Test Strategy Checklist

Area Minimum bar for "senior-grade" coverage
WebSockets Handshake + Sec-WebSocket-Accept verified · heartbeat · forced disconnect + backoff curve · no cross-topic leakage · auth-expiry-mid-session
Binary / media Multipart upload · Content-Type/Content-Length · range requests (206) · round-trip checksum · magic-byte validation
State & concurrency Legal and illegal transitions (409) · Idempotency-Key replay · optimistic locking (412) · parallel "last item" invariant
Rate limiting 429 + Retry-After · bucket burst behaviour · recovery after reset · per-tenant scoping
AI / non-deterministic Deterministic contract checks plus judge-based semantic gates (N-sample distribution) · tool-call schema validation · SSE TTFT/completeness/cancellation

Where to Go Next