gRPC testen#
pytest-grpc
#
gRPC lässt sich automatisiert testen z.B. mit pytest-grpc.
Zunächst installieren wir pytest-grpc:
$ pipenv install pytest-grpc Installing pytest-grpc… Adding pytest-grpc to Pipfile's [packages]… ✔ Installation Succeeded …
Dann erstellen wir für unser gRPC-Beispiel ein Test Fixture mit:
tests/test_accounts.py#import grpc import pytest from accounts_pb2 import CreateAccountRequest, GetAccountsRequest @pytest.fixture(scope="module") def grpc_add_to_server(): from accounts_pb2_grpc import add_AccountsServicer_to_server return add_AccountsServicer_to_server @pytest.fixture(scope="module") def grpc_servicer(): from accounts_server import AccountsService return AccountsService() @pytest.fixture(scope="module") def grpc_stub(grpc_channel): from accounts_pb2_grpc import AccountsStub return AccountsStub(grpc_channel)
Siehe auch
Anschließend können wir Tests schreiben, z.B.:
def test_create_account(grpc_stub): value = "test-data" nl = "\n" request = CreateAccountRequest(account_name=value) response = grpc_stub.CreateAccount(request) assert ( f"{response.account}" == f'account_id: 1{nl}account_name: "test-data"{nl}' )
Auch die Authentifizierung lässt sich testen, z.B. mit:
from pathlib import Path @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)
Anschließend können wir gegen einen realen gRPC-Server testen mit:
$ pipenv run pytest --fixtures tests/
oder direkt gegen den Python-Code:
$ pipenv run pytest --fixtures tests/ --grpc-fake-server ============================= test session starts ============================== platform darwin -- Python 3.7.3, pytest-6.2.2, py-1.10.0, pluggy-0.13.1 rootdir: /Users/veit/cusy/trn/jupyter-tutorial/docs/data/grpc plugins: grpc-0.8.0 collected 2 items tests/test_accounts.py .F [100%] …
Wireshark#
Wireshark ist ein Open-Source-Tool zur Analyse von Netzwerkprotokollen. Im Folgenden zeigen wir Euch, wie ihr den gRPC- und den Protobuf-Dissectors verwenden könnt. Sie erleichtern Euch das Zerlegen (Dekodieren) von gRPC-Nachrichten, die im Protobuf- oder JSON-Format serialisiert sind. Zudem könnt ihr damit das Server-, Client- und bidirektionales gRPC-Streaming analysieren.
Bemerkung
Üblicherweise kann Wireshark nur gRPC-Messages im Klartext analysieren. Für das Sezieren von TLS-Session benötigt Wireshark den geheimen Schlüssel, deren Export jedoch zum heutigen Zeitpunkt nur Go gRPC unterstützt [1].
Siehe auch