Test gRPC¶

`pytest-grpc`¶

gRPC can be tested automatically with pytest-grpc.

First, we install

$ pipenv install 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¶

# SPDX-License-Identifier: BSD-3-Clause

from pathlib import Path

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")

See also

pytest fixtures

Afterwards we can write tests, for example:

    return AccountsStub(grpc_channel)


def test_create_account(grpc_stub):
    value = "test-data"
    nl = "\n"

Authentication can also be tested, for example with:

# SPDX-FileCopyrightText: 2021 Veit Schiele
#

    assert response.name == f"test-{request.name}"


@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)

Afterwards we can test against a real gRPC server with:

$ pipenv run pytest --fixtures tests/

or directly against the 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/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`¶