Wire-level / live-server streaming tests

The SKILL.md categories run against in-process harnesses (bufconn / InProcessServer); this reference covers the live-server variant - real channels against localhost:50051 - where deadline propagation, cancellation observed server-side, metadata round-trips, and ghz load behaviour are exercised over a real transport.

When to use

• Service exposes streaming RPCs (price ticker, log tail, IoT telemetry, AI streaming inference).
• Pre-deploy gate: deadline + cancellation propagate correctly, partial-stream errors return correct status codes.
• Load test gate: streams handle backpressure without OOM or silent drops.

How to use

1. Pick the test tool for the job (Step 1): native stubs for unit/integration, grpcurl for smoke, ghz for load.
2. Prove the plumbing with a unary sanity call before touching streams (Step 2).
3. Cover each streaming shape the service exposes - server-, client-, and bidirectional-streaming (Steps 3-5).
4. Assert deadline propagation and client-initiated cancellation are observed server-side (Steps 6-7).
5. Check error paths return the exact status code, not just "an error", and that request metadata round-trips (status-codes-metadata-load.md).
6. Run a ghz load pass to confirm streams handle backpressure without OOM or silent drops (same reference).
7. Gate the suite on the anti-patterns table before merge.

Step 1 - Pick the test tool

Step 2 - Unary RPC sanity (baseline)

Per the gRPC core concepts docs, Unary = "single request, single response." Use this to verify the RPC plumbing before testing streams:

import grpc
from orders_pb2 import OrderRequest
from orders_pb2_grpc import OrdersStub

def test_unary_create_order():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = OrdersStub(ch)
        resp = stub.CreateOrder(OrderRequest(item_count=2), timeout=5.0)
        assert resp.order_id != ""

Step 3 - Server-streaming test

Per the gRPC core concepts docs, server-streaming = "client sends a request and gets a stream to read a sequence of messages back."

def test_server_streaming_price_ticker():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = PricesStub(ch)
        stream = stub.SubscribePrices(SubscribeRequest(symbol="AAPL"), timeout=10.0)
        ticks = []
        for tick in stream:
            ticks.append(tick)
            if len(ticks) >= 5:
                stream.cancel()
                break

        assert len(ticks) == 5
        assert all(t.symbol == "AAPL" for t in ticks)

Step 4 - Client-streaming test

Per the gRPC core concepts docs, client-streaming = "client writes a sequence of messages and sends them to the server."

def test_client_streaming_upload():
    def chunks():
        for i in range(10):
            yield UploadChunk(seq=i, data=b"x" * 1024)

    with grpc.insecure_channel("localhost:50051") as ch:
        stub = UploadsStub(ch)
        resp = stub.Upload(chunks(), timeout=10.0)

        assert resp.total_chunks == 10
        assert resp.total_bytes == 10 * 1024

Step 5 - Bidirectional streaming + ordering

Per the gRPC core concepts docs, bidirectional streams "operate independently" - server may emit messages before reading any client message, after, or interleaved.

import asyncio

async def test_bidi_chat():
    async def client_messages():
        for msg in ["hello", "how are you", "bye"]:
            yield ChatMessage(text=msg)
            await asyncio.sleep(0.1)

    async with grpc.aio.insecure_channel("localhost:50051") as ch:
        stub = ChatStub(ch)
        responses = []
        async for resp in stub.Chat(client_messages()):
            responses.append(resp)

        assert len(responses) >= 3

Step 6 - Deadline propagation

Per the gRPC core concepts docs, "Clients specify maximum wait time; RPCs terminate with DEADLINE_EXCEEDED if exceeded."

def test_deadline_returns_correct_status():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = SlowStub(ch)
        with pytest.raises(grpc.RpcError) as exc_info:
            stub.SlowOperation(SlowRequest(), timeout=0.5)

        assert exc_info.value.code() == grpc.StatusCode.DEADLINE_EXCEEDED

Verify the server-side:

def test_server_observes_deadline_propagation():
    # Service should respect deadline and cancel its own downstream calls
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = OrchestratorStub(ch)
        with pytest.raises(grpc.RpcError):
            stub.Compose(ComposeRequest(), timeout=0.1)

    # Verify downstream call observed the cancellation
    downstream_state = fetch_downstream_state()
    assert downstream_state.cancelled_count >= 1

Step 7 - Cancellation behavior

Per the gRPC core concepts docs, "Either party can terminate an RPC immediately. Changes made before a cancellation are not rolled back."

def test_cancellation_is_observed_server_side():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = LongRunningStub(ch)
        future = stub.LongOperation.future(LongRequest())
        time.sleep(0.5)
        future.cancel()

        # Server should record cancellation
        time.sleep(0.5)
        state = fetch_server_metrics()
        assert state.cancelled_count >= 1

Status codes, metadata, and load testing

See status-codes-metadata-load.md for the full status-code matrix (OK, CANCELLED, DEADLINE_EXCEEDED, INVALID_ARGUMENT, UNAVAILABLE, ...), a request/response metadata round-trip test, and load testing with ghz.

Worked example

A prices service exposes SubscribePrices, a server-streaming RPC. QA needs to confirm the client receives ordered ticks and that cancelling the stream is observed server-side.

1. Start with the unary sanity call (Step 2) to confirm the channel and stubs are wired.
2. Open the stream with a 10s deadline and read 5 ticks: stream = stub.SubscribePrices(SubscribeRequest(symbol="AAPL"), timeout=10.0).
3. After the 5th tick, call stream.cancel() and break (Step 3).
4. Assert len(ticks) == 5 and every t.symbol == "AAPL".
5. Add a cancellation check (Step 7): fetch server metrics and assert cancelled_count >= 1, proving the server observed the client cancel rather than orphaning work.

Result: the ticker stream is verified for ordered delivery, a clean 10s deadline, and server-side cancellation - the three behaviors a server-streaming RPC most often regresses on.

Anti-patterns

Limitations

• gRPC-Web uses HTTP/1.1 fallback; some streaming patterns (client/bidi) are not supported. Test gRPC-Web specifically if used.
• Long-lived bidi streams hide individual-message error codes; channel-level state matters more.
• ghz protobuf reflection requires the server to enable reflection service (not always on in production builds).

References

• gRPC core concepts docs - RPC patterns, deadlines, cancellation, status codes, metadata
• websocket-tests (qa-realtime-protocols) - WebSocket alternative for non-gRPC stacks
• server-sent-events-tests (qa-realtime-protocols) - one-way HTTP streaming alternative