nownab.log

nownabe's daily posts

PythonのgRPCサーバのテスト

Posted on Feb 25, 2018

概要

PythonのgRPCのテストの方法です。 今回のサンプルコードはこちらにあります。

gRPCサーバのテスト

ググるとE2Eテストがひっかかります。1 つまり実際にサーバを立ち上げて、クライアントからアクセスしてそのレスポンスをチェックする感じです。

gRPCパッケージのテストでそれをやるのは当たり前なんですが、gRPCを使う側としては単体テストでわざわざサーバを立ち上げたくはないですよね。サービスとメッセージを定義したらコードが自動生成されてサービスが受け取るリクエストのメッセージは保証されてるはずなので、リクエストに対するサービスの振る舞いのみをテストしたいところです。

雑に図にするとこんな感じです。

スクリーンショット 2018-02-25 14.42.56.png (684.8 kB)

なので、これを実現する方法を紹介します。

実装

テストするサービスの定義

こんな感じの単純な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を引数として受け取ります。

contextgrpc.ServicerContext2のオブジェクトです3grpc.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_codeset_detailsのみを実装して、エラーコードと詳細のテストができるようにしました。

おわりに

なんかごく普通のことを長々と書いてしまった気がしますが、あまり情報がなかったので。gRPCまわりはまだまだ情報が少ないですね。


  1. How to write a GRPC python unittest - Stack Overflow ↩︎

  2. https://grpc.io/grpc/python/grpc.html#grpc.ServicerContext ↩︎

  3. https://grpc.io/docs/tutorials/basic/python.html#simple-rpc ↩︎

  4. https://grpc.io/grpc/python/grpc.html#grpc.RpcContext ↩︎