grpc的pytest插件
pytest-grpc的Python项目详细描述
pytest grpc
用pytest为grpc编写测试。
exmaple
请参阅示例目录和/或读取“用法”。
用法
例如，您有一些带有rpc声明的proto文件。
syntax="proto3";packagetest.v1;serviceEchoService{rpchandler(EchoRequest)returns(EchoResponse){}}messageEchoRequest{stringname=1;}messageEchoResponse{stringname=1;}
使用grpcio工具编译后，您将得到*\u pb2.py和*\u pb2\u grpc.py文件，现在您可以编写您的服务了。
fromstub.test_pb2importEchoRequest,EchoResponsefromstub.test_pb2_grpcimportEchoServiceServicerclassServicer(EchoServiceServicer):defhandler(self,request:EchoRequest,context)->EchoResponse:returnEchoResponse(name=f'test-{request.name}')deferror_handler(self,request:EchoRequest,context)->EchoResponse:raiseRuntimeError('Some error')
使用存根和服务点pytest：
importpytestfromstub.test_pb2importEchoRequest@pytest.fixture(scope='module')defgrpc_add_to_server():fromstub.test_pb2_grpcimportadd_EchoServiceServicer_to_serverreturnadd_EchoServiceServicer_to_server@pytest.fixture(scope='module')defgrpc_servicer():fromservicerimportServicerreturnServicer()@pytest.fixture(scope='module')defgrpc_stub_cls(grpc_channel):fromstub.test_pb2_grpcimportEchoServiceStubreturnEchoServiceStub
写一点测试：
deftest_some(grpc_stub):request=EchoRequest()response=grpc_stub.handler(request)assertresponse.name==f'test-{request.name}'deftest_example(grpc_stub):request=EchoRequest()response=grpc_stub.error_handler(request)assertresponse.name==f'test-{request.name}'
测试安全服务器
frompathlibimportPathimportpytestimportgrpc@pytest.fixture(scope='module')defgrpc_add_to_server():fromstub.test_pb2_grpcimportadd_EchoServiceServicer_to_serverreturnadd_EchoServiceServicer_to_server@pytest.fixture(scope='module')defgrpc_servicer():fromservicerimportServicerreturnServicer()@pytest.fixture(scope='module')defgrpc_stub_cls(grpc_channel):fromstub.test_pb2_grpcimportEchoServiceStubreturnEchoServiceStub@pytest.fixture(scope='session')defmy_ssl_key_path():returnPath('/path/to/key.pem')@pytest.fixture(scope='session')defmy_ssl_cert_path():returnPath('/path/to/cert.pem')@pytest.fixture(scope='module')defgrpc_server(_grpc_server,grpc_addr,my_ssl_key_path,my_ssl_cert_path):""" Overwrites default `grpc_server` fixture with ssl credentials """credentials=grpc.ssl_server_credentials([(my_ssl_key_path.read_bytes(),my_ssl_cert_path.read_bytes())])_grpc_server.add_secure_port(grpc_addr,server_credentials=credentials)_grpc_server.start()yield_grpc_server_grpc_server.stop(grace=None)@pytest.fixture(scope='module')defmy_channel_ssl_credentials(my_ssl_cert_path):# If we're using self-signed certificate it's necessarily to pass root certificate to channelreturngrpc.ssl_channel_credentials(root_certificates=my_ssl_cert_path.read_bytes())@pytest.fixture(scope='module')defgrpc_channel(my_channel_ssl_credentials,create_channel):""" Overwrites default `grpc_channel` fixture with ssl credentials """withcreate_channel(my_channel_ssl_credentials)aschannel:yieldchannel@pytest.fixture(scope='module')defgrpc_authorized_channel(my_channel_ssl_credentials,create_channel):""" Channel with authorization header passed """grpc_channel_credentials=grpc.access_token_call_credentials("some_token")composite_credentials=grpc.composite_channel_credentials(my_channel_ssl_credentials,grpc_channel_credentials)withcreate_channel(composite_credentials)aschannel:yieldchannel@pytest.fixture(scope='module')defmy_authorized_stub(grpc_stub_cls,grpc_channel):""" Stub with authorized channel """returngrpc_stub_cls(grpc_channel)
在真正的GRPC服务器上运行测试
对在另一线程中工作的read grpc服务器运行测试：
py.test
cachedir: .pytest_cache
plugins: grpc-0.0.0
collected 2 items
example/test_example.py::test_some PASSED
example/test_example.py::test_example FAILED
=================================== FAILURES ====================================
_________________________________ test_example __________________________________
grpc_stub = <stub.test_pb2_grpc.EchoServiceStub object at 0x107a9b390>
def test_example(grpc_stub):
request = EchoRequest()
> response = grpc_stub.error_handler(request)
example/test_example.py:35:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.env/lib/python3.7/site-packages/grpc/_channel.py:547: in __call__
return _end_unary_response_blocking(state, call, False, None)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
state = <grpc._channel._RPCState object at 0x107b263c8>
call = <grpc._cython.cygrpc.SegregatedCall object at 0x107b323c8>
with_call = False, deadline = None
def _end_unary_response_blocking(state, call, with_call, deadline):
if state.code is grpc.StatusCode.OK:
if with_call:
rendezvous = _Rendezvous(state, call, None, deadline)
return state.response, rendezvous
else:
return state.response
else:
> raise _Rendezvous(state, None, None, deadline)
E grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with:
E status = StatusCode.UNKNOWN
E details = "Exception calling application: Some error"
E debug_error_string = "{"created":"@1544451353.148337000","description":"Error received from peer","file":"src/core/lib/surface/call.cc","file_line":1036,"grpc_message":"Exception calling application: Some error","grpc_status":2}"
E >
.env/lib/python3.7/site-packages/grpc/_channel.py:466: _Rendezvous
------------------------------- Captured log call -------------------------------
_server.py 397 ERROR Exception calling application: Some error
Traceback (most recent call last):
File "pytest-grpc/.env/lib/python3.7/site-packages/grpc/_server.py", line 389, in _call_behavior
return behavior(argument, context), True
File "pytest-grpc/example/src/servicer.py", line 10, in error_handler
raise RuntimeError('Some error')
RuntimeError: Some error
================ 1 failed, 1 passed, 1 warnings in 0.16 seconds =================
直接对python代码运行测试
直接调用处理程序，使用伪造的grpc内部组件：
py.test --grpc-fake-server
在这种情况下，您可以直接获得很好的例外：
============================= test session starts =============================
cachedir: .pytest_cache
plugins: grpc-0.0.0
collected 2 items
example/test_example.py::test_some PASSED
example/test_example.py::test_example FAILED
================================== FAILURES ===================================
________________________________ test_example _________________________________
grpc_stub = <stub.test_pb2_grpc.EchoServiceStub object at 0x10e06f518>
def test_example(grpc_stub):
request = EchoRequest()
> response = grpc_stub.error_handler(request)
example/test_example.py:35:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
pytest_grpc/plugin.py:42: in fake_handler
return real_method(request, context)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <servicer.Servicer object at 0x10ce75278>, request =
context = <pytest_grpc.plugin.FakeContext object at 0x10e083e48>
def error_handler(self, request: EchoRequest, context) -> EchoResponse:
> raise RuntimeError('Some error')
E RuntimeError: Some error
example/src/servicer.py:10: RuntimeError
=============== 1 failed, 1 passed, 1 warnings in 0.10 seconds ================
欢迎加入QQ群-->： 979659372
