gRPC Testing — Unit Tests, Integration Tests, and Mock Servers for Services
In this tutorial, you will learn about grpc testing. We cover key concepts, practical examples, and best practices to help you master this topic.
gRPC testing validates service implementations, client-server interactions, error handling, and streaming behavior through unit tests with mocks, integration tests with real servers, and end-to-end test suites.
What You'll Learn
- Unit testing gRPC service implementations
- Creating mock gRPC servers for client testing
- Testing unary, server-streaming, client-streaming, and bidirectional RPCs
- Testing error handling and edge cases
- Integration Testing with real gRPC servers
Why It Matters
gRPC services handle complex streaming logic, error propagation, and metadata that are difficult to test manually. Automated tests catch regressions in resolver behavior, timeout handling, and error formatting before they reach production. DodaTech's Durga Antivirus Pro runs 500+ gRPC unit tests on every commit, covering all service methods and error conditions.
Real-World Use
A developer adds a new field to the ThreatReport protobuf message. The unit test for the threat analysis service catches that the new field isn't being populated in the response. The fix takes 2 minutes instead of a production incident.
flowchart TB
A["Test Suite"] --> B["Unit Tests\n(service logic)"]
A --> C["Integration Tests\n(real server)"]
A --> D["Client Tests\n(mock server)"]
B --> E["Test Resolvers\nin isolation"]
C --> F["Test full\nrequest flow"]
D --> G["Test client\nretry/timeout"]
E --> H["Fast (<10ms)"]
F --> I["Medium (<500ms)"]
G --> J["Slow (<5s)"]
style A fill:#dbeafe,stroke:#2563eb
Code Examples
Example 1: Unit Testing a Service in Go
package main
import (
"context"
"testing"
"google.golang.org/grpc"
"google.golang.org/grpc/test/bufconn"
)
func TestReportThreat(t *testing.T) {
// Create in-memory listener
lis := bufconn.Listen(1024 * 1024)
s := grpc.NewServer()
pb.RegisterThreatServiceServer(s, &server{
db: newMockDB(),
})
go func() {
if err := s.Serve(lis); err != nil {
t.Fatalf("Server failed: %v", err)
}
}()
defer s.Stop()
// Create client connected to in-memory server
conn, _ := grpc.Dial("bufnet",
grpc.WithContextDialer(
func(ctx context.Context, s string) (net.Conn, error) {
return lis.Dial()
},
),
grpc.WithInsecure(),
)
defer conn.Close()
client := pb.NewThreatServiceClient(conn)
// Test successful threat report
resp, err := client.ReportThreat(ctx, &pb.ThreatRequest{
Name: "TestMalware",
Severity: pb.Severity_CRITICAL,
DeviceId: "dev-001",
})
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
if resp.Status != pb.ThreatStatus_QUARANTINED {
t.Errorf("Expected QUARANTINED, got %v", resp.Status)
}
}
Example 2: Mock Server for Client Testing in Python
import grpc
import unittest
from concurrent import futures
from unittest.mock import Mock, MagicMock
class MockThreatService(grpc_health_v1.ThreatServiceServicer):
def __init__(self):
self.reported_threats = []
self.should_fail = False
self.fail_with = grpc.StatusCode.UNAVAILABLE
def ReportThreat(self, request, context):
if self.should_fail:
context.abort(self.fail_with, "Simulated failure")
self.reported_threats.append(request)
return pb.ThreatResponse(
status=pb.ThreatStatus.QUARANTINED,
threat_id="threat-001",
)
def StreamThreats(self, request, context):
threats = [
pb.ThreatAlert(name="Malware-A", severity=pb.Severity.HIGH),
pb.ThreatAlert(name="Malware-B", severity=pb.Severity.MEDIUM),
]
for threat in threats:
yield threat
class TestThreatClient(unittest.TestCase):
def setUp(self):
self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
self.mock_service = MockThreatService()
pb.add_ThreatServiceServicer_to_server(self.mock_service, self.server)
self.port = self.server.add_insecure_port("[::]:0")
self.server.start()
self.channel = grpc.insecure_channel(f"[::]:{self.port}")
self.client = pb.ThreatServiceStub(self.channel)
def tearDown(self):
self.server.stop(None)
def test_report_threat_success(self):
request = pb.ThreatRequest(name="Test", severity=pb.Severity.HIGH)
response = self.client.ReportThreat(request)
self.assertEqual(response.status, pb.ThreatStatus.QUARANTINED)
def test_report_threat_retry_on_unavailable(self):
self.mock_service.should_fail = True
self.mock_service.fail_with = grpc.StatusCode.UNAVAILABLE
with self.assertRaises(grpc.RpcError) as context:
self.client.ReportThreat(
pb.ThreatRequest(name="Test"),
timeout=1,
)
self.assertEqual(
context.exception.code(),
grpc.StatusCode.UNAVAILABLE,
)
Example 3: Testing Streaming RPCs in Node.js
const grpc = require('@grpc/grpc-js');
const protoLoader = require('@grpc/proto-loader');
const assert = require('assert');
class MockServer {
start() {
this.server = new grpc.Server();
this.server.addService(proto.ThreatService.service, {
StreamThreats: (call) => {
// Simulate streaming threats
const threats = [
{ name: 'Threat-1', severity: 'HIGH' },
{ name: 'Threat-2', severity: 'MEDIUM' },
];
threats.forEach(t => call.write(t));
call.end();
},
ReportThreat: (call, callback) => {
callback(null, { status: 'QUARANTINED', threatId: 't-1' });
},
});
this.server.bindAsync('0.0.0.0:0', grpc.ServerCredentials.createInsecure(),
(err, port) => {
this.port = port;
this.server.start();
}
);
}
stop() {
this.server.forceShutdown();
}
}
// Test streaming
it('should receive all streamed threats', (done) => {
const mock = new MockServer();
mock.start();
const client = new proto.ThreatServiceClient(
`localhost:${mock.port}`,
grpc.credentials.createInsecure(),
);
const call = client.StreamThreats({ deviceId: 'dev-1' });
const received = [];
call.on('data', (threat) => received.push(threat));
call.on('end', () => {
assert.strictEqual(received.length, 2);
assert.strictEqual(received[0].name, 'Threat-1');
mock.stop();
done();
});
});
Common Mistakes
- Not testing error paths — tests should cover every gRPC status code the service can return, including INVALID_ARGUMENT, NOT_FOUND, and INTERNAL.
- Using real servers in unit tests — unit tests should use in-memory transport (bufconn) to avoid port conflicts and slow startup times.
- Forgetting to test streaming — streaming RPCs have complex lifecycle events (data, end, error, status). Test all of them.
- Not testing deadline propagation — clients should test that deadlines are propagated correctly through middleware and interceptors.
- Testing only the happy path — test edge cases: empty requests, very large messages, concurrent calls, and canceled contexts.
Practice Questions
- What is the advantage of using bufconn for gRPC tests?
- How do you test a server-streaming RPC?
- How do you simulate error responses in a mock server?
- Why should you test concurrent gRPC calls?
- How do you test deadline and timeout behavior?
Challenge: Write a comprehensive test suite for a gRPC bidirectional streaming service that tests: normal message exchange, client disconnection mid-stream, server error mid-stream, message ordering, and concurrent stream handling.
Mini Project
Build a gRPC testing framework with: in-memory test server, mock client and server stubs for all RPC types (unary, server-stream, client-stream, bidirectional), error simulation helpers, deadline testing utilities, and performance benchmarks for Serialization and streaming.
FAQ
What's Next
Learn about gRPC interceptors
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro