CtrlK
BlogDocsLog inGet started
Tessl Logo

testland/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.

74

Quality

93%

Does it follow best practices?

Run evals on this skill

Adds up to 20 points to the overall score

View guide

SecuritybySnyk

Passed

No findings from the security scan

Overview
Quality
Evals
Security
Files

status-codes-metadata-load.mdreferences/

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

CodeWhen
OKSuccess
CANCELLEDClient cancelled
DEADLINE_EXCEEDEDDeadline elapsed
INVALID_ARGUMENTClient error in request
UNAUTHENTICATEDNo / bad credentials
PERMISSION_DENIEDAuthenticated but not authorized
RESOURCE_EXHAUSTEDQuota / rate limit
INTERNALServer bug
UNAVAILABLEServer 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_ARGUMENT

Metadata

Per the gRPC core concepts docs, 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 trailing

Load test with ghz

ghz \
  --insecure \
  --proto orders.proto \
  --call orders.Orders/CreateOrder \
  -d '{"item_count":1}' \
  -c 50 \
  -n 10000 \
  localhost:50051

Reports RPS, p50/p95/p99 latency. For streaming RPCs use --stream-call-count flag (consult ghz docs).

references

status-codes-metadata-load.md

SKILL.md

tile.json