Test gRPC¶

`pytest-grpc`¶

gRPC can be tested automatically with pytest-grpc.

First, we install

$ uv add pytest-grpc
Installing pytest-grpc…
Adding pytest-grpc to Pipfile's [packages]…
✔ Installation Succeeded
…

Then we create a Test Fixture for our gRPC-Example with:

tests/test_accounts.py¶

import accounts_server
import grpc
import pytest


@pytest.fixture(scope="module")
def grpc_add_to_server():
    return add_AccountsServicer_to_server


@pytest.fixture(scope="module")
def grpc_servicer():
    return accounts_server.AccountsService()


@pytest.fixture(scope="module")
def grpc_stub(grpc_channel):
    return AccountsStub(grpc_channel)

See also

pytest fixtures

Afterwards we can write tests, for example:

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}'
    )

Authentication can also be tested, for example with:

from accounts_pb2 import CreateAccountRequest, GetAccountsRequest
from accounts_pb2_grpc import AccountsStub, add_AccountsServicer_to_server


@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):
    """For self-signed certificate root certificate in channel are required."""
    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)

Afterwards we can test against a real gRPC server with:

$ uv run pytest --fixtures tests/

or directly against the Python code:

$ uv 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/Python4DataScience/docs/data/grpc
plugins: grpc-0.8.0
collected 2 items

tests/test_accounts.py .F                                                [100%]
…

Wireshark¶

Wireshark is an open source tool for analysing network protocols. In the following, we will show you how to use the gRPC and Protobuf dissectors. They make it easier for you to decode gRPC messages that are serialised in Protobuf or JSON format. You can also use them to analyse server, client and bidirectional gRPC streaming.

Note

Usually, Wireshark can only analyse gRPC messages in plain text. For dissecting a TLS session, Wireshark needs the secret key, the export of which is currently only supported by Go gRPC [1].

Test gRPC¶

pytest-grpc¶

Wireshark¶

`pytest-grpc`¶