- Регистрация
- 23 Август 2023
- Сообщения
- 3 641
- Лучшие ответы
- 0
- Реакции
- 0
- Баллы
- 243
Offline
Добрый день, утро, вечер или ночь. Меня зовут Константин, я тестировщик, занимаюсь написанием авто-тестов на Python и в данной статье опишу пример тестирования gRPC и подготовки авто-тестов на примере программного обеспечения для сбора, обработки и передачи данных в системах промышленной автоматизации.
Что будет описано в статье
Первое, что нам понадобится, - это само тестируемое приложения в его роли будет выступать «Эликонт‑КС» версии 2.8 или выше (производитель предоставляет возможность пользоваться данным ПО любому пользователю в демо режиме, за что выражаю им свою благодарность) скачать можно тут.
Немного о том что же такое «Эликонт‑КС» — программное обеспечение для сбора, обработки и передачи данных в системах промышленной автоматизации и интернета вещей (IIoT), выступающее в роли преобразователя промышленных протоколов, концентратора данных, коммуникационного шлюза и мультисервера.
Разработчики данного ПО в версии 2.8 реализовали "Пользовательский протокол" который и даст нам возможность поработать с этим приложением по gRPC. Написать методы для того чтобы передать и собрать из него данные и адаптировать их для автоматизации тестирования.
После скачивания и установки запускаем данное приложение и конфигурируем тестовый проект (это будет gRPC сервер с которым мы и будем работать).
Создание проекта в «Эликонт‑КС»:
Создание проекта в «Эликонт-КС»
Создание проекта в «Эликонт-КС»
Создание проекта в «Эликонт-КС»
Настройка окружения для работы с gRPC
Предварительно я создал небольшой проект со следующей структурой:
Структура проекта
Создаем .proto файл
Создадим директорию для проекта и ".proto" файл для этого зайдем в приложение "Конфигуратор" и выберем там пункт меню "Справка" > "Показать справку" откроется локальная страничка с документацией где в п.п 6.2.11. мы можем найти ссылку "elecont.proto" перейдя по которой получим структуру файла, скопируем все содержимое далее создадим файл с расширением .proto (например elecont.proto) и поместим в него все содержимое данной ссылки.
О том что такое .proto файл
Файл с расширением .proto используется в протоколе Protobuf (Protocol Buffers). Это инструмент для сериализации структурированных данных, разработанный Google. Файлы формата .proto содержат описания структуры данных (сообщений), используемых приложениями для передачи данных между различными системами и языками программирования.
Пример .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:
print(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]"
}
}
Разберем полученную структуру:
На этом наш первый запрос готов.
Примечание: достаточно часто я сталкивался с проблемой когда 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): Тип сигнала (используя значения перечисления ElecontSignalType).
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:
print(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:
print(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)}")
# Устанавливаем значение сигнала для каждого GUID
for guid in guides:
try:
# Устанавливаем новое значение сигнала
set_signal.set_signal(ip_user_channel_client, guid, '77')
except Exception as exc:
print(f"Ошибка обновления сигнала с GUID={guid}: {exc}")
except Exception as general_exc:
print(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 сервер, а также вы визуально можете наблюдать за передачей данных в приложении.
Также в данном приложении вы можете реализовать свой собственный протокол для сбора\обработки\передачи данных и передавать ваши данные в поддерживаемые приложением промышленные протоколы связи.
На этом все, спасибо за внимание, надеюсь данная статья была полезна для вас.
Что будет описано в статье
Как настроить тестируемое приложение (ведь оно будет нашим gRPC сервером);
Как настроить окружение для работы с gRPC;
Примеры gRPC запросов (GetAllSignals, GetSignalsByGuid, SetSignal);
Пример авто-теста Pytest.
Ссылка на Github с описанным в статье примером
Первое, что нам понадобится, - это само тестируемое приложения в его роли будет выступать «Эликонт‑КС» версии 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 содержат описания структуры данных (сообщений), используемых приложениями для передачи данных между различными системами и языками программирования.
Пример .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:
print(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): Тип сигнала (используя значения перечисления ElecontSignalType).
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:
print(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:
print(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)}")
# Устанавливаем значение сигнала для каждого GUID
for guid in guides:
try:
# Устанавливаем новое значение сигнала
set_signal.set_signal(ip_user_channel_client, guid, '77')
except Exception as exc:
print(f"Ошибка обновления сигнала с GUID={guid}: {exc}")
except Exception as general_exc:
print(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 сервер, а также вы визуально можете наблюдать за передачей данных в приложении.
Также в данном приложении вы можете реализовать свой собственный протокол для сбора\обработки\передачи данных и передавать ваши данные в поддерживаемые приложением промышленные протоколы связи.
На этом все, спасибо за внимание, надеюсь данная статья была полезна для вас.
