grpc-streaming-tests
Test gRPC streaming RPCs - Server-streaming (server returns sequence), Client-streaming (client sends sequence), Bidirectional (both sides stream independently). Cover deadline + cancellation + flow control + status codes (CANCELLED, DEADLINE_EXCEEDED) + metadata. Use ghz for load, grpcurl for ad-hoc, language-native test stubs for unit/integration. Use when a service exposes server-, client-, or bidirectional-streaming RPCs and deadline, cancellation, or partial-stream status-code behavior is unverified.
Install with skills.sh (any agent)
npx skills add testland/qa --skill grpc-streaming-testsgrpc-streaming-tests
Streaming RPCs need test coverage for deadline behavior, cancellation propagation, flow control under backpressure, and status-code semantics that differ from unary calls - covering all four patterns (Unary, Server-streaming, Client-streaming, Bidirectional-streaming) per the gRPC core concepts docs (opens in new window).
When to use
How to use
Step 1 - Pick the test tool
| Tool | Strength |
|---|---|
Language-native stubs (Go grpc.WithBlock(), Python grpc.aio, Java ManagedChannel) | Unit/integration tests |
grpcurl | Ad-hoc + smoke tests + scripts |
ghz | Load testing + benchmarks (concurrency, RPS) |
mockgrpc / mockery (Go), grpc-mock (Node) | Mock server stubs in unit tests |
Step 2 - Unary RPC sanity (baseline)
Per the gRPC core concepts docs (opens in new window), 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 (opens in new window), 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 (opens in new window), 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 * 1024Step 5 - Bidirectional streaming + ordering
Per the gRPC core concepts docs (opens in new window), 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) >= 3Step 6 - Deadline propagation
Per the gRPC core concepts docs (opens in new window), "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_EXCEEDEDVerify 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 >= 1Step 7 - Cancellation behavior
Per the gRPC core concepts docs (opens in new window), "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 >= 1Status codes, metadata, and load testing
See references/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.
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
| Anti-pattern | Why it fails | Fix |
|---|---|---|
| Skip deadline + cancellation tests | Production cancellation orphans server-side work | Steps 6 + 7 |
| Test only OK and INTERNAL paths | Status-code regressions go silently | Test the matrix (status-codes reference) |
| Use BatchSpanProcessor or similar buffering on test client | Streams "complete" before all messages flush | Always synchronous in tests |
| Tests share a single channel across goroutines | Channel state contamination flakes | Per-test channel |
| Generate proto stubs at test runtime | CI flakes on plugin churn | Generate in build phase + commit |
Limitations
References
gRPC status codes, metadata, and load testing
View source (opens in new window)gRPC status codes, metadata, and load testing
Cross-cutting call semantics beyond the four streaming patterns: the full status-code matrix, request/response metadata round-trip, and load testing with ghz.
Status codes
| Code | When |
|---|---|
| OK | Success |
| CANCELLED | Client cancelled |
| DEADLINE_EXCEEDED | Deadline elapsed |
| INVALID_ARGUMENT | Client error in request |
| UNAUTHENTICATED | No / bad credentials |
| PERMISSION_DENIED | Authenticated but not authorized |
| RESOURCE_EXHAUSTED | Quota / rate limit |
| INTERNAL | Server bug |
| UNAVAILABLE | Server transient unreachable (clients should retry) |
Test the error path returns the right code, not just "an error":
def test_invalid_argument_returns_correct_code():
with grpc.insecure_channel("localhost:50051") as ch:
stub = OrdersStub(ch)
with pytest.raises(grpc.RpcError) as exc:
stub.CreateOrder(OrderRequest(item_count=-1))
assert exc.value.code() == grpc.StatusCode.INVALID_ARGUMENTMetadata
Per the gRPC core concepts docs (opens in new window), metadata is "key-value pairs" case-insensitive ASCII keys; binary values use -bin suffix.
def test_request_metadata_round_trip():
with grpc.insecure_channel("localhost:50051") as ch:
stub = OrdersStub(ch)
metadata = (("x-trace-id", "abc123"),)
resp, call = stub.CreateOrder.with_call(OrderRequest(), metadata=metadata)
# Server reflects request-id in response trailing metadata
trailing = call.trailing_metadata()
assert ("x-trace-id-echo", "abc123") in trailingLoad test with ghz
ghz \
--insecure \
--proto orders.proto \
--call orders.Orders/CreateOrder \
-d '{"item_count":1}' \
-c 50 \
-n 10000 \
localhost:50051Reports RPS, p50/p95/p99 latency. For streaming RPCs use --stream-call-count flag (consult ghz docs).
Related skills
mqtt-tests
Test MQTT v5.0 with Mosquitto broker in CI + paho-mqtt clients - QoS 0 / 1 / 2 delivery semantics, retained messages, Last Will and Testament (LWT), shared subscriptions ($share/group/topic), $SYS topic introspection. Critical for IoT, embedded, and M2M systems where wire-level guarantees matter. Use when a product speaks MQTT on the wire and QoS 1 / 2 redelivery, retained-message state, or LWT behavior needs a broker-backed test - including smoke-testing a new broker auth / ACL / persistence config.
server-sent-events-tests
Test Server-Sent Events (SSE) flows, one-way server-to-client push only (not bidirectional, use websocket-tests for client-to-server messaging): `EventSource` API on the browser side (`onmessage`, `onerror`, `readyState` 0/1/2), event stream format (`data:`, `event:`, `id:`, `retry:`), `Last-Event-ID` reconnect-with-replay header, content-type `text/event-stream`, and HTTP/1.1 connection-pool limits. Use Playwright for browser-side, raw HTTP client for server-side stream tests. Use when a feature pushes updates over `text/event-stream` and the reconnect interval, `Last-Event-ID` replay, or per-origin connection ceiling has no coverage.
sse-load-tests
Load-tests SSE endpoints at scale with k6 - measures concurrent-stream capacity, connection churn, and server memory pressure. Covers the HTTP/1.1 6-connection-per-origin browser ceiling vs HTTP/2 multiplexing, a custom k6 SSE client built on ReadableStream, and threshold gates for TTFB and data throughput. Use when validating whether a server can sustain N concurrent EventSource connections without connection starvation or memory growth.
stomp-amqp-tests
Tests STOMP over WebSocket (Spring, ActiveMQ, RabbitMQ Web STOMP) and AMQP 0-9-1 (RabbitMQ Java client) - frame connect/subscribe/send/ack sequences, ack modes (auto/client/client-individual), exchange and queue declarations, binding routing, Testcontainers RabbitMQ broker, and delivery assertion. Use when validating enterprise Spring or RabbitMQ messaging stacks before deploy.
webhook-replay-tests
Tests inbound webhook receivers for replay-attack resistance: capture incoming webhook payloads + headers, replay against the receiver under test, validate the Standard Webhooks signature scheme (svix-id + svix-timestamp + svix-signature, HMAC-SHA256 over `{id}.{timestamp}.{payload}`), svix-id idempotency dedup, and 5-minute timestamp-window enforcement by signing fixtures at runtime. Does NOT cover outbound delivery, retry-on-5xx, or failure-event exhaustion - those belong to an outbound webhook delivery harness. Use when testing the receiving side of a webhook integration.
websocket-tests
Test WebSocket protocol behavior - opening handshake (HTTP Upgrade with Sec-WebSocket-Key + Sec-WebSocket-Version: 13), control frames (ping 0x9 / pong 0xA / close 0x8), close-frame status codes (1000 normal, 1001 going-away, 1006 abnormal, 1011 server error), subprotocol negotiation, backpressure, and reconnect with jitter. Works with ws (Node), websockets (Python), or Playwright frame inspection per language. Use when a feature holds a long-lived WebSocket open and reconnect, close-code, or backpressure behavior is unverified.