概要
PythonのgRPCのテストの方法です。 今回のサンプルコードはこちらにあります。
gRPCサーバのテスト
ググるとE2Eテストがひっかかります。1 つまり実際にサーバを立ち上げて、クライアントからアクセスしてそのレスポンスをチェックする感じです。
gRPCパッケージのテストでそれをやるのは当たり前なんですが、gRPCを使う側としては単体テストでわざわざサーバを立ち上げたくはないですよね。サービスとメッセージを定義したらコードが自動生成されてサービスが受け取るリクエストのメッセージは保証されてるはずなので、リクエストに対するサービスの振る舞いのみをテストしたいところです。
雑に図にするとこんな感じです。
なので、これを実現する方法を紹介します。
実装
テストするサービスの定義
こんな感じの単純なEchoサービスを考えます。
syntax = "proto3";
service Echo {
rpc Echo (Request) returns (Response) {}
}
message Request {
string message = 1;
}
message Response {
string message = 1;
}
コードを生成するにはこんな感じです。
python -m grpc_tools.protoc -I=. --python_out=. --grpc_python_out=. echo.proto
サービスの実装
サービスの実装はこんな感じです。今回のEchoサービスには禁止ワードの場合エラーを返す機能がついています。
import grpc
import echo_pb2
import echo_pb2_grpc
_BANNED_WORDS = ["unko"]
class Servicer(echo_pb2_grpc.EchoServicer):
def __init__(self):
pass
def Echo(self, request, context):
if request.message in _BANNED_WORDS:
context.set_code(grpc.StatusCode.OUT_OF_RANGE)
context.set_details(f"{request.message} is a prohibited word.")
raise Exception(f"'{request.message}' is a prohibited word.")
return echo_pb2.Response(message=request.message)
エラーコードがOUT_OF_RANGEなのは気にしないでください。
このサービスのクラスのみを単体テストするのが目標です。
サーバ/クライアントの実装
サーバとクライアントの実装です。gRPCのサンプルから持ってきただけです。
from concurrent import futures
import time
import grpc
import echo_pb2_grpc
from servicer import Servicer
_ONE_DAY_IN_SECONDS = 60 * 60 * 24
def serve():
servicer = Servicer()
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
echo_pb2_grpc.add_EchoServicer_to_server(servicer, server)
server.add_insecure_port("[::]:50051")
server.start()
try:
while True:
time.sleep(_ONE_DAY_IN_SECONDS)
except KeyboardInterrupt:
server.stop(0)
if __name__ == "__main__":
serve()
import grpc
import grpc
import echo_pb2
import echo_pb2_grpc
def run():
channel = grpc.insecure_channel("[::]:50051")
stub = echo_pb2_grpc.EchoStub(channel)
message = "Hello, world!"
response = stub.Echo(echo_pb2.Request(message=message))
print(f"Response: {response.message}")
if __name__ == "__main__":
run()
実際に動かすにはpython server.pyでサーバを起動して、別のターミナルなどでpython client.pyを実行します。こんな感じですね。
$ python server.py & sleep 1 && python client.py
[1] 76052
Response: Hello, world!
これらのサーバとクライアントのコードを単体テストでは使わないことが目標になります。
サービスの引数
再掲しますがサービスの本体はこれです。
class Servicer(echo_pb2_grpc.EchoServicer):
def Echo(self, request, context):
if request.message in _BANNED_WORDS:
context.set_code(grpc.StatusCode.OUT_OF_RANGE)
context.set_details(f"{request.message} is a prohibited word.")
raise Exception(f"'{request.message}' is a prohibited word.")
return echo_pb2.Response(message=request.message)
サービスのメソッドはprotoで定義されたリクエストメッセージであるrequestと、contextを引数として受け取ります。
contextはgrpc.ServicerContext2のオブジェクトです3がgrpc.ServicerContextは抽象クラスであり、その実装は公開されていません。
メソッド内でcontextを使う場合はcontextを渡す必要がありますが、公開されていないので自前で実装するひつようがあります。本家のテストでもgrpc.ServicerContextを実装したテスト用のクラスを作っています。
また、その場合はgrpc.ServicerContextは親抽象クラスのgprc.RpcContext4を継承しているので、そちらのメソッドも実装する必要があります。
テストの実装
以上を踏まえEchoサービスの単体テストを書くとこんな感じになります。
import unittest
import grpc
import echo_pb2
from servicer import Servicer
class ServicerContext(grpc.ServicerContext):
def __init__(self):
self.code = None
self.details = None
def add_callback(self):
pass
def cancel(self):
pass
def is_active(self):
pass
def time_remaining(self):
pass
def abort(self, code, details):
pass
def auth_context(self):
pass
def invocation_metadata(self):
pass
def peer(self):
pass
def peer_identities(self):
pass
def peer_identity_key(self):
pass
def send_initial_metadata(self, initial_metadata):
pass
def set_code(self, code):
self.code = code
def set_details(self, details):
self.details = details
def set_trailing_metadata(self, trailing_metadata):
pass
class TestServicer(unittest.TestCase):
def test_Servicer(self):
servicer = Servicer()
request = echo_pb2.Request(message="Test Message")
context = ServicerContext()
response = servicer.Echo(request, context)
self.assertEqual(echo_pb2.Response(message="Test Message"), response)
def test_Servicer_error(self):
servicer = Servicer()
request = echo_pb2.Request(message="unko")
context = ServicerContext()
with self.assertRaises(Exception):
servicer.Echo(request, context)
self.assertEqual(grpc.StatusCode.OUT_OF_RANGE, context.code)
self.assertEqual("unko is a prohibited word.", context.details)
if __name__ == "__main__":
unittest.main()
ServicerContextは必要に応じてメソッドを実装すればいいと思います。今回は
set_codeとset_detailsのみを実装して、エラーコードと詳細のテストができるようにしました。
おわりに
なんかごく普通のことを長々と書いてしまった気がしますが、あまり情報がなかったので。gRPCまわりはまだまだ情報が少ないですね。