Добрый день, утро, вечер или ночь. Меня зовут Константин, я тестировщик, занимаюсь написанием авто-тестов на Python и в данной статье опишу пример тестирования gRPC и подготовки авто-тестов на примере программного обеспечения для сбора, обработки и передачи данных в системах промышленной автоматизации.
Что будет описано в статье
Как настроить тестируемое приложение (ведь оно будет нашим gRPC сервером);
Как настроить окружение для работы с gRPC;
Примеры gRPC запросов (GetAllSignals, GetSignalsByGuid, SetSignal);
Пример авто-теста Pytest.
Подготовка тестируемого приложения
Первое, что нам понадобится, - это само тестируемое приложения в его роли будет выступать «Эликонт‑КС» версии 2.8 или выше (производитель предоставляет возможность пользоваться данным ПО любому пользователю в демо режиме, за что выражаю им свою благодарность) скачать можно тут.
Немного о том что же такое «Эликонт‑КС» — программное обеспечение для сбора, обработки и передачи данных в системах промышленной автоматизации и интернета вещей (IIoT), выступающее в роли преобразователя промышленных протоколов, концентратора данных, коммуникационного шлюза и мультисервера.
Разработчики данного ПО в версии 2.8 реализовали "Пользовательский протокол" который и даст нам возможность поработать с этим приложением по gRPC. Написать методы для того чтобы передать и собрать из него данные и адаптировать их для автоматизации тестирования.
После скачивания и установки запускаем данное приложение и конфигурируем тестовый проект (это будет gRPC сервер с которым мы и будем работать).
Создание проекта в «Эликонт‑КС»:
Выбрать меню «Проект»;
«Создать проект»;
«Локальный проект»;
Задать имя проекта. Я напишу «gRPC», после чего в окне конфигурация у вас появится ярлык вашего проекта;
Далее правым кликом мыши по проекту вызываем модальное окно и добавляем «Коммуникационный сервер»;
Также как и в предыдущем шаге вызываем модальное окно и добавляем «User channel клиент» и «User channel сервер»;
Для удобства на уровне «Канал» у обоих протоколов выставляем IP адрес «127.0.0.1», порт можно оставить как есть или поменять на удобный вам (учтите это в будущем);
Теперь в протоколе «User channel клиент» перейдем на уровень сигналы и с помощью кнопки в рабочей области такой большой плюс в кружочке добавим сигнал и выберем у этого сигнала тип данных «int16»;
Теперь перейдем в протокол «User channel сервер» на уровень сигналы, после того как вы создали сигнал в клиенте в рабочей области серверного канала появится этот сигнал готовый для публикации (если данный тип данных поддерживается этим каналом) с помощью Drag‑and‑drop перетащите сигнал в верхнюю часть рабочей области;
Теперь перейдем на уровень «КС» и с помощью кнопки в рабочей области или в модальном окне «Загрузить конфигурацию» загрузим проект.
Куда загружается проект? Данное приложение имеет 2 компонента «Конфигуратор» в котором мы создаем проект и «Коммуникационный сервер» который работает с этим проектом и поднимет нам gRPC сервер. На этом наш проект готов к работе.
Настройка окружения для работы с gRPC
Предварительно я создал небольшой проект со следующей структурой:
В директории config я разместил конфигурационный файл «Эликонт‑КС»;
В директории proto будет размешен .proto файл и файлы сгенерированные на его основе;
В директории tests будут размешены тесты;
В директории utils будут размещены методы для работы с gRPC.
Создаем .proto файл
Создадим директорию для проекта и ".proto" файл для этого зайдем в приложение "Конфигуратор" и выберем там пункт меню "Справка" > "Показать справку" откроется локальная страничка с документацией где в п.п 6.2.11. мы можем найти ссылку "elecont.proto" перейдя по которой получим структуру файла, скопируем все содержимое далее создадим файл с расширением .proto (например elecont.proto) и поместим в него все содержимое данной ссылки.
О том что такое .proto файл
Файл с расширением .proto используется в протоколе Protobuf (Protocol Buffers). Это инструмент для сериализации структурированных данных, разработанный Google. Файлы формата .proto содержат описания структуры данных (сообщений), используемых приложениями для передачи данных между различными системами и языками программирования.
Теперь необходимо установить библиотеку Protocol Buffers (Protobuf) и генератор для gRPC на Python для этого выполните указанную ниже команду.
pip install grpcio-tools protobuf
Возможные проблемы:
Если Python установлен, но pip не виден, скорее всего, проблема в настройках переменных среды. Так как я часто сталкивался с такой проблемой как отсутствие пути к библиотеке в переменной Path (как на работе, так и дома) здесь я кратко опишу возможный способ ее решения.
Добавляем Python в переменную Path:
Добавьте в системную переменную Path пути к каталогу куда установлен Python "C:\Users<Ваш_Пользователь>\AppData\Local\Programs\Python\Python3X" и подкаталоги который содержит исполняемые файлы, включая pip "C:\Users<Ваш_Пользователь>\AppData\Local\Programs\Python\Python3X\Scripts"
После установки библиотек необходимо сгенерировать gRPC код выполнив команду
python -m grpc_tools.protoc --python_out=. --grpc_python_out=. --proto_path=. elecont.proto
После выполнения данной команды у вас сгенерируется 2 файла "elecont_pb2_grpc.py" и "elecont_pb2.py" которые будут содержать Python классы для работы с сообщениями определенными в ".proto" файле.
Первый запрос
В этой главе описан модуль содержащий запрос к gRPC серверу "GetAllSignals" (что стоит проверить перед началом то что "Эликонт-КС" запущена и конфигурация загружена, у вас уже есть .proto файл и 2 файла с сгенерированным кодом на Python "elecont_pb2_grpc.py" и "elecont_pb2.py")
В .proto файле мы можем найти описание первого метода с которым мы будем работать
/** * Получить пул всех сигналов из текущей конфигурации КС. * Для больших конфигураций с большим количеством сигналов этот вызов может быть ресурсоёмким, поэтому его рекомендуется его использовать с осторожностью */ rpc GetAllSignals (Empty) returns (SignalPool) {}
В описании нам сообщают что данный метод вернет нам пул всех сигналов из канала к которому мы будем обращаться.
Далее приведен пример кода для вызова данного метода и вывода в консоль полученных данных (данный код я разместил в файле "get_all_signals.py" в директории utils).
""" Модуль "get_all_signals.py" для демонстрации запроса к gRPC-серверу с целью получения сигнала. Используется протокол буферов Protobuf и библиотека gRPC. """ # Импортируем библиотеку gRPC import grpc # Импортируем сгенерированные файлы from proto import elecont_pb2, elecont_pb2_grpc def get_all_signals(ip_address_and_port): """ Осуществляет подключение к gRPC-серверу и запрашивает список всех сигналов. Parameters: ip_address_and_port (str): IP-адрес и порт, по которым доступен gRPC-сервер. """ try: # Создаем соединение с сервером channel = grpc.insecure_channel(ip_address_and_port) stub = elecont_pb2_grpc.ElecontStub(channel) # Передаем пустой объект Empty в качестве аргумента empty_request = elecont_pb2.Empty() # Вызываем метод GetAllSignals response = stub.GetAllSignals(empty_request) # Возвращаем полученные данные return response # Обработка ошибок except grpc.RpcError as e: raise grpc.RpcError(f'gRPC ошибка: {e.details()}') # IP-адрес и порт, по которым мы работаем ip_user_channel_client = '127.0.0.1:29041' # Запускаем получение сигналов response_get_all_signals = get_all_signals(ip_user_channel_client) # Вывод в консоль полученного результата print(response_get_all_signals)
Пример вывода данных:
int16_signal { sigprop { id: 2 quality: 66 raw_quality: 66 source_id: 9 guid: "b6ae1b69-faae-4464-b93b-5a961f485287" str_quality: "Invalid [Failure]" } }
Разберем полученную структуру:
int16_signal: Тип данных сигнала который мы сконфигурировали
sigprop: Внутренняя группа свойств сигнала.
id: Идентификационный номер сигнала.
quality: Качество сигнала в числовом выражении.
raw_quality: Исходное (не обработанное) значение качества сигнала
source_id: Источник сигнала (идентификатор устройства с которого поступил сигнал).
guid: Глобальный уникальный идентификатор (GUID), позволяющий однозначно идентифицировать сигнал среди множества аналогичных сигналов.
str_quality: Строковое представление текущего состояния качества сигнала.
На этом наш первый запрос готов.
Примечание: достаточно часто я сталкивался с проблемой когда Python не видит модули, например "elecont_pb2_grpc.py" и "elecont_pb2.py" решение этой проблемы - записать путь до директорий в которой находятся эти файлы в переменную "PYTHONPATH"
Запрос для отправки данных
В этой главе описан модуль содержащий запросы "SetSignal" (отправка данных на сервер) и "GetSignalByGuid" (получение данных по guid сигнала).
Описание данных запросов из .proto файла
/** * Задать сигнал. Запрос должен содержать сообщение Signal с GUID, который существует в текущей конфигурации КС. Если сигнал с таким GUID не будет обраружен, то запрос выполнен не будет. * Является обегчённой версией функции SetSignals. Принимает в качестве запроса простое по структуре сообщение Signal */ rpc SetSignal (Signal) returns (Result) {} /** * Получить сигнал по заданному GUID. Если сигнала с заданными GUID не окажется в текущей конфигурации КС, то вернётся сообщение Signal с данными по умолчанию. * Является обегчённой версией функции GetSignalsByGuid. Возвращает простое по структуре сообщение Signal */ rpc GetSignalByGuid (Guid) returns (Signal) {}
Ниже приведен пример кода для передачи данных на сервер и проверки валидности переданного значения методами "SetSignal" и "GetSignalByGuid". В результате получилась проверяющая сама себя функция, которая отправляет данные и проверяет верные ли данные пришли на сервер (данный код я разместил в файле "set_signal.py" в директории utils).
В результате выполнения данной функции в приложении "Конфигуратор" отобразятся отправленные вами данные в режиме "Исполнения".
Итоговый модуль "set_signal.py"
""" Модуль "set_signal.py" для демонстрации запроса к gRPC-серверу с целью отправки сигнала. Используется протокол буферов Protobuf и библиотека gRPC. """ # Импортируем библиотеку gRPC для осуществления коммуникаций с удалённым сервером import grpc # Импортируем сгенерированные файлы на основе протокола буферов (ProtoBuf) from proto import elecont_pb2, elecont_pb2_grpc # Стандартная библиотека Python для работы со временем import time def set_signal(ip_address_and_port, guid, value): """Отправляет сигнал на удалённый сервер с указанными параметрами. Параметры: ip_address_and_port (str): Адрес и порт сервера (пример: '127.0.0.1:29041'). guid (str): Уникальный идентификатор сигнала. quality (int): Показатель качества сигнала. timestamp (int): Временная отметка в миллисекундах. type_valye (int or enum): Тип сигнала. value (str): Текущее значение сигнала. str_quality (str): Строковое представление качества сигнала. """ # Получаем текущее время в миллисекундах timestamp_ms = int(time.time() * 1000) try: # Устанавливаем соединение с удалённым сервером по указанному адресу и порту channel = grpc.insecure_channel(ip_address_and_port) stub = elecont_pb2_grpc.ElecontStub(channel) # Создаём объект Signal и заполняем его необходимыми данными request_signal = elecont_pb2.Signal() request_signal.guid = guid # Идентификатор сигнала request_signal.quality = 0 # Показатель качества (В текущей реализации всегда 0 - Good) request_signal.time = timestamp_ms # Временная метка в миллисекундах request_signal.type.value = elecont_pb2.ElecontSignalType.INT16 # Тип сигнала (В текущей реализации всегда int16) request_signal.value = value # Значение сигнала request_signal.str_quality = "GOOD" # Строковое описание качества (В текущей реализации всегда Good) # Выполняем запрос на сервер, вызывая метод SetSignal stub.SetSignal(request_signal) # Получаем сигнал по GUID и сравниваем значение response_get_signal_by_guid = stub.GetSignalByGuid(elecont_pb2.Guid(guid=guid)) # Проверяем совпадение значения сигнала if value == response_get_signal_by_guid.value: # Сообщаем об успешной отправке print(f"Сигнал с GUID={guid} успешно обновлён.") else: # Если отправленное значение не совпадает со значением в приложение вызываем ошибку raise Exception("Произошла ошибка, значения сигнала не совпадают!") # Обработка ошибок except grpc.RpcError as e: raise Exception(f'gRPC ошибка: {e.details()}')
Перед тем как переходить к написанию авто-теста реализуем функцию для получения Guid из всех доступных сигналов в канале в файле "get_all_signals.py" а также уберем все вызовы функции.
Итоговый модуль "get_all_signals.py"
""" Модуль "get_all_signals.py" для демонстрации запроса к gRPC-серверу с целью получения сигнала. Используется протокол буферов Protobuf и библиотека gRPC. """ # Импортируем библиотеку gRPC import grpc # Импортируем сгенерированные файлы from proto import elecont_pb2, elecont_pb2_grpc # Импортируем библиотеку для преобразования объектов Protobuf в обычные Python-словари формата JSON. from google.protobuf.json_format import MessageToDict def get_all_signals(ip_address_and_port): """ Осуществляет подключение к gRPC-серверу и запрашивает список всех сигналов. Parameters: ip_address_and_port (str): IP-адрес и порт, по которым доступен gRPC-сервер. """ try: # Создаем соединение с сервером channel = grpc.insecure_channel(ip_address_and_port) stub = elecont_pb2_grpc.ElecontStub(channel) # Передаем пустой объект Empty в качестве аргумента empty_request = elecont_pb2.Empty() # Вызываем метод GetAllSignals response = stub.GetAllSignals(empty_request) # Возвращаем полученные данные return response # Обработка ошибок except grpc.RpcError as e: print(f'gRPC ошибка: {e.details()}') # функция для получения все GUID def get_guids(signals_pool): """ Извлекает все уникальные идентификаторы (GUID) из различных типов сигналов. Параметры: signals_pool: Объект Protobuf, содержащий различные типы сигналов. Возвращаемое значение: list: Список уникальных идентификаторов (GUID) всех сигналов, содержащихся в переданном объекте. """ try: # Преобразуем объект Protobuf в словарь Python signals_data = MessageToDict(signals_pool) # Список для гуидов guides = [] # Сбор всех GUID проходим по каждому типу сигнала for signal_type in signals_data.keys(): # keys() вернут названия ключей ('booleanSignal', 'int16Signal', и т.д "Тестируемое приложение подерживает 13 типов сигналов") # Получаем список сигналов для текущего типа signals_list = signals_data.get(signal_type) # Проходим по каждому сигналу и добавляем GUID в список for signal in signals_list: guides.append(signal['sigprop']['guid']) # Возвращаем все GUID return guides except Exception as ex: raise Exception(f"Произошла непредвиденная ошибка: {ex}")
Пишем авто-тест
Установите Pytest (Pytest — фреймворк тестирования для языка программирования Python) командой:
pip install pytest
Далее в папке "tests" создайте файл "test_signal.py" и поместите в него приведенный ниже код.
Данный тест демонстрирует подключение к gRPC серверу, сбор данных и передачу данных.
""" Модуль "test_signal.py" для демонстрации примера авто-теста. """ from utils import get_all_signals, set_signal def test_signals(): """ Тестирует работу с сигналами типа данных int16: получает все сигналы сервера, выделяет их GUID и устанавливает новое значение сигнала. Эта функция демонстрирует последовательность шагов: 1. Подключение к серверу и получение всех сигналов. 2. Извлечение всех GUID из полученных сигналов. 3. Установку нового значения для каждого сигнала. """ ip_user_channel_client = '127.0.0.1:29041' try: # Получаем все сигналы с сервера all_signals = get_all_signals.get_all_signals(ip_user_channel_client) # Выделяем все GUID из сигналов guides = get_all_signals.get_guids(all_signals) # Сообщаем о количестве GUID print(f"Количество GUID: {len(guides)}") # Устанавливаем значение для каждого сигнала for guid in guides: try: # Устанавливаем новое значение сигнала set_signal.set_signal(ip_user_channel_client, guid, '79') except Exception as exc: raise Exception(f"Ошибка обновления сигнала с GUID={guid}: {exc}") except Exception as general_exc: raise Exception(f"Возникла общая ошибка: {general_exc}")
Команда для запуска
pytest -s
Результатом выполнения данной команды будет обновление значения сигналов и метки времени у сигналов в приложении «Эликонт‑КС».
Пример вывода в консоль после выполнения команды запуска.
platform win32 -- Python 3.13.7, pytest-8.4.2, pluggy-1.6.0 rootdir: C:\Проекты\grpc plugins: allure-pytest-2.15.0, anyio-4.12.0, asyncio-1.3.0, base-url-2.1.0, ordering-0.6, playwright-0.7.1, rerunfailures-16.1 asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function collected 1 item tests\test_signal.py Количество GUID: 5 Сигнал с GUID=bc0c92fd-f30e-41d9-84fe-660fcb7756b9 успешно обновлён. Сигнал с GUID=378a25d0-3140-4ce6-b48c-737582b237b3 успешно обновлён. Сигнал с GUID=67542890-3788-4b66-aafe-a96e6ce6856c успешно обновлён. Сигнал с GUID=d027b3bf-b3ff-41e0-96f0-b134709df0a2 успешно обновлён. Сигнал с GUID=078f096f-7b6b-4ee8-86d8-7b91377fb65f успешно обновлён. .
Заключение
В данной статье продемонстрирован пример простого авто-теста gRPC и несколько примеров удалённых процедурных вызовов (Remote Procedure Calls). Данное тестируемое приложение достаточно удобно чтобы практиковаться с работой по gRPC, т.к оно предоставляет готовый gRPC сервер, а также вы визуально можете наблюдать за передачей данных в приложении.
Также в данном приложении вы можете реализовать свой собственный протокол для сбора\обработки\передачи данных и передавать ваши данные в поддерживаемые приложением промышленные протоколы связи.
На этом все, спасибо за внимание, надеюсь данная статья была полезна для вас.
