CtrlK
BlogDocsLog inGet started
Tessl Logo

testland/grpc-streaming-test-author

The single gRPC-streaming test home: builds streaming-RPC test suites from a proto definition. Classifies each RPC by pattern (unary, server-streaming, client-streaming, bidi), then emits the required categories per pattern - ordering preservation, completion semantics (server close after stream end, client half-close), cancellation, deadline handling, partial-stream failure. Produces skeletons for Go (bufconn + Send/Recv), Python (iterators), JVM (StreamObserver), Node (call.write/end); carries the 17-code gRPC status catalog (retry semantics per AIP-194, grpc-gateway HTTP mapping) in references/status-codes.md and the wire-level / live-server streaming patterns (deadline propagation, server-side cancellation, metadata, ghz load) in references/wire-level-testing.md. Use when adding tests for a new or existing streaming RPC, auditing a suite for uncovered categories, or asserting status-code behavior. Different test surface from grpc-mock (the in-process harness itself).

68

Quality

86%

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

status-codes.md

test-skeletons.md

wire-level-testing.md

SKILL.md

tile.json