Skip to main content

Тести мікросервісів

gRPC - це відкритий фреймворк від Google, який дозволяє здійснювати високоефективне взаємодію між серверами та клієнтами. Тестування gRPC сервісів вимагає ретельного підходу, який забезпечує достовірність та надійність комунікації між різними компонентами. Нижче представлений підхід до створення тестів для сервісів gRPC, базуючись на Python та фреймворку pytest

Алгоритм створення тестів​

Усі тести для gRPC створюються мовою програмування Python та фреймворка pytest.

Алгоритм створення тестів виглядає наступним чином:

  1. Створення фікстури для середовища тестування
  2. Створення функцій тестування

Структура тесту​

Так, як gRPC потребує стабільного з'єднання з сервером - то для тестування, ми імітуємо потрібне середовище за рахунок створення фікстур. Операція створення фікстури - це створення власного локального середовища виконання з наперед створеними компонентами, які потрібні для стабільної роботи сервісу. Також тестування сервісу вимагає активного сервісу, який наразі працює на localhost:4003.

Для створення фікстур фреймворк pytest надає можливість використання декоратора pytest.fixture.

Для коректної роботи тестів, у реалізації будь-якого тесту для gRPC повинні бути наступні фікстури:

  • grpc_channel - опис каналу роботи gRPC
  • grpc_stub_cls - опис stub-функції для gRPC сервера
  • grpc_servicer - сервісер для сервісу gRPC
  • grpc_stub - функція приєднання сервісу до певного каналу
  • grpc_add_to_server - функція запуску сервісу

Опис реалізації даних фікстур на прикладі сервісу конвертації даних наведено нижче

@pytest.fixture(scope='module')
def grpc_channel():
    with grpc.insecure_channel('localhost:50051') as channel:
        yield channel

Після створення усіх потрібних фікстур, створюються функції тестування.

Приклад функції тестування:

def test_csvToXls(grpc_stub):
    with open('test_files/test.csv', 'rb') as f:
        buffer = f.read()
        b64_string = base64.b64encode(buffer)
    request = convertp_pb2.csvToXlsParams(buffer=b64_string, header="TEST", subheader="test", separator=",")
    response = grpc_stub.csvToXls(request)
    assert response is not None

Ключове слово assert визначає умову перевірки правильності результату виконання функції. У прикладі, який наведено вище, перевіркою на правильність є прпосто отримання результату відмінного від None

💡 Tip

Для задання більше ніж однієї умови, можна використати логічні оператори and, or та not. Така комбінація дає можливість перевіряти не тільки наявність відповіді, а ще й її коректність і багато чого ще.

❗️ Warning

У мові програмування Python, ключове слово assert використовується для встановлення умови, яка має бути істинною в певній точці програми. Якщо вираз після assert є хибним, Python викидає виняток AssertionError. З цього випливає, що використання двох ключових слів assert недопустимо

Запуск тестування​

Запуск усіх тестів:

pytest path/to/test_file.py

Запуск конкретного тесту:

pytest path/to/test_file.py::test_name

Результат тестування​

Приклад результату тестування при виявлені помилки:

test.py::test_csvToXls FAILED                                            [  7%]
test.py:29 (test_csvToXls)
grpc_stub = <convertp_pb2_grpc.ConvertStub object at 0x00000194F9EE3D40>

    def test_csvToXls(grpc_stub):
        with open('test_files/test.csv', 'rb') as f:
            buffer = f.read()
            b64_string = base64.b64encode(buffer)
        request = convertp_pb2.csvToXlsParams(buffer=b64_string, header="TEST", subheader="test", separator=";")
>       response = grpc_stub.csvToXls(request)

test.py:35: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv\Lib\site-packages\grpc\_channel.py:1160: in __call__
    return _end_unary_response_blocking(state, call, False, None)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

state = <grpc._channel._RPCState object at 0x00000194F9F061E0>
call = <grpc._cython.cygrpc.SegregatedCall object at 0x00000194F9F03E40>
with_call = False, deadline = None

    def _end_unary_response_blocking(
        state: _RPCState,
        call: cygrpc.SegregatedCall,
        with_call: bool,
        deadline: Optional[float],
    ) -> Union[ResponseType, Tuple[ResponseType, grpc.Call]]:
        if state.code is grpc.StatusCode.OK:
            if with_call:
                rendezvous = _MultiThreadedRendezvous(state, call, None, deadline)
                return state.response, rendezvous
            else:
                return state.response
        else:
>           raise _InactiveRpcError(state)  # pytype: disable=not-instantiable
E           grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
E            status = StatusCode.UNKNOWN
E            details = "Exception calling application: {"funcs": "PrintMap", "error": "JSONDecodeError('Extra data: line 1 column 3 (char 2)')", "request": "buffer: \"77u/VGVzdDtuYW1lDQoxOzENCjE7MQ0KMTsxDQoxOzENCg==\"\nheader: \"TEST\"\nsubheader: \"test\"\nseparator: \";\"\n"}"
E            debug_error_string = "UNKNOWN:Error received from peer  {grpc_message:"Exception calling application: {\"funcs\": \"PrintMap\", \"error\": \"JSONDecodeError(\'Extra data: line 1 column 3 (char 2)\')\", \"request\": \"buffer: \\\"77u/VGVzdDtuYW1lDQoxOzENCjE7MQ0KMTsxDQoxOzENCg==\\\"\\nheader: \\\"TEST\\\"\\nsubheader: \\\"test\\\"\\nseparator: \\\";\\\"\\n\"}", grpc_status:2, created_time:"2023-12-12T16:53:40.0589021+00:00"}"
E           >

venv\Lib\site-packages\grpc\_channel.py:1003: _InactiveRpcError

Приклад результату тестування без виявлення помилок:

test.py::test_htmlToImage PASSED                                         [ 61%]

Корисні посилання​