grpc-mock
Overview
Mocking a gRPC server lets client-side tests exercise success
paths, every grpc.StatusCode (per the status-code catalog in
grpc-streaming-test-author, references/status-codes.md),
timeouts, and streaming sequences without a real backend. The
in-process harness is also the substrate for interceptor tests -
see references/interceptors.md.
Three approaches dominate, picked by language:
| Approach | Mechanism |
|---|
| In-process gRPC server | A real grpc.Server listens on an in-memory transport (bufconn in Go, InProcessServerBuilder in JVM). Tests exercise the full client stack. |
| Interface mock | mockgen / gomock (Go) / Mockito (JVM) / unittest.mock (Python) replace the generated client stub with a programmable mock. Faster but skips marshalling. |
| Standalone mock server | Run a tool like grpcmock / dishwasher as a subprocess. Cross-language client testing. |
When to use
- Client-side tests need controllable gRPC responses (success +
error matrix).
- Streaming-RPC tests need a server that sends a deterministic
sequence.
- Mid-test failure injection (deadline / cancellation / status
code).
- Tests must not depend on a running backend (offline CI).
- Cross-service contract testing: want the client to see what
the contract says, not what the current server happens to do.
Authoring
Go: bufconn + in-process server
Per
pkg.go.dev/google.golang.org/grpc/test/bufconn,
bufconn.Listener is the canonical in-memory transport:
package myservice_test
import (
"context"
"net"
"testing"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/status"
"google.golang.org/grpc/test/bufconn"
pb "example.com/proto"
)
const bufSize = 1024 * 1024
type fakeServer struct {
pb.UnimplementedUserServiceServer
nextResponse *pb.User
nextErr error
}
func (f *fakeServer) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
if f.nextErr != nil {
return nil, f.nextErr
}
return f.nextResponse, nil
}
func setupClient(t *testing.T, fake *fakeServer) pb.UserServiceClient {
lis := bufconn.Listen(bufSize)
s := grpc.NewServer()
pb.RegisterUserServiceServer(s, fake)
go func() { _ = s.Serve(lis) }()
t.Cleanup(func() { s.Stop() })
conn, err := grpc.DialContext(context.Background(), "bufnet",
grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
return lis.Dial()
}),
grpc.WithTransportCredentials(insecure.NewCredentials()),
)
if err != nil { t.Fatal(err) }
t.Cleanup(func() { conn.Close() })
return pb.NewUserServiceClient(conn)
}
func TestGetUser_NotFound(t *testing.T) {
fake := &fakeServer{
nextErr: status.Error(codes.NotFound, "user does not exist"),
}
client := setupClient(t, fake)
_, err := client.GetUser(context.Background(), &pb.GetUserRequest{Id: "missing"})
st, _ := status.FromError(err)
if st.Code() != codes.NotFound {
t.Fatalf("got %v, want NotFound", st.Code())
}
}
Per the status-code catalog in grpc-streaming-test-author
(references/status-codes.md):
assert on status.Code(), not on error message strings.
Go: gomock / mockgen (interface mock)
For tests that don't need the marshalling/transport stack:
go install go.uber.org/mock/mockgen@latest
mockgen -source=gen/user_grpc.pb.go -destination=mocks/user_mock.go
import (
"testing"
"go.uber.org/mock/gomock"
pb "example.com/proto"
mocks "example.com/mocks"
)
func TestServiceWithMockClient(t *testing.T) {
ctrl := gomock.NewController(t)
mockClient := mocks.NewMockUserServiceClient(ctrl)
mockClient.EXPECT().
GetUser(gomock.Any(), gomock.Eq(&pb.GetUserRequest{Id: "u1"})).
Return(&pb.User{Id: "u1", Name: "Alice"}, nil)
// Test the code that uses mockClient ...
}
Tradeoff: doesn't exercise serialisation; faster, less fidelity.
Python: in-process server + pytest fixture
import grpc
import pytest
from concurrent import futures
from user_pb2 import User, GetUserRequest
from user_pb2_grpc import UserServiceServicer, add_UserServiceServicer_to_server, UserServiceStub
class FakeUserService(UserServiceServicer):
next_response = None
next_status = None
def GetUser(self, request, context):
if self.next_status is not None:
context.abort(self.next_status, "fake error")
return self.next_response
@pytest.fixture
def fake_service():
return FakeUserService()
@pytest.fixture
def grpc_channel(fake_service):
server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
add_UserServiceServicer_to_server(fake_service, server)
port = server.add_insecure_port("[::]:0")
server.start()
channel = grpc.insecure_channel(f"localhost:{port}")
yield channel
server.stop(grace=0)
def test_get_user_not_found(fake_service, grpc_channel):
fake_service.next_status = grpc.StatusCode.NOT_FOUND
stub = UserServiceStub(grpc_channel)
with pytest.raises(grpc.RpcError) as exc:
stub.GetUser(GetUserRequest(id="missing"))
assert exc.value.code() == grpc.StatusCode.NOT_FOUND
server.add_insecure_port("[::]:0") lets the OS pick a free
port - important for parallel test execution.
Python: unittest.mock patching of stub
from unittest.mock import patch, MagicMock
import grpc
def test_service_with_mock_stub():
with patch("myapp.user_pb2_grpc.UserServiceStub") as MockStub:
instance = MockStub.return_value
instance.GetUser.return_value = User(id="u1", name="Alice")
# Test the code that uses UserServiceStub ...
JVM: InProcessServerBuilder
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.testing.GrpcCleanupRule;
@Rule public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
@Test
public void getUser_notFound() throws Exception {
String serverName = InProcessServerBuilder.generateName();
grpcCleanup.register(InProcessServerBuilder
.forName(serverName)
.directExecutor()
.addService(new UserServiceGrpc.UserServiceImplBase() {
@Override
public void getUser(GetUserRequest req, StreamObserver<User> obs) {
obs.onError(Status.NOT_FOUND
.withDescription("user does not exist")
.asRuntimeException());
}
})
.build()
.start());
UserServiceGrpc.UserServiceBlockingStub stub = UserServiceGrpc.newBlockingStub(
grpcCleanup.register(InProcessChannelBuilder
.forName(serverName)
.directExecutor()
.build()));
StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
() -> stub.getUser(GetUserRequest.newBuilder().setId("missing").build()));
assertEquals(Status.Code.NOT_FOUND, e.getStatus().getCode());
}
Node / TypeScript: @grpc/grpc-js + port 0
import * as grpc from "@grpc/grpc-js";
import { UserServiceService } from "./generated/user_grpc_pb";
function createServer(handlers: Partial<UserServiceServer>) {
const server = new grpc.Server();
server.addService(UserServiceService, handlers);
return new Promise<{ port: number; server: grpc.Server }>((resolve, reject) => {
server.bindAsync("127.0.0.1:0", grpc.ServerCredentials.createInsecure(), (err, port) => {
if (err) return reject(err);
server.start();
resolve({ port, server });
});
});
}
test("GetUser returns NOT_FOUND", async () => {
const { port, server } = await createServer({
getUser: (_call, callback) => {
callback({ code: grpc.status.NOT_FOUND, details: "user does not exist" });
},
});
const client = new UserServiceClient(`localhost:${port}`, grpc.credentials.createInsecure());
await expect(() => promisify(client.getUser.bind(client))({ id: "missing" }))
.rejects.toMatchObject({ code: grpc.status.NOT_FOUND });
server.forceShutdown();
});
Running
These tests run as ordinary unit tests:
go test ./... # Go
pytest tests/ # Python
mvn test # JVM
npm test # Node
Per-language test runners; no separate harness needed.
Parsing results
Test failures point to:
- Wrong status code - fix server-side error mapping or
test expectation per the status-code catalog in
grpc-streaming-test-author (references/status-codes.md).
- Marshalling errors - proto definition / generated code
drift; regenerate via
buf-cli-lint-breaking-build.
- Timeout - server not responding; fake handler hung;
check
context.WithTimeout usage.
CI integration
jobs:
unit-tests-with-grpc-mocks:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v5
- uses: actions/setup-go@v5
- run: go test ./... -race -timeout=60s
-race is critical for mock-server tests - concurrent client +
server goroutines often surface races.
Anti-patterns
| Anti-pattern | Why it fails | Fix |
|---|
| Asserting on error message strings | Brittle to i18n / wording | Assert on status.Code() |
Hard-coded ports (8080) in tests | Port conflicts in parallel CI | Use bufconn (Go), [::]:0 (Python), InProcessChannel (JVM), port 0 (Node) |
| Sharing one mock server across tests | Test order matters; flaky | Per-test setup; t.Cleanup / fixture teardown |
| Mocking gRPC stub without server registration | Tests skip codec, marshalling, error mapping | In-process server preferred over interface mock for service-level tests |
Returning a Go error directly (not status.Error) | Client sees Code: Unknown | Always wrap with status.Errorf(codes.X, "...") |
| Mocking streaming methods with one response | Tests don't exercise multi-message logic | Use a real stream + Send multiple times |
Forgetting server.Stop() in teardown | Goroutine leaks; future tests pollute | t.Cleanup / pytest fixture yield |
No -race flag in Go tests | Concurrent races slip through | Always go test -race in CI |
Limitations
- No wire-level fault injection in-process. For partial-byte
cutoffs or middlebox-induced errors, use a real network +
toxiproxy / tc.
- bufconn / InProcessServer skip TLS. Tests that exercise
TLS-specific code paths need a real server.
- Streaming-test ergonomics differ per language. Bidi
streaming with deterministic interleavings is fiddly
everywhere; see
grpc-streaming-test-author.
- Mock-stub approach loses error-mapping fidelity. A real
status.Errorf wraps differently than a hand-constructed
Go error.
- Doesn't replace contract tests. Mocks reflect what this
test expects; they don't enforce that the real server actually
produces those responses. Pair with
protobuf-compat-checking (in the qa-contract-testing plugin).
References