### gRPC Project Setup and Testing Commands Source: https://github.com/kataev/pytest-grpc/blob/master/example/README.md Provides a sequence of shell commands to set up a gRPC development environment, compile protocol buffers, and execute tests using pytest-grpc, including an option for a fake gRPC server. ```bash pip install grpcio grpcio-tools python -m grpc_tools.protoc -Isrc --python_out=src/stub --grpc_python_out=src/stub src/test.proto PYTHONPATH=src py.test # or PYTHONPATH=src py.test --grpc-fake-server ``` -------------------------------- ### Define Pytest Fixtures for gRPC Service Setup Source: https://github.com/kataev/pytest-grpc/blob/master/README.md This snippet defines essential pytest fixtures for setting up a gRPC testing environment. `grpc_add_to_server` provides the function to add the servicer to the gRPC server, `grpc_servicer` instantiates the service implementation, and `grpc_stub_cls` provides the gRPC client stub class for making RPC calls in tests. ```python import pytest from stub.test_pb2 import EchoRequest @pytest.fixture(scope='module') def grpc_add_to_server(): from stub.test_pb2_grpc import add_EchoServiceServicer_to_server return add_EchoServiceServicer_to_server @pytest.fixture(scope='module') def grpc_servicer(): from servicer import Servicer return Servicer() @pytest.fixture(scope='module') def grpc_stub_cls(grpc_channel): from stub.test_pb2_grpc import EchoServiceStub return EchoServiceStub ``` -------------------------------- ### Execute pytest-grpc with fake gRPC internals Source: https://github.com/kataev/pytest-grpc/blob/master/README.md This example shows how to run `pytest-grpc` tests using the `--grpc-fake-server` flag. This mode bypasses the real gRPC communication layer, allowing direct calls to Python handlers and providing clearer, direct Python exceptions instead of gRPC-specific errors, which is useful for debugging application logic. ```bash py.test --grpc-fake-server ``` ```text ============================= 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 = 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 = , request = context = 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 ================ ``` -------------------------------- ### Define Pytest Fixtures for Secure gRPC Server Testing Source: https://github.com/kataev/pytest-grpc/blob/master/README.md This comprehensive set of pytest fixtures extends the testing setup to include secure gRPC communication using SSL/TLS. It defines fixtures for SSL key/certificate paths, custom `grpc_server` and `grpc_channel` fixtures with SSL credentials, and fixtures for authorized channels and stubs, enabling robust testing of secure and authenticated gRPC services. ```python from pathlib import Path import pytest import grpc @pytest.fixture(scope='module') def grpc_add_to_server(): from stub.test_pb2_grpc import add_EchoServiceServicer_to_server return add_EchoServiceServicer_to_server @pytest.fixture(scope='module') def grpc_servicer(): from servicer import Servicer return Servicer() @pytest.fixture(scope='module') def grpc_stub_cls(grpc_channel): from stub.test_pb2_grpc import EchoServiceStub return EchoServiceStub @pytest.fixture(scope='session') def my_ssl_key_path(): return Path('/path/to/key.pem') @pytest.fixture(scope='session') def my_ssl_cert_path(): return Path('/path/to/cert.pem') @pytest.fixture(scope='module') def grpc_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') def my_channel_ssl_credentials(my_ssl_cert_path): # If we're using self-signed certificate it's necessarily to pass root certificate to channel return grpc.ssl_channel_credentials( root_certificates=my_ssl_cert_path.read_bytes() ) @pytest.fixture(scope='module') def grpc_channel(my_channel_ssl_credentials, create_channel): """ Overwrites default `grpc_channel` fixture with ssl credentials """ with create_channel(my_channel_ssl_credentials) as channel: yield channel @pytest.fixture(scope='module') def grpc_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 ) with create_channel(composite_credentials) as channel: yield channel @pytest.fixture(scope='module') def my_authorized_stub(grpc_stub_cls, grpc_channel): """ Stub with authorized channel """ return grpc_stub_cls(grpc_channel) ``` -------------------------------- ### Write Basic Pytest gRPC Client Tests Source: https://github.com/kataev/pytest-grpc/blob/master/README.md These pytest functions demonstrate how to write basic tests for a gRPC service using the `grpc_stub` fixture. They show how to make RPC calls, pass requests, and assert on the responses, including a test for an error handling scenario to ensure proper behavior. ```python def test_some(grpc_stub): request = EchoRequest() response = grpc_stub.handler(request) assert response.name == f'test-{request.name}' def test_example(grpc_stub): request = EchoRequest() response = grpc_stub.error_handler(request) assert response.name == f'test-{request.name}' ``` -------------------------------- ### Define gRPC Service and Messages in Protobuf Source: https://github.com/kataev/pytest-grpc/blob/master/README.md This snippet defines a gRPC service `EchoService` with a `handler` RPC method, and corresponding `EchoRequest` and `EchoResponse` messages using Protocol Buffers (proto3 syntax). This forms the contract for the gRPC communication between client and server. ```proto syntax = "proto3"; package test.v1; service EchoService { rpc handler(EchoRequest) returns (EchoResponse) { } } message EchoRequest { string name = 1; } message EchoResponse { string name = 1; } ``` -------------------------------- ### Implement gRPC Service in Python Source: https://github.com/kataev/pytest-grpc/blob/master/README.md This Python class `Servicer` implements the `EchoServiceServicer` generated from the proto file. It provides a `handler` method that processes `EchoRequest` and returns `EchoResponse`, demonstrating basic request handling. An `error_handler` is also included to show how to raise exceptions within a gRPC method. ```python from stub.test_pb2 import EchoRequest, EchoResponse from stub.test_pb2_grpc import EchoServiceServicer class Servicer(EchoServiceServicer): def handler(self, request: EchoRequest, context) -> EchoResponse: return EchoResponse(name=f'test-{request.name}') def error_handler(self, request: EchoRequest, context) -> EchoResponse: raise RuntimeError('Some error') ``` -------------------------------- ### Execute pytest-grpc against a real gRPC server Source: https://github.com/kataev/pytest-grpc/blob/master/README.md This snippet demonstrates running `pytest-grpc` tests against a gRPC server operating in a separate thread. It illustrates the typical output, including a `grpc._channel._Rendezvous` error, which indicates an issue within the gRPC communication layer when an application exception occurs. ```bash py.test ``` ```text 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 = 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 = call = 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 ================ ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.