Загрузка данных
stationary_status_scenarios.py
"""
Сценарии тестов - функции-обёртки без
pytest маркеров.
Каждая функция содержит логику одного теста.
Pytest маркеры и allure декораторы применяются в тестовых файлах.
"""
import time
from collections import defaultdict
from datetime import datetime, timedelta
import allure
import pytest
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.enums import (
ConfirmationStatus,
Direction,
ExportedDataType,
ExportStatus,
GravityPipe,
LdsStatus,
LeakStatus,
MessagePriority,
MessageType,
ReplyStatus,
ReservedType,
SignalType,
SiteKpKp,
StationaryStatus,
UserActions,
)
from constants.test_constants import BaseTN3Constants as TestConst
from constants.test_constants import ExportLdsStatusReportConstants as LdsReportConst
from constants.test_constants import ExportMtModeReportConstants as MtReportConst
from constants.test_constants import ExportReportConstants as ReportConst
from models.get_messages_model import Filtering, FilteringObjects, Pagination
from test_config.models_for_tests import (
BaseSuiteConfig,
CaseData,
ExportLdsStatusReportState,
ExportLeaksReportState,
ExportMtModeReportState,
LDSStatusConfig,
LeakTestConfig,
SmokeSuiteConfig,
)
from utils.helpers import lds_status_report_xlsx_utils as lds_report_utils
from utils.helpers import mt_mode_report_xlsx_utils as mt_report_utils
from utils.helpers import report_xlsx_utils as report_utils
from utils.helpers import ws_test_utils as t_utils
from utils.helpers.asserts import SoftAssertions, StepCheck
from utils.helpers.ws_message_parser import ws_message_parser as parser
from utils.helpers.ws_test_utils import get_value
def basic_info(http_client, cfg: BaseSuiteConfig):
"""
Проверка базовой информации СОУ: список ТУ.
"""
with allure.step("Http запрос, получение и обработка ответа типа: BasicInfoContent"):
expected_tu = [(cfg.tu_id, cfg.tu_name)]
response = http_client.post_request(HttpConst.GET_BASIC_INFO_URL_PATH, {})
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_basic_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
tus = getattr(getattr(parsed_payload.replyContent, 'basicInfo', None), 'tus', [])
actual_tu = [(tu.tuId, tu.tuName) for tu in tus if tu.tuId == cfg.tu_id]
StepCheck("Проверка наличия данных с базовой информацией СОУ", "tus").actual(actual_tu).is_not_none()
with allure.step(f"Поверка наличия {cfg.tu_name} в списке доступных ТУ на сервере"):
# Критическая проверка: если нужного ТУ нет в BasicInfoContent — считаем что ТУ отключен (через Zookeeper)
# и прерываем весь прогон.
if expected_tu[0] not in actual_tu:
msg = (
f"ТУ отключен: в BasicInfoContent отсутствует ТУ для запущенного набора данных: "
f"tuId={cfg.tu_id}, tuName='{cfg.tu_name}', suite={cfg.suite_name}. "
f"Необходимо убедиться, что ТУ включен (Zookeeper) и перезапустить прогон."
)
allure.attach(
f"Ожидаемый ТУ: {expected_tu}\nПолученные ТУ: {actual_tu}",
name="Предварительная проверка: ТУ отключен",
attachment_type=allure.attachment_type.TEXT,
)
pytest.fail(msg, pytrace=False)
with SoftAssertions() as soft_failures:
StepCheck("Проверка наличия объектов в списке ТУ", "tus", soft_failures).actual(
parsed_payload.replyContent.basicInfo.tus
).is_not_empty()
StepCheck(
f"Проверка наличия ТУ: {cfg.tu_name} в списке ТУ ",
"(tuId, tuName)",
soft_failures,
).actual(
actual_tu
).expected(expected_tu).equal_to()
def messages_exist_in_journal(http_client):
"""
Проверка наличия сообщений в журнале.
"""
with allure.step("Http запрос, получение и обработка ответа типа: MessagesInfoContent"):
request_body = t_utils.create_journal_req_body()
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
async def lds_status_initialization(ws_client, cfg: SmokeSuiteConfig):
"""
Проверка режима работы СОУ: Инициализация.
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"CommonSchemeContent",
"SubscribeCommonSchemeRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
parsed_payload = parser.parse_common_scheme_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
# Получает список участков карты течения
flow_areas = getattr(parsed_payload.replyContent, 'flowAreas', [])
# Получает самый протяженный участок карты течения
longest_flow_area = t_utils.get_longest_flow_area(flow_areas)
# Получает список ДУ
diagnostic_areas = getattr(longest_flow_area, 'diagnosticAreas', [])
StepCheck("Проверка наличия данных диагностических участков", "diagnosticAreas").actual(
diagnostic_areas
).is_not_empty()
allure.attach(
f"Самый протяженный участок карты течений: {longest_flow_area}",
name="flowArea. Инициализация",
attachment_type=allure.attachment_type.TEXT,
)
# Получает коллекцию статусов списка ДУ
lds_status_set = {diagnostic_area.ldsStatus for diagnostic_area in diagnostic_areas}
# Определяет режим работы СОУ по приоритету
lds_status_int = t_utils.determine_lds_status_by_priority(lds_status_set)
lds_status = LdsStatus(lds_status_int) if lds_status_int else None
StepCheck("Проверка режима работы СОУ", "ldsStatus").actual(lds_status).expected(
LdsStatus.INITIALIZATION
).equal_to()
async def diagnostics_of_signals_after_initialization(ws_client, cfg: SmokeSuiteConfig, test_data: CaseData):
"""
Проверка выходных сигналов после окончания режима Инициализация по причине "холодного" пуска СОУ.
"""
# Распаковка данных для теста
exp_first_site_reg_lu, exp_first_site_reg_sou = test_data.expected_result.get("exp_tixoreczkaya_novovelichkovskaya")
exp_second_site_reg_lu, exp_second_site_reg_sou = test_data.expected_result.get("exp_novovelichkovskaya_krymskaya")
exp_third_site_reg_lu, exp_third_site_reg_sou = test_data.expected_result.get("exp_krymskaya_grushovaya")
exp_fourth_site_reg_lu, exp_fourth_site_reg_sou = test_data.expected_result.get("exp_backup_route_bejsug")
exp_fifth_site_reg_lu, exp_fifth_site_reg_sou = test_data.expected_result.get("exp_backup_route_ponura")
exp_sixth_site_reg_lu, exp_sixth_site_reg_sou = test_data.expected_result.get("exp_backup_route_kuban")
exp_seventh_site_reg_lu, exp_seventh_site_reg_sou = test_data.expected_result.get("exp_npz_afipskij")
exp_eighth_site_reg_lu, exp_eighth_site_reg_sou = test_data.expected_result.get("exp_npz_ilinskij")
with allure.step("Подключение по ws, получение и обработка сообщения типа: OutputSignalsInfo"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"OutputSignalsInfo",
"SubscribeOutputSignalsRequest",
{
'objects': {
'linearParts': [],
'controlledSites': [
SiteKpKp.TIXORECZKAYA_NOVOVELICHKOVSKAYA.value,
SiteKpKp.NOVOVELICHKOVSKAYA_KRYMSKAYA.value,
SiteKpKp.KRYMSKAYA_GRUSHOVAYA.value,
SiteKpKp.BACKUP_ROUTE_BEJSUG.value,
SiteKpKp.BACKUP_ROUTE_PONURA.value,
SiteKpKp.BACKUP_ROUTE_KUBAN.value,
SiteKpKp.NPZ_AFIPSKIJ.value,
SiteKpKp.NPZ_ILINSKIJ.value,
],
},
'signalTypes': 1023,
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_output_signals_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
controlled_site_dict = {
"controlled_site_first": SiteKpKp.TIXORECZKAYA_NOVOVELICHKOVSKAYA.value,
"controlled_site_second": SiteKpKp.NOVOVELICHKOVSKAYA_KRYMSKAYA.value,
"controlled_site_third": SiteKpKp.KRYMSKAYA_GRUSHOVAYA.value,
"controlled_site_fourth": SiteKpKp.BACKUP_ROUTE_BEJSUG.value,
"controlled_site_fifth": SiteKpKp.BACKUP_ROUTE_PONURA.value,
"controlled_site_sixth": SiteKpKp.BACKUP_ROUTE_KUBAN.value,
"controlled_site_seventh": SiteKpKp.NPZ_AFIPSKIJ.value,
"controlled_site_eight": SiteKpKp.NPZ_ILINSKIJ.value,
}
controlled_site_messages = {}
for name, key in controlled_site_dict.items():
controlled_site_messages[name] = t_utils.find_object_by_a_few_fields(
parsed_payload.replyContent.controlledSiteSignals, key
)
all_signals = {}
for site_name, site_message in controlled_site_messages.items():
signal_dict = {'pump': None, 'sou': None, 'gravity': None}
if site_message:
all_signals[site_name] = {
'pump': t_utils.get_signal(site_message, SignalType.REGLU),
'sou': t_utils.get_signal(site_message, SignalType.REGSOU),
'gravity': t_utils.get_signal(site_message, SignalType.GRAVITYPIPE),
}
else:
all_signals[site_name] = signal_dict
first_kp_kp = all_signals.get("controlled_site_first") or {}
if first_kp_kp:
first_site_signal_pump = (
StationaryStatus(int(get_value(first_kp_kp.get("pump")))) if first_kp_kp.get("pump") else None
)
first_site_signal_sou = (
LdsStatus(int(get_value(first_kp_kp.get("sou")))) if first_kp_kp.get("sou") else None
)
first_site_signal_gravity = get_value(first_kp_kp.get("gravity"))
second_kp_kp = all_signals.get("controlled_site_second") or {}
if second_kp_kp:
second_site_signal_pump = (
StationaryStatus(int(get_value(second_kp_kp.get("pump")))) if second_kp_kp.get("pump") else None
)
second_site_signal_sou = (
LdsStatus(int(get_value(second_kp_kp.get("sou")))) if second_kp_kp.get("sou") else None
)
second_site_signal_gravity = get_value(second_kp_kp.get("gravity"))
third_kp_kp = all_signals.get("controlled_site_third") or {}
if third_kp_kp:
third_site_signal_pump = (
StationaryStatus(int(get_value(third_kp_kp.get("pump")))) if third_kp_kp.get("pump") else None
)
third_site_signal_sou = (
LdsStatus(int(get_value(third_kp_kp.get("sou")))) if third_kp_kp.get("sou") else None
)
third_site_signal_gravity = get_value(third_kp_kp.get("gravity"))
fourth_kp_kp = all_signals.get("controlled_site_fourth") or {}
if fourth_kp_kp:
fourth_site_signal_pump = (
StationaryStatus(int(get_value(fourth_kp_kp.get("pump")))) if fourth_kp_kp.get("pump") else None
)
fourth_site_signal_sou = (
LdsStatus(int(get_value(fourth_kp_kp.get("sou")))) if fourth_kp_kp.get("sou") else None
)
fourth_site_signal_gravity = get_value(fourth_kp_kp.get("gravity"))
fifth_kp_kp = all_signals.get("controlled_site_fifth") or {}
if fifth_kp_kp:
fifth_site_signal_pump = (
StationaryStatus(int(get_value(fifth_kp_kp.get("pump")))) if fifth_kp_kp.get("pump") else None
)
fifth_site_signal_sou = (
LdsStatus(int(get_value(fifth_kp_kp.get("sou")))) if fifth_kp_kp.get("sou") else None
)
fifth_site_signal_gravity = get_value(fifth_kp_kp.get("gravity"))
sixth_kp_kp = all_signals.get("controlled_site_sixth") or {}
if sixth_kp_kp:
sixth_site_signal_pump = (
StationaryStatus(int(get_value(sixth_kp_kp.get("pump")))) if sixth_kp_kp.get("pump") else None
)
sixth_site_signal_sou = (
LdsStatus(int(get_value(sixth_kp_kp.get("sou")))) if sixth_kp_kp.get("sou") else None
)
sixth_site_signal_gravity = get_value(sixth_kp_kp.get("gravity"))
seventh_kp_kp = all_signals.get("controlled_site_seventh") or {}
if seventh_kp_kp:
seventh_site_signal_pump = (
StationaryStatus(int(get_value(seventh_kp_kp.get("pump")))) if seventh_kp_kp.get("pump") else None
)
seventh_site_signal_sou = (
LdsStatus(int(get_value(seventh_kp_kp.get("sou")))) if seventh_kp_kp.get("sou") else None
)
seventh_site_signal_gravity = get_value(seventh_kp_kp.get("gravity"))
eighth_kp_kp = all_signals.get("controlled_site_eight") or {}
if eighth_kp_kp:
eight_site_signal_pump = (
StationaryStatus(int(get_value(eighth_kp_kp.get("pump")))) if eighth_kp_kp.get("pump") else None
)
eight_site_signal_sou = (
LdsStatus(int(get_value(eighth_kp_kp.get("sou")))) if eighth_kp_kp.get("sou") else None
)
eight_site_signal_gravity = get_value(eighth_kp_kp.get("gravity"))
with SoftAssertions() as soft_failures:
StepCheck(
"Проверка сигнала - режим МТ на участке Тихорецкая-Нововеличковская",
"Режим МТ",
soft_failures,
).actual(first_site_signal_pump).expected(exp_first_site_reg_lu).equal_to()
StepCheck(
"Проверка сигнала - режим СОУ на участке Тихорецкая-Нововеличковская",
"Режим СОУ",
soft_failures,
).actual(first_site_signal_sou).expected(exp_first_site_reg_sou).equal_to()
StepCheck(
f"Проверка {GravityPipe.absent_gravity.description} \n" f"на участке Тихорецкая-Нововеличковская",
"Количество самотеков",
soft_failures,
).actual(first_site_signal_gravity).expected(str(GravityPipe.absent_gravity.id)).equal_to()
StepCheck(
"Проверка сигнала - режим МТ на участке Нововеличковская-Крымская",
"Режим МТ",
soft_failures,
).actual(second_site_signal_pump).expected(exp_second_site_reg_lu).equal_to()
StepCheck(
f"Проверка {GravityPipe.absent_gravity.description}\n" f"на участке Нововеличковская-Крымская",
"Количество самотеков",
soft_failures,
).actual(second_site_signal_gravity).expected(str(GravityPipe.absent_gravity.id)).equal_to()
StepCheck(
"Проверка сигнала - режим СОУ на участке Нововеличковская-Крымская",
"Режим СОУ",
soft_failures,
).actual(second_site_signal_sou).expected(exp_second_site_reg_sou).equal_to()
StepCheck(
"Проверка сигнала - режим МТ на участке Крымская-Грушовая",
"Режим МТ",
soft_failures,
).actual(
third_site_signal_pump
).expected(exp_third_site_reg_lu).equal_to()
StepCheck(
f"Проверка {GravityPipe.present_gravity.description} на участке Крымская-Грушовая",
"Количество самотеков",
soft_failures,
).actual(third_site_signal_gravity).expected(str(GravityPipe.present_gravity.id)).equal_to()
StepCheck(
"Проверка сигнала - режим СОУ на участке Крымская-Грушовая",
"Режим СОУ",
soft_failures,
).actual(
third_site_signal_sou
).expected(exp_third_site_reg_sou).equal_to()
StepCheck(
"Проверка сигнала - режим МТ на резервной нитке Бейсуг",
"Режим МТ",
soft_failures,
).actual(
fourth_site_signal_pump
).expected(exp_fourth_site_reg_lu).equal_to()
StepCheck(
f"Проверка {GravityPipe.absent_gravity.description} на резервной нитке Бейсуг",
"Количество самотеков",
soft_failures,
).actual(fourth_site_signal_gravity).expected(str(GravityPipe.absent_gravity.id)).equal_to()
StepCheck(
"Проверка сигнала - режим СОУ на резервной нитке Бейсуг",
"Режим СОУ",
soft_failures,
).actual(
fourth_site_signal_sou
).expected(exp_fourth_site_reg_sou).equal_to()
StepCheck(
"Проверка сигнала - режим МТ на резервной нитке Понура",
"Режим МТ",
soft_failures,
).actual(
fifth_site_signal_pump
).expected(exp_fifth_site_reg_lu).equal_to()
StepCheck(
"Проверка сигнала - режим СОУ на резервной нитке Понура",
"Режим СОУ",
soft_failures,
).actual(
fifth_site_signal_sou
).expected(exp_fifth_site_reg_sou).equal_to()
StepCheck(
f"Проверка {GravityPipe.absent_gravity.description} на резервной нитке Понура",
"Количество самотеков",
soft_failures,
).actual(fifth_site_signal_gravity).expected(str(GravityPipe.absent_gravity.id)).equal_to()
StepCheck(
"Проверка сигнала - режим МТ на резервной нитке Кубань",
"Режим МТ",
soft_failures,
).actual(
sixth_site_signal_pump
).expected(exp_sixth_site_reg_lu).equal_to()
StepCheck(
"Проверка сигнала - режим СОУ на резервной нитке Кубань",
"Режим СОУ",
soft_failures,
).actual(
sixth_site_signal_sou
).expected(exp_sixth_site_reg_sou).equal_to()
StepCheck(
f"Проверка {GravityPipe.absent_gravity.description} на резервной нитке Кубань",
"Количество самотеков",
soft_failures,
).actual(sixth_site_signal_gravity).expected(str(GravityPipe.absent_gravity.id)).equal_to()
StepCheck(
"Проверка сигнала - режим МТ на НПЗ Афипский",
"Режим МТ",
soft_failures,
).actual(
seventh_site_signal_pump
).expected(exp_seventh_site_reg_lu).equal_to()
StepCheck(
"Проверка сигнала - режим СОУ на НПЗ Афипский",
"Режим СОУ",
soft_failures,
).actual(
seventh_site_signal_sou
).expected(exp_seventh_site_reg_sou).equal_to()
StepCheck(
f"Проверка {GravityPipe.absent_gravity.description} на НПЗ Афипский",
"Количество самотеков",
soft_failures,
).actual(seventh_site_signal_gravity).expected(str(GravityPipe.absent_gravity.id)).equal_to()
StepCheck(
"Проверка сигнала - режим МТ на НПЗ Ильинский",
"Режим МТ",
soft_failures,
).actual(
eight_site_signal_pump
).expected(exp_eighth_site_reg_lu).equal_to()
StepCheck(
"Проверка сигнала - режим СОУ на НПЗ Ильинский",
"Режим СОУ",
soft_failures,
).actual(
eight_site_signal_sou
).expected(exp_eighth_site_reg_sou).equal_to()
StepCheck(
f"Проверка {GravityPipe.absent_gravity.description} на НПЗ Ильинский",
"Количество самотеков",
soft_failures,
).actual(eight_site_signal_gravity).expected(str(GravityPipe.absent_gravity.id)).equal_to()
def lds_status_init_in_journal(http_client, cfg: SmokeSuiteConfig | LDSStatusConfig, imitator_start_time):
"""
Проверка наличия записи в журнале о входе СОУ в режим Инициализация.
"""
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=LDS_STATUS"):
end_time = datetime.now()
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(messageTypes=int(MessageType.LDS_STATUS), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step("Фильтрация сообщений по времени и technologicalSection"):
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
time_filtered = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
time_filtered.sort(key=lambda msg: t_utils.ensure_moscow_timezone(msg.time), reverse=True)
lds_msg = next(
(
msg
for msg in time_filtered
if msg.technologicalSection == cfg.tu_name and msg.event == TestConst.JOURNAL_EVENT_LDS_INIT_COLD_START
),
None,
)
allure.attach(
f"Всего получено сообщений: {len(messages_info)}\n"
f"После фильтрации по времени ({filter_start_msk} - {filter_end_msk}): {len(time_filtered)}\n"
f"проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_LDS_INIT_ACCUM_DATA}': {'True' if lds_msg else 'False'}",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(
f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_LDS_INIT_COLD_START}'"
):
if lds_msg is None:
pytest.fail(
f"Сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_LDS_INIT_COLD_START}' "
f"не найдено среди {len(time_filtered)} отфильтрованных по времени сообщений"
)
with allure.step("Проверка актуальности сообщения"):
msg_time_msk = t_utils.ensure_moscow_timezone(lds_msg.time)
start_time_msk = t_utils.localize_as_moscow(imitator_start_time)
StepCheck(
f"Проверка: время сообщения позднее времени старта имитатора {msg_time_msk} > {start_time_msk}",
"time",
).actual(msg_time_msk > start_time_msk).expected(True).equal_to()
with SoftAssertions() as soft_failures:
StepCheck("Проверка event", "event", soft_failures).actual(lds_msg.event).expected(
TestConst.JOURNAL_EVENT_LDS_INIT_COLD_START
).equal_to()
StepCheck("Проверка technologicalSection", "technologicalSection", soft_failures).actual(
lds_msg.technologicalSection
).expected(cfg.tu_name).equal_to()
StepCheck("Проверка technologicalObject не пустой", "technologicalObject", soft_failures).actual(
lds_msg.technologicalObject
).is_not_none()
StepCheck("Проверка priority не пустой", "priority", soft_failures).actual(lds_msg.priority).is_not_none()
StepCheck("Проверка messageType", "messageType", soft_failures).actual(lds_msg.messageType).expected(
TestConst.JOURNAL_MESSAGE_TYPE_LDS_STATUS
).equal_to()
async def main_page_info(ws_client, cfg: SmokeSuiteConfig):
"""
Проверка установки режима МТ.
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: MainPageInfoContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"MainPageInfoContent",
"subscribeMainPageInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_payload = parser.parse_main_page_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
tu_info = getattr(parsed_payload.replyContent, 'tuInfo', None)
StepCheck("Проверка наличия данных по ТУ", "tuInfo").actual(tu_info).is_not_none()
main_pipeline_stationary_status = (
StationaryStatus(tu_info.stationaryStatus) if tu_info.stationaryStatus else None
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_payload.replyContent.tuId
).expected(cfg.tu_id).equal_to()
StepCheck(
f"Проверка установки стационара для ТУ {cfg.tu_name}",
"stationaryStatus",
soft_failures,
).actual(
main_pipeline_stationary_status
).expected(cfg.expected_stationary_status).equal_to()
async def main_page_info_signals(ws_client, cfg: SmokeSuiteConfig):
"""
Проверка счетчиков состояния сигналов
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: MainPageSignalsInfoContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"MainPageSignalsInfoContent",
"subscribeMainPageSignalsInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_payload = parser.parse_main_page_signals_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
signals_info = getattr(parsed_payload.replyContent, 'signalsInfo', None)
StepCheck("Проверка данных сигналов ТУ", "signalsInfo").actual(signals_info).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_payload.replyContent.tuId
).expected(cfg.tu_id).equal_to()
field_name = "numberOfRejectedSignals"
# Проверяет что количество отбракованных сигналов больше или равно ОР
StepCheck(
f"Проверка количества отбракованных сигналов ТУ {cfg.tu_name}",
field_name,
soft_failures,
).actual(
signals_info.numberOfRejectedSignals
).is_greater_than_or_equal_to(cfg.expected_main_page_signals[field_name])
field_name = "numberOfMaskedSignals"
StepCheck(
f"Проверка количества маскированных сигналов ТУ {cfg.tu_name}",
field_name,
soft_failures,
).actual(
signals_info.numberOfMaskedSignals
).expected(cfg.expected_main_page_signals[field_name]).equal_to()
field_name = "numberOfImitatedSignals"
StepCheck(
f"Проверка количества имитированных сигналов ТУ {cfg.tu_name}",
field_name,
soft_failures,
).actual(
signals_info.numberOfImitatedSignals
).expected(cfg.expected_main_page_signals[field_name]).equal_to()
async def main_page_info_unstationary(ws_client, cfg: SmokeSuiteConfig):
"""
Проверка установки режима Нестационар (для наборов с несколькими утечками).
Запускается после первой утечки, когда режим переходит в Нестационар.
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: MainPageInfoContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"MainPageInfoContent",
"subscribeMainPageInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_payload = parser.parse_main_page_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
tu_info = getattr(parsed_payload.replyContent, 'tuInfo', None)
StepCheck("Проверка наличия данных по ТУ", "tuInfo").actual(tu_info).is_not_none()
main_pipeline_stationary_status = (
StationaryStatus(tu_info.stationaryStatus) if tu_info.stationaryStatus else None
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_payload.replyContent.tuId
).expected(cfg.tu_id).equal_to()
StepCheck(
f"Проверка установки режима Нестационар для ТУ {cfg.tu_name}",
"stationaryStatus",
soft_failures,
).actual(main_pipeline_stationary_status).expected(StationaryStatus.UNSTATIONARY).equal_to()
async def leak_is_confirm_on_main_page(ws_client, cfg: SmokeSuiteConfig):
"""
MainPageInfoContent - проверка подтвержденной утечки на ЭФ Состояние МТ
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: MainPageInfoContent."):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"MainPageInfoContent",
"subscribeMainPageInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_payload = parser.parse_main_page_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
# Ставим object() как заглушку, что бы исключить получение None
leaks_info = getattr(getattr(parsed_payload.replyContent, 'tuInfo', object()), 'leaksInfo', [])
StepCheck("Проверка наличия списка сообщений об утечках", "leakStatus").actual(leaks_info).is_not_empty()
confirm_leak = t_utils.find_object_by_field(leaks_info, "leakStatus", LeakStatus.CONFIRMED.value)
lds_status_int = getattr(confirm_leak, 'leakStatus', None)
lds_status = LeakStatus(lds_status_int) if lds_status_int else None
StepCheck("Проверка подтвержденной утечки на ЭФ Состояние МТ", "leakStatus").actual(lds_status).expected(
LeakStatus.CONFIRMED
).equal_to()
async def leak_is_complete_on_main_page(ws_client, cfg: SmokeSuiteConfig):
"""
MainPageInfoContent - отсутствует подтвержденная утечка на ЭФ Состояние МТ
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: MainPageInfoContent."):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"MainPageInfoContent",
"subscribeMainPageInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_payload = parser.parse_main_page_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
main_page_leak_info = parsed_payload.replyContent.tuInfo.leaksInfo
confirmed_and_closed_leaks = t_utils.find_confirmed_leaks_on_main_page(main_page_leak_info)
StepCheck("Проверка подтвержденной утечки на ЭФ Состояние МТ", "leakStatus").actual(
confirmed_and_closed_leaks
).is_empty()
async def imitate_sensor_signal(
ws_client, http_client, cfg: SmokeSuiteConfig, test_data: CaseData, imitator_start_time
):
"""
Проверка имитации сигнала датчика.
"""
# Распаковка данных для теста
sensor_address = test_data.params.get("sensor_address")
sensor_val, sensor_quality, imitation_event, unimitation_event = test_data.expected_result
# Получение актуального id датчика
sensor_id = TestConst.SENSOR_IDS_BY_ADDRESS.get(sensor_address)
with allure.step(f"Http запрос и обработка ответа об имитации сигнала датчика с id: {sensor_id}"):
request_body = {
'id': sensor_id,
'tuId': cfg.tu_id,
'imitateInfo': {'value': str(sensor_val), 'quality': sensor_quality},
}
response = http_client.post_request(HttpConst.IMITATE_SIGNAL_URL_PATH, request_body)
sensor_imitate_reply_status = response.status_code
StepCheck("Проверка кода ответа на запрос об имитации", "replyStatus").actual(
sensor_imitate_reply_status
).expected(ReplyStatus.OK.value).equal_to()
time.sleep(cfg.basic_message_timeout)
with allure.step(
"Получение данных для проверки имитации. Тип сообщения: InputSignalsContent. ЭФ Диагностика Сигналов."
):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"InputSignalsContent",
"SubscribeInputSignalsRequest",
{
'signalIds': [sensor_id],
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_input_signals_info_msg(payload)
with allure.step(
"Подготовка данных для проверки имитации. Тип сообщения: InputSignalsContent. ЭФ Диагностика Сигналов."
):
sensor_data = getattr(parsed_payload.replyContent, 'inputSignals', [])
diagnostics_sensor_imitate_data = t_utils.find_object_by_field(sensor_data, "id", sensor_id)
value_diagnostics_sensor_imitate = getattr(
getattr(diagnostics_sensor_imitate_data, 'imitation', []), 'value', []
)
is_imitated_diagnostics_sensor_imitate = getattr(diagnostics_sensor_imitate_data, 'isImitated', [])
quality_diagnostics_sensor_imitate = getattr(diagnostics_sensor_imitate_data, 'quality', [])
with allure.step("Подготовка данных для проверки имитации. Тип сообщения: SchemeSignalsStateContent. ЭФ Схема."):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"SchemeSignalsStateContent",
"SubscribeSchemeSignalsStateRequest",
{'tuId': cfg.tu_id},
)
parsed_payload = parser.parse_scheme_signals_state_msg(payload)
journal_signal_state_list = parsed_payload.replyContent.signalsStates
scheme_sensor_imitate_data = t_utils.find_object_by_field(journal_signal_state_list, "id", sensor_id)
is_imitated_scheme_sensor_imitate = getattr(scheme_sensor_imitate_data, 'isImitated', [])
with allure.step(f"Http запрос и обработка ответа о снятии имитации датчика с id: {sensor_id}"):
request_body = {'id': sensor_id, 'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.UNIMITATE_SIGNAL_URL_PATH, request_body)
sensor_unimitate_reply_status = response.status_code
StepCheck("Проверка кода ответа на запрос о снятии имитации", "replyStatus").actual(
sensor_unimitate_reply_status
).expected(ReplyStatus.OK.value).equal_to()
time.sleep(cfg.basic_message_timeout)
with allure.step("Получение данных для проверки имитации. InputSignalsContent. ЭФ Диагностика Сигналов."):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"InputSignalsContent",
"SubscribeInputSignalsRequest",
{
'signalIds': [sensor_id],
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_input_signals_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки снятия имитации"):
sensor_data = getattr(parsed_payload.replyContent, 'inputSignals', [])
diagnostics_sensor_unimitate_data = t_utils.find_object_by_field(sensor_data, 'id', sensor_id)
is_unimitated_diagnostics_sensor_unimitate = getattr(diagnostics_sensor_unimitate_data, 'isImitated', [])
with allure.step(
"Получение данных для проверки снятия имитации. Тип сообщения: SchemeSignalsStateContent. ЭФ Схема"
):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"SchemeSignalsStateContent",
"SubscribeSchemeSignalsStateRequest",
{'tuId': cfg.tu_id},
)
parsed_payload = parser.parse_scheme_signals_state_msg(payload)
journal_signal_state_list = parsed_payload.replyContent.signalsStates
scheme_sensor_unimitate = t_utils.find_object_by_field(journal_signal_state_list, "id", sensor_id)
is_unimitated_scheme_sensor_unimitate = getattr(scheme_sensor_unimitate, 'isImitated', [])
with allure.step("Http запрос сообщений журнала с фильтром userActions"):
# Запрос сообщений по типу действий пользователя маскирование и имитация сигналов
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(userActions=int(UserActions.SIGNAL_MASK_SIM), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
# Фильтр полученных сообщений по времени
end_time = datetime.now()
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
filter_time_messages = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
filter_by_imitation_messages = [
msg
for msg in filter_time_messages
if getattr(msg, "event", None) == imitation_event and getattr(msg, "tag", None) == sensor_address
]
filter_by_unimitation_messages = [
msg
for msg in filter_time_messages
if getattr(msg, "event", None) == imitation_event and getattr(msg, "tag", None) == sensor_address
]
with SoftAssertions() as soft_failures:
StepCheck(
f"Проверка имитации датчика с id: {sensor_id}. ЭФ Диагностика Сигналов.", "isImitated", soft_failures
).actual(is_imitated_diagnostics_sensor_imitate).expected(True).equal_to()
StepCheck(f"Проверка имитации датчика с id: {sensor_id}. ЭФ Схема.", "isImitated", soft_failures).actual(
is_imitated_scheme_sensor_imitate
).expected(True).equal_to()
StepCheck(
f"Проверка показаний датчика с id: {sensor_id}. ЭФ Диагностика Сигналов.", "value", soft_failures
).actual(value_diagnostics_sensor_imitate).expected(sensor_val).equal_to()
StepCheck(
f"Проверка качества сигнала датчика с id: {sensor_id}. ЭФ Диагностика Сигналов.", "quality", soft_failures
).actual(quality_diagnostics_sensor_imitate).expected(sensor_quality).equal_to()
StepCheck(
f"Проверка наличия сообщения с событием {imitation_event} и тегом {sensor_address}. ЭФ Журнал.",
"event",
soft_failures,
).actual(filter_by_imitation_messages).is_not_empty()
StepCheck(
f"Проверка снятия имитации датчика с id: {sensor_id}. ЭФ Диагностика Сигналов.", "isImitated", soft_failures
).actual(is_unimitated_diagnostics_sensor_unimitate).expected(False).equal_to()
StepCheck(f"Проверка снятия имитации датчика с id: {sensor_id}. ЭФ Схема.", "isImitated", soft_failures).actual(
is_unimitated_scheme_sensor_unimitate
).expected(False).equal_to()
StepCheck(
f" Проверка наличия сообщения с событием {unimitation_event} и тегом {sensor_address}. ЭФ Журнал.",
"event",
soft_failures,
).actual(filter_by_unimitation_messages).is_not_empty()
async def mask_signal_test(ws_client, http_client, cfg: SmokeSuiteConfig, test_data: CaseData):
"""
Проверка маскирования датчиков.
"""
if not test_data:
pytest.fail("Данные датчиков отсутствуют")
pressure_sensor_address = test_data.params.get("pressure_sensor_address")
flowmeter_address = test_data.params.get("flowmeter_address")
# Получение актуальных id датчиков
pressure_sensor_id = TestConst.SENSOR_IDS_BY_ADDRESS.get(pressure_sensor_address)
flowmeter_id = TestConst.SENSOR_IDS_BY_ADDRESS.get(flowmeter_address)
with allure.step("Маскирование датчиков"):
with allure.step(f"Http запрос и обработка ответа о маскировании датчика давления с id: {pressure_sensor_id}"):
request_body = {'id': pressure_sensor_id, 'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.MASK_SIGNAL_URL_PATH, request_body)
pressure_sensor_mask_reply_status = response.status_code
StepCheck("Проверка кода ответа на запрос о маскировании", "replyStatus").actual(
pressure_sensor_mask_reply_status
).expected(ReplyStatus.OK.value).equal_to()
with allure.step(f"Http запрос и обработка ответа о маскировании расходомера с id: {flowmeter_id}"):
request_body = {'id': flowmeter_id, 'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.MASK_SIGNAL_URL_PATH, request_body)
flowmeter_mask_reply_status = response.status_code
StepCheck("Проверка кода ответа на запрос о маскировании", "replyStatus").actual(
flowmeter_mask_reply_status
).expected(ReplyStatus.OK.value).equal_to()
with allure.step(
"Подключение по ws, получение и обработка данных о статусе датчиков из сообщения типа: InputSignalsContent"
):
time.sleep(cfg.basic_message_timeout)
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"InputSignalsContent",
"SubscribeInputSignalsRequest",
{
'signalIds': [pressure_sensor_id, flowmeter_id],
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_input_signals_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки маскирования"):
sensor_data = getattr(parsed_payload.replyContent, 'inputSignals', [])
pressure_sensor_mask_data = t_utils.find_object_by_field(sensor_data, "id", pressure_sensor_id)
flowmeter_mask_data = t_utils.find_object_by_field(sensor_data, "id", flowmeter_id)
with SoftAssertions() as soft_failures:
StepCheck("Проверка наличия данных о маскировании датчика давления", "inputSignals", soft_failures).actual(
pressure_sensor_mask_data
).is_not_none()
StepCheck("Проверка наличия данных о маскировании расходомера", "inputSignals", soft_failures).actual(
flowmeter_mask_data
).is_not_none()
with allure.step("Снятие маскирования датчиков"):
with allure.step(f"Http запрос о снятии маскирования датчика давления с id: {pressure_sensor_id}"):
request_body = {'id': pressure_sensor_id, 'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.UNMASK_SIGNAL_URL_PATH, request_body)
pressure_sensor_unmask_reply_status = response.status_code
StepCheck("Проверка кода ответа на запрос о снятии маскирования", "replyStatus").actual(
pressure_sensor_unmask_reply_status
).expected(ReplyStatus.OK.value).equal_to()
with allure.step(f"Http запрос о снятии маскирования расходомера с id: {flowmeter_id}"):
request_body = {'id': flowmeter_id, 'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.UNMASK_SIGNAL_URL_PATH, request_body)
flowmeter_unmask_reply_status = response.status_code
StepCheck("Проверка кода ответа на запрос о снятии маскирования", "replyStatus").actual(
flowmeter_unmask_reply_status
).expected(ReplyStatus.OK.value).equal_to()
with allure.step(
"Подключение по ws, получение и обработка данных о статусе датчиков из сообщения типа: InputSignalsContent"
):
time.sleep(cfg.basic_message_timeout)
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"InputSignalsContent",
"SubscribeInputSignalsRequest",
{
'signalIds': [pressure_sensor_id, flowmeter_id],
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_input_signals_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки снятия маскирования"):
sensor_data = getattr(parsed_payload.replyContent, 'inputSignals', [])
pressure_sensor_unmask_data = t_utils.find_object_by_field(sensor_data, "id", pressure_sensor_id)
flowmeter_unmask_data = t_utils.find_object_by_field(sensor_data, "id", flowmeter_id)
with SoftAssertions() as soft_failures:
StepCheck(
"Проверка наличия данных о снятии маскировании датчика давления", "inputSignals", soft_failures
).actual(pressure_sensor_unmask_data).is_not_none()
StepCheck(
"Проверка наличия данных о снятии маскировании расходомера", "inputSignals", soft_failures
).actual(flowmeter_unmask_data).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
f"Проверка маскирования датчика давления с id: {pressure_sensor_id}", "isMasked", soft_failures
).actual(pressure_sensor_mask_data.isMasked).expected(True).equal_to()
StepCheck(f"Проверка маскирования расходомера с id: {flowmeter_id}", "isMasked", soft_failures).actual(
flowmeter_mask_data.isMasked
).expected(True).equal_to()
StepCheck(
f"Проверка снятия маскирования датчика давления с id: {pressure_sensor_id}", "isMasked", soft_failures
).actual(pressure_sensor_unmask_data.isMasked).expected(False).equal_to()
StepCheck(f"Проверка снятия маскирования расходомера с id: {flowmeter_id}", "isMasked", soft_failures).actual(
flowmeter_unmask_data.isMasked
).expected(False).equal_to()
async def mask_info_in_journal(http_client, cfg: SmokeSuiteConfig, imitator_start_time):
"""
Проверка записей журнала о маскировании и размаскировании.
"""
with allure.step("Http запрос сообщений журнала с фильтром userActions"):
end_time = datetime.now()
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_MASK_PAGINATION_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(userActions=int(UserActions.SIGNAL_MASK_SIM), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step("Фильтрация сообщений по событиям маскирования и временному диапазону"):
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
mask_unmask_msgs = [
msg
for msg in messages_info
if msg.event in TestConst.JOURNAL_MASK_EXPECTED_EVENTS
and msg.signalName in TestConst.JOURNAL_MASK_EXPECTED_SIGNALS
]
journal_messages = [
msg
for msg in mask_unmask_msgs
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
allure.attach(
f"Всего получено сообщений: {len(messages_info)}\n"
f"После фильтрации по event и signalName осталось сообщений: {len(mask_unmask_msgs)}\n"
f"После фильтрации по времени ({filter_start_msk} - {filter_end_msk}) "
f"осталось сообщений: {len(journal_messages)}",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Группировка отфильтрованных сообщений"):
pressure_msgs = [msg for msg in journal_messages if msg.signalName == TestConst.JOURNAL_SIGNAL_PRESSURE]
flow_msgs = [msg for msg in journal_messages if msg.signalName == TestConst.JOURNAL_SIGNAL_FLOW]
mask_event_msgs = [msg for msg in journal_messages if msg.event == TestConst.JOURNAL_EVENT_MASK]
unmask_event_msgs = [msg for msg in journal_messages if msg.event == TestConst.JOURNAL_EVENT_UNMASK]
mask_signal_names = {msg.signalName for msg in mask_event_msgs}
unmask_signal_names = {msg.signalName for msg in unmask_event_msgs}
with SoftAssertions() as journal_soft_failures:
StepCheck(
"Проверка соответствия количества сообщений о действиях пользователя (снятие и установка "
"маскирования для датчиков давления и расходомеров)",
"total_count",
journal_soft_failures,
).actual(len(journal_messages)).expected(TestConst.JOURNAL_EXPECTED_MASK_MSG_TOTAL).equal_to()
StepCheck(
f"Проверка соответствия количества сообщений "
f"о действиях пользователя для датчиков давления - '{TestConst.JOURNAL_SIGNAL_PRESSURE}'",
"count",
journal_soft_failures,
).actual(len(pressure_msgs)).expected(TestConst.JOURNAL_EXPECTED_MSG_COUNT_PER_SIGNAL).equal_to()
StepCheck(
f"Проверка соответствия количества сообщений "
f"о действиях пользователя для расходомеров - '{TestConst.JOURNAL_SIGNAL_FLOW}'",
"count",
journal_soft_failures,
).actual(len(flow_msgs)).expected(TestConst.JOURNAL_EXPECTED_MSG_COUNT_PER_SIGNAL).equal_to()
StepCheck(
f"Проверка: событие '{TestConst.JOURNAL_EVENT_MASK}' содержит '{TestConst.JOURNAL_SIGNAL_PRESSURE}'",
"signalName",
journal_soft_failures,
).actual(TestConst.JOURNAL_SIGNAL_PRESSURE in mask_signal_names).expected(True).equal_to()
StepCheck(
f"Проверка: событие '{TestConst.JOURNAL_EVENT_MASK}' содержит '{TestConst.JOURNAL_SIGNAL_FLOW}'",
"signalName",
journal_soft_failures,
).actual(TestConst.JOURNAL_SIGNAL_FLOW in mask_signal_names).expected(True).equal_to()
StepCheck(
f"Проверка: событие '{TestConst.JOURNAL_EVENT_UNMASK}' содержит '{TestConst.JOURNAL_SIGNAL_PRESSURE}'",
"signalName",
journal_soft_failures,
).actual(TestConst.JOURNAL_SIGNAL_PRESSURE in unmask_signal_names).expected(True).equal_to()
StepCheck(
f"Проверка: событие '{TestConst.JOURNAL_EVENT_UNMASK}' содержит '{TestConst.JOURNAL_SIGNAL_FLOW}'",
"signalName",
journal_soft_failures,
).actual(TestConst.JOURNAL_SIGNAL_FLOW in unmask_signal_names).expected(True).equal_to()
for signal_name in [TestConst.JOURNAL_SIGNAL_PRESSURE, TestConst.JOURNAL_SIGNAL_FLOW]:
mask_msg_for_signal = next((msg for msg in mask_event_msgs if msg.signalName == signal_name), None)
unmask_msg_for_signal = next((msg for msg in unmask_event_msgs if msg.signalName == signal_name), None)
if mask_msg_for_signal and unmask_msg_for_signal:
StepCheck(
f"Проверка совпадения tag для '{signal_name}' между маскированием и снятием",
"tag",
journal_soft_failures,
).actual(mask_msg_for_signal.tag).expected(unmask_msg_for_signal.tag).equal_to()
for msg in journal_messages:
msg_label = f"{msg.event} - {msg.signalName}"
StepCheck(
f"Проверка user не пустой [{msg_label}]",
"user",
journal_soft_failures,
).actual(msg.user).is_not_none()
StepCheck(
f"Проверка object не пустой [{msg_label}]",
"object",
journal_soft_failures,
).actual(msg.object).is_not_none()
StepCheck(
f"Проверка technologicalObject не пустой [{msg_label}]",
"technologicalObject",
journal_soft_failures,
).actual(msg.technologicalObject).is_not_none()
StepCheck(
f"Проверка technologicalSection [{msg_label}]",
"technologicalSection",
journal_soft_failures,
).actual(msg.technologicalSection).expected(cfg.tu_name).equal_to()
StepCheck(
f"Проверка priority не пустой [{msg_label}]",
"priority",
journal_soft_failures,
).actual(msg.priority).is_not_none()
StepCheck(
f"Проверка messageType [{msg_label}]",
"messageType",
journal_soft_failures,
).actual(
msg.messageType
).expected(TestConst.JOURNAL_MESSAGE_TYPE_USER_ACTIONS).equal_to()
StepCheck(
f"Проверка status [{msg_label}]",
"status",
journal_soft_failures,
).actual(
msg.status
).expected(TestConst.JOURNAL_STATUS_SUCCESS).equal_to()
async def mask_du_on_mini_scheme(ws_client, http_client, cfg: SmokeSuiteConfig):
"""
Маскирование ДУ на мини-схеме
Проверка маскированного участка в выходных сигналах
"""
linear_part_id = cfg.linear_part_identifier_for_mask
mask_reason = cfg.mask_reason
with allure.step("Http запрос на маскирование ДУ утечки типа: MaskLdsRequest"):
request_body = {
'tuId': cfg.tu_id,
'maskInfo': [
{
'linearPartId': linear_part_id,
'reason': mask_reason,
'additionalProperties': None,
}
],
'additionalProperties': None,
}
response = http_client.post_request(HttpConst.MASK_LDS_URL_PATH, request_body)
mask_du_reply_status = response.status_code
time.sleep(cfg.basic_message_timeout)
with allure.step("Подключение по ws и http, получение и обработка данных сообщений для теста"):
with allure.step("Http запрос, получение и обработка ответа типа: GetOutputSignalsRequest"):
request_body = {'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.GET_OUTPUT_SIGNALS_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_output_signals_msg(payload)
# Получение данных линейного участка утечки по id
with allure.step(
"Извлечение и подготовка данных типов выходных сигналов из обработанных данных ЭФ Выходные сигналы"
):
leak_linear_part = t_utils.find_object_by_field(
parsed_payload.replyContent.linearPartSignals,
TestConst.LEAK_LINEAR_PART_ID_KEY,
linear_part_id,
)
leak_signals_list = leak_linear_part.signals
mask_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_MASK
)
StepCheck(
"Проверка получения типов выходных сигналов в ЭФ Выходные сигналы", "linearPartSignals"
).actual(mask_signal_type).is_not_none()
with allure.step(
"Получение сообщения типа: OutputSignalsInfo. "
f"С данными выходных сигналов для линейного участка с id: {linear_part_id}\n"
):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"OutputSignalsInfo",
"SubscribeOutputSignalsRequest",
{
'objects': {
'linearParts': [{'linearPartId': linear_part_id}],
'controlledSites': [],
},
'signalTypes': 1023,
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_output_signals_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки маскирования ДУ ЭФ Выходные сигналы"):
leak_linear_part = t_utils.find_object_by_field(
parsed_payload.replyContent.linearPartSignals,
TestConst.LEAK_LINEAR_PART_ID_KEY,
linear_part_id,
)
StepCheck("Проверка наличия данных о маскировании в ЭФ Выходные сигналы", "linearPartSignals").actual(
leak_linear_part
).is_not_none()
leak_signals_list = leak_linear_part.signals
mask_leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, mask_signal_type)
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent. ЭФ Схема"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"CommonSchemeContent",
"SubscribeCommonSchemeRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
parsed_payload = parser.parse_common_scheme_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки маскирования ДУ ЭФ Схема"):
linear_parts = parsed_payload.replyContent.linearParts
mask_linear_part = next((lp for lp in linear_parts if lp.id == linear_part_id), None)
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=MASKING_LDS"):
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=10, direction=Direction.FIRST.value),
filtering=Filtering(
messageTypes=int(MessageType.MASKING_LDS), objects=FilteringObjects(tuId=cfg.tu_id)
),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки маскирования ДУ ЭФ Журнал"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', None)
if cfg.technological_section:
journal_mask_message = t_utils.find_object_by_field(
messages_info, "technologicalSection", cfg.technological_section
)
else:
journal_mask_message = parsed_payload.replyContent.messagesInfo[0]
StepCheck("Проверка наличия данных о маскировании в журнале", "technologicalSection").actual(
journal_mask_message
).is_not_none()
with allure.step(
"Подключение по ws, получение и обработка сообщения типа: MainPageInfoContent. "
"Получен результат маскирования ДУ на ЭФ Состояние МТ"
):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"MainPageInfoContent",
"subscribeMainPageInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_payload = parser.parse_main_page_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки маскирования ДУ ЭФ Состояние МТ"):
lds_status = getattr(getattr(parsed_payload.replyContent, "tuInfo", None), 'ldsStatus', None)
number_of_masked_lps = getattr(lds_status, 'numberOfMaskedLps', None)
mask_du_list = getattr(lds_status, "maskedLps", None)
if mask_du_list:
masked_lps_name = next(iter(mask_du_list), None)
else:
masked_lps_name = None
# Проверки сообщений
with SoftAssertions() as soft_failures:
StepCheck("Проверка кода ответа на запрос о маскировании ДУ", "replyStatus", soft_failures).actual(
mask_du_reply_status
).expected(ReplyStatus.OK.value).equal_to()
StepCheck(
"Проверка сигнала маскирования ДУ в выходных сигналах", TestConst.ADDRESS_SUFFIX_MASK, soft_failures
).actual(mask_leak_value).expected(TestConst.OUTPUT_IS_MASK).equal_to()
StepCheck("Проверка статуса маскирования ДУ на схеме", "isMasked", soft_failures).actual(
mask_linear_part.isMasked
).expected(True).equal_to()
StepCheck("Проверка причины маски на схеме", "maskReason").actual(mask_linear_part.maskReason).expected(
cfg.mask_reason
).equal_to()
StepCheck("Проверка имени ТУ в журнале", "mainPipeline", soft_failures).actual(
journal_mask_message.mainPipeline
).expected(cfg.main_pipe_line).equal_to()
StepCheck("Проверка имени ДУ в журнале", "technologicalObject", soft_failures).actual(
journal_mask_message.technologicalObject
).expected(cfg.mask_du_name).equal_to()
StepCheck("Проверка события в журнале", "event", soft_failures).actual(journal_mask_message.event).expected(
cfg.mask_du_event
).equal_to()
StepCheck("Проверка количества маскированных ДУ. ЭФ Состояние МТ", "numberOfMaskedLps", soft_failures).actual(
number_of_masked_lps
).expected(cfg.mask_one_du).equal_to()
StepCheck("Проверка имени маскированного ДУ. ЭФ Состояние МТ", "maskedLps", soft_failures).actual(
masked_lps_name
).expected(cfg.mask_du_name).equal_to()
async def unmask_du_on_mini_scheme(ws_client, http_client, cfg: SmokeSuiteConfig):
"""
Размаскирование ДУ на мини-схеме
Проверка маскированного участка в выходных сигналах
"""
linear_part_id = cfg.linear_part_identifier_for_mask
unmask_reason = cfg.unmask_reason
with allure.step("Http запрос на снятие маскирования ДУ типа: UnmaskLdsRequest"):
request_body = {
'tuId': cfg.tu_id,
'maskInfo': [
{
'linearPartId': linear_part_id,
'reason': unmask_reason,
'additionalProperties': None,
}
],
'additionalProperties': None,
}
response = http_client.post_request(HttpConst.UNMASK_LDS_URL_PATH, request_body)
unmask_du_reply_status = response.status_code
time.sleep(cfg.basic_message_timeout)
with allure.step("Подключение по ws, получение и обработка данных сообщений для теста"):
with allure.step("Http запрос, получение и обработка ответа типа: GetOutputSignalsRequest"):
request_body = {'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.GET_OUTPUT_SIGNALS_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_output_signals_msg(payload)
with allure.step("Извлечение и подготовка данных типов выходных сигналов ЭФ Выходные сигналы"):
# Получение данных линейного участка утечки по id
leak_linear_part = t_utils.find_object_by_field(
parsed_payload.replyContent.linearPartSignals,
TestConst.LEAK_LINEAR_PART_ID_KEY,
linear_part_id,
)
leak_signals_list = leak_linear_part.signals
mask_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_MASK
)
StepCheck("Проверка наличия данных типов выходных сигналов ЭФ Выходные сигналы", "signalType").actual(
mask_signal_type
).is_not_none()
with allure.step(
"Получение сообщения типа: OutputSignalsInfo. "
f"С данными выходных сигналов для линейного участка с id: {linear_part_id}\n"
):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"OutputSignalsInfo",
"SubscribeOutputSignalsRequest",
{
'objects': {
'linearParts': [{'linearPartId': linear_part_id}],
'controlledSites': [],
},
'signalTypes': 1023,
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_output_signals_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки снятия маскирования ДУ ЭФ Выходные сигналы"):
leak_linear_part = t_utils.find_object_by_field(
parsed_payload.replyContent.linearPartSignals,
TestConst.LEAK_LINEAR_PART_ID_KEY,
linear_part_id,
)
StepCheck(
"Проверка наличия данных о снятии маскирования в ЭФ Выходные сигналы", "linearPartSignals"
).actual(leak_linear_part).is_not_none()
leak_signals_list = leak_linear_part.signals
mask_leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, mask_signal_type)
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=MASKING_LDS"):
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=10, direction=Direction.FIRST.value),
filtering=Filtering(
messageTypes=int(MessageType.MASKING_LDS), objects=FilteringObjects(tuId=cfg.tu_id)
),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки снятия маскирования ДУ ЭФ Журнал"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', None)
if cfg.technological_section:
journal_unmask_message = t_utils.find_object_by_field(
messages_info, "technologicalSection", cfg.technological_section
)
else:
journal_unmask_message = parsed_payload.replyContent.messagesInfo[0]
StepCheck("Проверка наличия данных о снятии маскирования в журнале", "technologicalSection").actual(
journal_unmask_message
).is_not_none()
with allure.step(
"Подключение по ws, получение и обработка сообщения типа: MainPageInfoContent. "
"Получен результат маскирования ДУ на ЭФ Состояние МТ"
):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"MainPageInfoContent",
"subscribeMainPageInfoRequest",
{'tuIds': [cfg.tu_id], 'additionalProperties': None},
)
parsed_payload = parser.parse_main_page_msg(payload)
lds_status = getattr(getattr(parsed_payload.replyContent, "tuInfo", None), 'ldsStatus', None)
number_of_masked_lps = getattr(lds_status, 'numberOfMaskedLps', None)
masked_lps_name = getattr(lds_status, "maskedLps", None)
# Проверки сообщений
with SoftAssertions() as soft_failures:
StepCheck("Проверка кода ответа на запрос о размаскировании", "replyStatus", soft_failures).actual(
unmask_du_reply_status
).expected(ReplyStatus.OK.value).equal_to()
StepCheck(
"Проверяем, что тег маскирования ДУ в выходных сигналах равен null",
TestConst.ADDRESS_SUFFIX_MASK,
soft_failures,
).actual(mask_leak_value).expected(TestConst.OUTPUT_IS_NOT_MASK).equal_to()
StepCheck("Проверяем имя ТУ в сообщении в журнале", "mainPipeline", soft_failures).actual(
journal_unmask_message.mainPipeline
).expected(cfg.main_pipe_line).equal_to()
StepCheck("Проверяем имя ДУ в сообщении в журнале", "technologicalObject", soft_failures).actual(
journal_unmask_message.technologicalObject
).expected(cfg.mask_du_name).equal_to()
StepCheck("Проверка события в сообщении в журнале", "event", soft_failures).actual(
journal_unmask_message.event
).expected(cfg.unmask_du_event).equal_to()
StepCheck("Проверка количества маскированных ДУ", "numberOfMaskedLps", soft_failures).actual(
number_of_masked_lps
).expected(cfg.not_mask_du).equal_to()
StepCheck("Проверка счетчика маски. ЭФ Состояние МТ", "Количество замаскированных ДУ", soft_failures).actual(
number_of_masked_lps
).expected(cfg.not_mask_du).equal_to()
StepCheck(
"Проверка отсутствия списка маскированных ДУ. ЭФ Состояние МТ",
"Отсутствуют замаскированные ДУ",
soft_failures,
).actual(masked_lps_name).is_none()
async def lds_status_initialization_out(ws_client, cfg: SmokeSuiteConfig):
"""
Проверка выхода СОУ из Инициализации.
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"CommonSchemeContent",
"SubscribeCommonSchemeRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
parsed_payload = parser.parse_common_scheme_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
flow_areas = getattr(parsed_payload.replyContent, 'flowAreas', None)
longest_flow_area = t_utils.get_longest_flow_area(flow_areas)
diagnostic_areas = getattr(longest_flow_area, 'diagnosticAreas', [])
StepCheck("Проверка наличия данных диагностических участков", "diagnosticAreas").actual(
diagnostic_areas
).is_not_empty()
allure.attach(
f"Самый протяженный участок карты течений: {longest_flow_area}",
name="flowArea. Выход из Инициализации",
attachment_type=allure.attachment_type.TEXT,
)
lds_status_set = {diagnostic_area.ldsStatus for diagnostic_area in diagnostic_areas}
lds_status_int = t_utils.determine_lds_status_by_priority(lds_status_set)
lds_status = LdsStatus(lds_status_int) if lds_status_int else None
StepCheck(
"Проверка: СОУ находится не в режиме 'Инициализация'",
"ldsStatus",
).actual(
lds_status
).expected(LdsStatus.INITIALIZATION).is_not_equal_to()
def lds_status_init_out_in_journal(http_client, cfg: SmokeSuiteConfig, imitator_start_time):
"""
Проверка наличия записи в журнале о выходе СОУ из режима Инициализация.
"""
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=LDS_STATUS"):
end_time = datetime.now()
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(messageTypes=int(MessageType.LDS_STATUS), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step("Фильтрация сообщений по времени и technologicalSection"):
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
time_filtered = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
time_filtered.sort(key=lambda msg: t_utils.ensure_moscow_timezone(msg.time), reverse=True)
lds_msg = next(
(msg for msg in time_filtered if msg.technologicalSection == cfg.tu_name),
None,
)
allure.attach(
f"Всего получено сообщений: {len(messages_info)}\n"
f"После фильтрации по времени ({filter_start_msk} - {filter_end_msk}): {len(time_filtered)}\n"
f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}': {'True' if lds_msg else 'False'}",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}'"):
if lds_msg is None:
pytest.fail(
f"Сообщение с technologicalSection='{cfg.tu_name}' "
f"не найдено среди {len(time_filtered)} отфильтрованных по времени сообщений"
)
with allure.step("Проверка актуальности сообщения"):
msg_time_msk = t_utils.ensure_moscow_timezone(lds_msg.time)
start_time_msk = t_utils.localize_as_moscow(imitator_start_time)
StepCheck(
f"Проверка: время сообщения позднее времени старта имитатора {msg_time_msk} > {start_time_msk}",
"time",
).actual(msg_time_msk > start_time_msk).expected(True).equal_to()
with SoftAssertions() as soft_failures:
StepCheck("Проверка: event не является Инициализацией", "event", soft_failures).actual(lds_msg.event).expected(
TestConst.JOURNAL_EVENT_LDS_INIT_ACCUM_DATA
).is_not_equal_to()
StepCheck("Проверка technologicalSection", "technologicalSection", soft_failures).actual(
lds_msg.technologicalSection
).expected(cfg.tu_name).equal_to()
StepCheck("Проверка technologicalObject не пустой", "technologicalObject", soft_failures).actual(
lds_msg.technologicalObject
).is_not_none()
StepCheck("Проверка priority не пустой", "priority", soft_failures).actual(lds_msg.priority).is_not_none()
StepCheck("Проверка messageType", "messageType", soft_failures).actual(lds_msg.messageType).expected(
TestConst.JOURNAL_MESSAGE_TYPE_LDS_STATUS
).equal_to()
async def leaks_content(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time):
"""
Проверка утечки через сообщение LeaksContent.
"""
with allure.step("Подключение по ws и получение сообщения об утечке типа: LeaksContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"LeaksContent",
"SubscribeLeaksRequest",
{'tuId': cfg.tu_id},
)
parsed_payload = parser.parse_leaks_content_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
leaks_list_info = getattr(parsed_payload.replyContent, 'leaksListInfo', [])
# Ищет подтвержденные утечки
confirmed_leaks_list = t_utils.find_confirmed_leaks(leaks_list_info)
first_leak_info = t_utils.find_leak_by_coordinate(confirmed_leaks_list, leak.coordinate_meters)
StepCheck("Проверка наличия собщения", "Leak").actual(first_leak_info).is_not_none()
# Конвертируем время обнаружения в московское время
leak_detected_at = t_utils.ensure_moscow_timezone(first_leak_info.detectedAt)
leak_wait_start_time, leak_wait_end_time = t_utils.get_leak_time_window(
imitator_start_time,
leak.leak_start_interval_seconds,
leak.allowed_time_diff_seconds,
detected_at_tz=leak_detected_at.tzinfo,
)
leak_volume_m3 = t_utils.convert_leak_volume_m3(first_leak_info.leakVolume)
leak_coordinate_round = round(first_leak_info.leakCoordinate, cfg.precision)
leak_algorithm_type = ReservedType(first_leak_info.type) if first_leak_info.type else None
leak_confirmation_status = (
ConfirmationStatus(first_leak_info.confirmationStatus) if first_leak_info.confirmationStatus else None
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_payload.replyContent.tuId
).expected(cfg.tu_id).equal_to()
StepCheck("Проверка наличия названия участка утечки", "diagnosticAreaName", soft_failures).actual(
first_leak_info.diagnosticAreaName
).is_not_none()
StepCheck("Проверка статуса утечки", "confirmationStatus", soft_failures).actual(
leak_confirmation_status
).expected(leak.expected_leak_status).equal_to()
StepCheck("Проверка источника события (алгоритм)", "type", soft_failures).actual(leak_algorithm_type).expected(
leak.expected_algorithm_type
).equal_to()
StepCheck("Проверка наличия id утечки", "id", soft_failures).actual(first_leak_info.id).is_not_none()
StepCheck("Проверка координаты утечки", "leakCoordinate", soft_failures).actual(
leak_coordinate_round
).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
StepCheck("Проверка времени обнаружения утечки", "leakDetectedAt", soft_failures).actual(
leak_detected_at
).is_between(leak_wait_start_time, leak_wait_end_time)
StepCheck("Проверка объема утечки", "volume", soft_failures).actual(leak_volume_m3).is_close_to(
leak.volume_m3,
leak.allowed_volume_m3,
f"значение допустимой погрешности по объему {leak.allowed_volume_m3}",
)
def possible_leak_in_journal(http_client, cfg: SmokeSuiteConfig, imitator_start_time):
"""
Проверка наличия сообщения 'Возможна утечка' в журнале.
"""
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=LEAKS"):
end_time = datetime.now()
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(messageTypes=int(MessageType.LEAKS), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step("Фильтрация сообщений по времени и technologicalSection"):
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
time_filtered = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
time_filtered.sort(key=lambda msg: t_utils.ensure_moscow_timezone(msg.time), reverse=True)
possible_leak_msg = next(
(
msg
for msg in time_filtered
if msg.technologicalSection == cfg.tu_name and msg.event == TestConst.JOURNAL_EVENT_POSSIBLE_LEAK
),
None,
)
allure.attach(
f"Всего получено сообщений: {len(messages_info)}\n"
f"После фильтрации по времени ({filter_start_msk} - {filter_end_msk}): {len(time_filtered)}\n"
f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_POSSIBLE_LEAK}': {'True' if possible_leak_msg else 'False'}",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(
f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_POSSIBLE_LEAK}'"
):
if possible_leak_msg is None:
pytest.fail(
f"Сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_POSSIBLE_LEAK}' "
f"не найдено среди {len(time_filtered)} отфильтрованных по времени сообщений"
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка статуса утечки в журнале", "event", soft_failures).actual(possible_leak_msg.event).expected(
TestConst.JOURNAL_EVENT_POSSIBLE_LEAK
).equal_to()
StepCheck("Проверка messageType", "messageType", soft_failures).actual(possible_leak_msg.messageType).expected(
TestConst.JOURNAL_MESSAGE_TYPE_LEAKS
).equal_to()
StepCheck("Проверка technologicalSection не пустой", "technologicalSection", soft_failures).actual(
possible_leak_msg.technologicalSection
).is_not_none()
StepCheck("Проверка technologicalObject не пустой", "technologicalObject", soft_failures).actual(
possible_leak_msg.technologicalObject
).is_not_none()
def leak_info_in_journal(http_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time):
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=LEAKS"):
end_time = datetime.now()
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(messageTypes=int(MessageType.LEAKS), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step("Фильтрация сообщений по времени и technologicalSection"):
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
time_filtered = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
time_filtered.sort(key=lambda msg: t_utils.ensure_moscow_timezone(msg.time), reverse=True)
leak_message = next(
(
msg
for msg in time_filtered
if msg.technologicalSection == cfg.tu_name and TestConst.JOURNAL_EVENT_DETECTED_LEAK in msg.event
),
None,
)
allure.attach(
f"Всего получено сообщений: {len(messages_info)}\n"
f"После фильтрации по времени ({filter_start_msk} - {filter_end_msk}): {len(time_filtered)}\n",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Первичная проверка после фильтрации"):
StepCheck(
f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}' "
f"и event содержит подстроку подтвержденной утечки'{TestConst.JOURNAL_EVENT_DETECTED_LEAK}'",
"event",
).actual(leak_message).is_not_none()
leak_coordinate_km, leak_volume_m3 = t_utils.parse_journal_msg_value(leak_message.value)
leak_coordinate_round = round(leak_coordinate_km * TestConst.KM_TO_METERS, TestConst.PRECISION)
leak_message_time = t_utils.ensure_moscow_timezone(leak_message.time)
with SoftAssertions() as soft_failures:
StepCheck("Проверка полученного события event", "event", soft_failures).contains(
leak_message.event, TestConst.JOURNAL_EVENT_DETECTED_LEAK
)
StepCheck("Проверка полученного ТУ", "technologicalSection", soft_failures).actual(
leak_message.technologicalSection
).expected(cfg.tu_name).equal_to()
StepCheck("Проверка типа полученного сообщения", "messageType", soft_failures).actual(
leak_message.messageType
).expected(TestConst.JOURNAL_MESSAGE_TYPE_LEAKS).equal_to()
StepCheck("Проверка имени ДУ", "technologicalObject", soft_failures).actual(
leak_message.technologicalObject
).is_not_none()
StepCheck("Проверка координаты утечки", "leakCoordinate", soft_failures).actual(
leak_coordinate_round
).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
StepCheck("Проверка времени обнаружения утечки", "leakDetectedAt", soft_failures).actual(
leak_message_time
).is_between(filter_start_msk, filter_end_msk)
StepCheck("Проверка объема утечки", "volume", soft_failures).actual(leak_volume_m3).is_close_to(
leak.volume_m3,
leak.allowed_volume_m3,
f"значение допустимой погрешности по объему {leak.allowed_volume_m3}",
)
def completed_leak_info_in_journal(http_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time):
"""
Проверка наличия сообщения 'Утечка завершена' в журнале.
"""
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=LEAKS"):
end_time = datetime.now()
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(messageTypes=int(MessageType.LEAKS), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step("Фильтрация сообщений по времени и technologicalSection"):
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
time_filtered = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
time_filtered.sort(key=lambda msg: t_utils.ensure_moscow_timezone(msg.time), reverse=True)
completed_leak_message = next(
(
msg
for msg in time_filtered
if msg.technologicalSection == cfg.tu_name and msg.event == TestConst.JOURNAL_EVENT_COMPLETED_LEAKS
),
None,
)
allure.attach(
f"Всего получено сообщений: {len(messages_info)}\n"
f"После фильтрации по времени ({filter_start_msk} - {filter_end_msk}): {len(time_filtered)}\n",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Первичная проверка после фильтрации"):
StepCheck(
f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_COMPLETED_LEAKS}'",
"event",
).actual(completed_leak_message).is_not_none()
leak_coordinate_km, leak_volume_m3 = t_utils.parse_journal_msg_value(completed_leak_message.value)
leak_coordinate_round = round(leak_coordinate_km * TestConst.KM_TO_METERS, TestConst.PRECISION)
leak_message_time = t_utils.ensure_moscow_timezone(completed_leak_message.time)
with SoftAssertions() as soft_failures:
StepCheck("Проверка статуса утечки в журнале", "event", soft_failures).actual(
completed_leak_message.event
).expected(TestConst.JOURNAL_EVENT_COMPLETED_LEAKS).equal_to()
StepCheck("Проверка полученного ТУ", "technologicalSection", soft_failures).actual(
completed_leak_message.technologicalSection
).expected(cfg.tu_name).equal_to()
StepCheck("Проверка координаты утечки", "leakCoordinate", soft_failures).actual(
leak_coordinate_round
).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
StepCheck("Проверка времени завершения утечки", "leakDetectedAt", soft_failures).actual(
leak_message_time
).is_between(filter_start_msk, filter_end_msk)
async def all_leaks_info(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time):
"""
Проверка сообщения AllLeaksInfo об утечке.
"""
with allure.step("Подключение по ws и получение сообщения об утечке типа: AllLeaksInfoContent"):
parsed_payload = await t_utils.connect_and_get_parsed_msg_by_tu_id(
cfg.tu_id,
ws_client,
"AllLeaksInfoContent",
"subscribeAllLeaksInfoRequest",
[],
)
with allure.step("Извлечение и подготовка данных для проверки"):
leaks_info = getattr(parsed_payload.replyContent, 'leaksInfo', [])
first_leak_info = t_utils.find_leak_by_coordinate(leaks_info, leak.coordinate_meters)
StepCheck("Проверка наличия сообщения об утечке типа AllLeaksInfoContent", "leaksInfo").actual(
first_leak_info
).is_not_none()
# Конвертируем время обнаружения в московское время
leak_detected_at = t_utils.ensure_moscow_timezone(first_leak_info.leakDetectedAt)
leak_wait_start_time, leak_wait_end_time = t_utils.get_leak_time_window(
imitator_start_time,
leak.leak_start_interval_seconds,
leak.allowed_time_diff_seconds,
detected_at_tz=leak_detected_at.tzinfo,
)
leak_volume_m3 = t_utils.convert_leak_volume_m3(first_leak_info.volume)
leak_coordinate_round = round(first_leak_info.leakCoordinate, cfg.precision)
leak_lds_status = LdsStatus(first_leak_info.ldsStatus) if first_leak_info.ldsStatus else None
leak_stationary_status = (
StationaryStatus(first_leak_info.stationaryStatus) if first_leak_info.stationaryStatus else None
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_payload.replyContent.tuId
).expected(cfg.tu_id).equal_to()
StepCheck("Проверка наличия названия участка утечки", "diagnosticAreaName", soft_failures).actual(
first_leak_info.diagnosticAreaName
).is_not_none()
StepCheck("Проверка статуса СОУ", "ldsStatus", soft_failures).actual(leak_lds_status).expected(
leak.expected_lds_status
).equal_to()
StepCheck("Проверка маскирования утечки", "isMasked", soft_failures).actual(first_leak_info.isMasked).expected(
False
).equal_to()
StepCheck("Проверка квитирования утечки", "isAcknowledged", soft_failures).actual(
first_leak_info.isAcknowledged
).expected(False).equal_to()
StepCheck("Проверка наличия id утечки", "id", soft_failures).actual(first_leak_info.id).is_not_none()
StepCheck("Проверка координаты утечки", "leakCoordinate", soft_failures).actual(
leak_coordinate_round
).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
StepCheck("Проверка времени обнаружения утечки", "leakDetectedAt", soft_failures).actual(
leak_detected_at
).is_between(leak_wait_start_time, leak_wait_end_time)
StepCheck("Проверка объема утечки", "volume", soft_failures).actual(leak_volume_m3).is_close_to(
leak.volume_m3,
leak.allowed_volume_m3,
f"значение допустимой погрешности по объему {leak.allowed_volume_m3}",
)
StepCheck("Проверка режима ТУ", "stationaryStatus", soft_failures).actual(leak_stationary_status).expected(
leak.expected_stationary_status
).equal_to()
async def all_leaks_is_empty(ws_client, cfg: SmokeSuiteConfig):
"""
Проверка отсутствия информации об утечке
"""
with allure.step("Подключение по ws и получение сообщения об утечке типа: AllLeaksInfoContent"):
parsed_payload = await t_utils.connect_and_get_parsed_msg_by_tu_id(
cfg.tu_id,
ws_client,
"AllLeaksInfoContent",
"subscribeAllLeaksInfoRequest",
[],
)
with allure.step("Извлечение и подготовка данных для проверки"):
leaks_info = getattr(parsed_payload.replyContent, 'leaksInfo', [])
StepCheck("Проверка отсутствия информации об утечке в сообщении AllLeaksInfoContent", "leaksInfo").actual(
leaks_info
).is_empty()
async def tu_leaks_info(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time):
"""
Проверка сообщения TuLeaksInfo об утечке.
"""
with allure.step("Подключение по ws и получение сообщения об утечке типа: TuLeaksInfoContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"TuLeaksInfoContent",
"subscribeTuLeaksInfoRequest",
{'tuId': cfg.tu_id},
)
parsed_payload = parser.parse_tu_leaks_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
tu_leaks_info_list = getattr(parsed_payload.replyContent, 'leaksInfo', [])
first_leak_info = t_utils.find_leak_by_coordinate(tu_leaks_info_list, leak.coordinate_meters)
StepCheck("Проверка наличия сообщения об утечке типа TuLeaksInfoContent", "leaksInfo").actual(
first_leak_info
).is_not_none()
# Конвертируем время обнаружения в московское время
leak_detected_at = t_utils.ensure_moscow_timezone(first_leak_info.leakDetectedAt)
leak_wait_start_time, leak_wait_end_time = t_utils.get_leak_time_window(
imitator_start_time,
leak.leak_start_interval_seconds,
leak.allowed_time_diff_seconds,
detected_at_tz=leak_detected_at.tzinfo,
)
leak_volume_m3 = t_utils.convert_leak_volume_m3(first_leak_info.volume)
leak_coordinate_round = round(first_leak_info.leakCoordinate, cfg.precision)
leak_lds_status = LdsStatus(first_leak_info.ldsStatus) if first_leak_info.ldsStatus else None
leak_stationary_status = (
StationaryStatus(first_leak_info.stationaryStatus) if first_leak_info.stationaryStatus else None
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_payload.replyContent.tuId
).expected(cfg.tu_id).equal_to()
StepCheck("Проверка наличия id участка утечки", "controlledSiteId", soft_failures).actual(
first_leak_info.controlledSiteId
).is_not_none()
StepCheck("Проверка статуса СОУ", "ldsStatus", soft_failures).actual(leak_lds_status).expected(
leak.expected_lds_status
).equal_to()
StepCheck("Проверка маскирования утечки", "isMasked", soft_failures).actual(first_leak_info.isMasked).expected(
False
).equal_to()
StepCheck("Проверка наличия pipeId в сообщении", "pipeId", soft_failures).actual(
first_leak_info.pipeId
).is_not_none()
StepCheck("Проверка наличия id утечки", "id", soft_failures).actual(first_leak_info.id).is_not_none()
StepCheck("Проверка координаты утечки", "leakCoordinate", soft_failures).actual(
leak_coordinate_round
).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
StepCheck("Проверка времени обнаружения утечки", "leakDetectedAt", soft_failures).actual(
leak_detected_at
).is_between(leak_wait_start_time, leak_wait_end_time)
StepCheck("Проверка объема утечки", "volume", soft_failures).actual(leak_volume_m3).is_close_to(
leak.volume_m3,
leak.allowed_volume_m3,
f"значение допустимой погрешности по объему {leak.allowed_volume_m3}",
)
StepCheck("Проверка режима ТУ", "stationaryStatus", soft_failures).actual(leak_stationary_status).expected(
leak.expected_stationary_status
).equal_to()
async def lds_status_during_leak(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig):
"""
Проверка режима работы СОУ во время утечки.
"""
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"CommonSchemeContent",
"SubscribeCommonSchemeRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
parsed_payload = parser.parse_common_scheme_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
flow_areas = getattr(parsed_payload.replyContent, 'flowAreas', [])
status_config = leak.lds_status_during_leak_config
if status_config is None:
pytest.fail("Не задан leak.lds_status_during_leak_config для теста lds_status_during_leak")
leak_diagnostic_area_name = status_config.leak_diagnostic_area_name
leak_diagnostic_area_id, neighbor_ids = TestConst.DIAGNOSTIC_AREA_BASE_IDS.get(
leak_diagnostic_area_name, (None, None)
)
leak_diagnostic_area = t_utils.find_diagnostic_area_by_id(flow_areas, leak_diagnostic_area_id)
leak_lds_status_int = getattr(leak_diagnostic_area, 'ldsStatus', None)
leak_lds_status = LdsStatus(leak_lds_status_int) if leak_lds_status_int else None
with SoftAssertions() as soft_failures:
StepCheck(
f"Проверка режима работы СОУ на ДУ с утечкой, pipe_id ДУ: {leak_diagnostic_area_id}",
"ldsStatus",
soft_failures,
).actual(leak_lds_status).expected(status_config.leak_du_expected_lds_status).equal_to()
if neighbor_ids:
found_neighbor_count = 0
for neighbor_id in neighbor_ids:
neighbor_area = t_utils.find_diagnostic_area_by_id(flow_areas, neighbor_id)
neighbor_lds_status_int = getattr(neighbor_area, 'ldsStatus', None)
neighbor_lds_status = LdsStatus(neighbor_lds_status_int) if neighbor_lds_status_int else None
if neighbor_lds_status:
found_neighbor_count += 1
StepCheck(
f"Проверка режима работы СОУ на соседнем ДУ, id ДУ: {neighbor_id}",
"ldsStatus",
soft_failures,
).actual(neighbor_lds_status).expected(status_config.neighbors_du_expected_lds_status).equal_to()
if found_neighbor_count == 0:
pytest.fail(f"Не найдены соседние с утечкой ДУ по _id: {neighbor_ids}")
async def acknowledge_leak_info(ws_client, http_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig = None):
"""
Проверка квитирования утечки.
Для multi-leak наборов: после квитирования проверяется что утечка удалена из списка.
Для single-leak наборов: проверяется что список утечек пуст.
"""
with allure.step("Получение id утечки"):
with allure.step("Подключение по ws, получение и обработка сообщения об утечке типа: TuLeaksInfoContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"TuLeaksInfoContent",
"subscribeTuLeaksInfoRequest",
{'tuId': cfg.tu_id},
)
parsed_payload = parser.parse_tu_leaks_info_msg(payload)
with allure.step("Извлечение и подготовка данных для получения id утечки"):
leaks_info = getattr(parsed_payload.replyContent, 'leaksInfo', [])
leak_to_ack = t_utils.find_leak_by_coordinate(leaks_info, leak.coordinate_meters)
StepCheck("Проверка наличия сообщения об утечке", "leaksInfo").actual(leak_to_ack).is_not_none()
acknowledged_leak_id = leak_to_ack.id
with allure.step("Http запрос на квитирование утечки типа: AcknowledgeLeakRequest"):
request_body = {'leakId': str(acknowledged_leak_id), 'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.ACKNOWLEDGE_LEAK_URL_PATH, request_body)
acknowledge_reply_status = response.status_code
with allure.step(
"Подключение по ws и получение сообщения об утечке типа: AllLeaksInfoContent для проверки квитирования"
):
with allure.step("Очистка очереди websocket сообщений"):
ws_client.clear_queue()
time.sleep(cfg.basic_message_timeout)
parsed_payload = await t_utils.connect_and_get_parsed_msg_by_tu_id(
cfg.tu_id,
ws_client,
"AllLeaksInfoContent",
"subscribeAllLeaksInfoRequest",
[],
)
with allure.step("Извлечение и подготовка данных для проверки"):
remaining_leaks = getattr(parsed_payload.replyContent, 'leaksInfo', None)
remaining_leak_ids = [leak.id for leak in remaining_leaks] if remaining_leaks else []
StepCheck("Проверка кода ответа на запрос о квитировании", "replyStatus").actual(acknowledge_reply_status).expected(
ReplyStatus.OK.value
).equal_to()
# Проверяем что квитированная утечка исчезла из списка
StepCheck("Проверка отсутствия квитированной утечки в списке AllLeaksInfo", "id").does_not_contain(
remaining_leak_ids, acknowledged_leak_id
)
def acknowledge_leak_in_journal(http_client, cfg: SmokeSuiteConfig, imitator_start_time):
"""
Проверка записи в журнале о квитировании утечки.
"""
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=LEAKS"):
end_time = datetime.now()
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(messageTypes=int(MessageType.LEAKS), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step("Фильтрация сообщений по времени и technologicalSection"):
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
time_filtered = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
time_filtered.sort(key=lambda msg: t_utils.ensure_moscow_timezone(msg.time), reverse=True)
ack_message = next(
(
msg
for msg in time_filtered
if msg.technologicalSection == cfg.tu_name and msg.event == TestConst.JOURNAL_EVENT_LEAK_ACKNOWLEDGED
),
None,
)
allure.attach(
f"Всего получено сообщений: {len(messages_info)}\n"
f"После фильтрации по времени ({filter_start_msk} - {filter_end_msk}): {len(time_filtered)}\n"
f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_LEAK_ACKNOWLEDGED}': {'True' if ack_message else 'False'}",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(
f"Проверка: найдено ли сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_LEAK_ACKNOWLEDGED}'"
):
if ack_message is None:
pytest.fail(
f"Сообщение с technologicalSection='{cfg.tu_name}' "
f"и event='{TestConst.JOURNAL_EVENT_LEAK_ACKNOWLEDGED}' "
f"не найдено среди {len(time_filtered)} отфильтрованных по времени сообщений"
)
with allure.step("Проверка актуальности сообщения"):
msg_time_msk = t_utils.ensure_moscow_timezone(ack_message.time)
start_time_msk = t_utils.localize_as_moscow(imitator_start_time)
StepCheck(
"Проверка: время сообщения позднее времени старта имитатора",
"time",
).actual(
msg_time_msk > start_time_msk
).expected(True).equal_to()
with SoftAssertions() as soft_failures:
StepCheck("Проверка event", "event", soft_failures).actual(ack_message.event).expected(
TestConst.JOURNAL_EVENT_LEAK_ACKNOWLEDGED
).equal_to()
StepCheck("Проверка technologicalSection", "technologicalSection", soft_failures).actual(
ack_message.technologicalSection
).expected(cfg.tu_name).equal_to()
StepCheck("Проверка technologicalObject не пустой", "technologicalObject", soft_failures).actual(
ack_message.technologicalObject
).is_not_none()
async def output_signals(ws_client, http_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time):
"""
Проверка наличия данных об утечке в выходных сигналах.
"""
linear_part_id = leak.linear_part_id
with allure.step("Http запрос, получение и обработка ответа типа: GetOutputSignalsRequest"):
request_body = {'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.GET_OUTPUT_SIGNALS_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_output_signals_msg(payload)
# Получение данных линейного участка утечки по id
with allure.step("Извлечение и подготовка данных для списка выходных сигналов"):
linear_part_signals = getattr(parsed_payload.replyContent, 'linearPartSignals', [])
leak_linear_part = t_utils.find_object_by_field(
linear_part_signals,
TestConst.LEAK_LINEAR_PART_ID_KEY,
linear_part_id,
)
StepCheck("Проверка наличия данных выходных сигналов", "linearPartSignals").actual(
leak_linear_part
).is_not_none()
with allure.step("Получение типов выходных сигналов из обработанных данных"):
leak_signals_list = leak_linear_part.signals
ack_leak_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_ACK_LEAK
)
leak_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_LEAK
)
mask_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_MASK
)
point_leak_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_POINT_LEAK
)
q_leak_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_Q_LEAK
)
time_leak_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_TIME_LEAK
)
with allure.step(f"Получение данных выходных сигналов для линейного участка с id: {linear_part_id}"):
with allure.step("Получение сообщения с данными выходных сигналов типа: OutputSignalsInfo"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"OutputSignalsInfo",
"SubscribeOutputSignalsRequest",
{
'objects': {
'linearParts': [{'linearPartId': linear_part_id}],
'controlledSites': [],
},
'signalTypes': 1023,
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_output_signals_info_msg(payload)
with allure.step("Извлечение и подготовка данных Выходных сигналов для проверки"):
linear_part_signals = getattr(parsed_payload.replyContent, 'linearPartSignals', [])
leak_linear_part = t_utils.find_object_by_field(
linear_part_signals,
TestConst.LEAK_LINEAR_PART_ID_KEY,
linear_part_id,
)
leak_signals_list = getattr(leak_linear_part, 'signals', [])
StepCheck(
f"Проверка наличия данных выходных сигналов линейного участка {linear_part_id}", "linearPartSignals"
).actual(leak_signals_list).is_not_empty()
ack_leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, ack_leak_signal_type)
leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, leak_signal_type)
mask_leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, mask_signal_type)
point_leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, point_leak_signal_type)
q_leak_leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, q_leak_signal_type)
time_leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, time_leak_signal_type).strip()
if not time_leak_value:
pytest.fail(f"В данных линейного участка: {leak_signals_list} \n Отсутствуют данные об утечке")
time_leak_value_datetime = t_utils.to_moscow_timezone(time_leak_value)
leak_wait_start_time, leak_wait_end_time = t_utils.get_leak_time_window(
imitator_start_time,
leak.leak_start_interval_seconds,
leak.output_allowed_time_diff_seconds,
detected_at_tz=time_leak_value_datetime.tzinfo,
)
q_leak_value_m3 = t_utils.convert_leak_volume_m3(float(q_leak_leak_value))
point_leak_value_round = round(float(point_leak_value), cfg.precision)
with SoftAssertions() as soft_failures:
StepCheck("Проверка сигнала квитирования утечки", TestConst.ADDRESS_SUFFIX_ACK_LEAK, soft_failures).actual(
ack_leak_value
).expected(TestConst.OUTPUT_IS_ACK_LEAK).equal_to()
StepCheck("Проверка сигнала наличия утечки", TestConst.ADDRESS_SUFFIX_LEAK, soft_failures).actual(
leak_value
).expected(TestConst.OUTPUT_IS_LEAK).equal_to()
StepCheck("Проверка сигнала маскирования утечки", TestConst.ADDRESS_SUFFIX_MASK, soft_failures).actual(
mask_leak_value
).expected(TestConst.OUTPUT_IS_NOT_MASK).equal_to()
StepCheck("Проверка сигнала координаты утечки", TestConst.ADDRESS_SUFFIX_POINT_LEAK, soft_failures).actual(
point_leak_value_round
).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
StepCheck("Проверка сигнала объема утечки", TestConst.ADDRESS_SUFFIX_Q_LEAK, soft_failures).actual(
q_leak_value_m3
).is_close_to(
leak.volume_m3,
leak.allowed_volume_m3,
f"значение допустимой погрешности по объему {leak.allowed_volume_m3}",
)
StepCheck("Проверка времени обнаружения утечки", TestConst.ADDRESS_SUFFIX_TIME_LEAK, soft_failures).actual(
time_leak_value_datetime
).is_between(leak_wait_start_time, leak_wait_end_time)
async def balance_algorithm_leak_waiting(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time):
"""
Проверка подозрения утечки через BalanceAlgorithmResults
Логика:
- Подписка на BalanceAlgorithmResults однократно
- Раз в BALANCE_ALGORITHM_POLL_INTERVAL секунд забираем из очереди свежее сообщение
- Собираем все diagnosticAreas (только из flowAreas с непустым списком)
- Проверяем, что на ДУ с утечкой хотя бы раз пришёл isLeakPossible=True
- Проверяем, что на всех остальных ДУ isLeakPossible всегда False
- Проверяем дебаланс на ДУ с будущей утечкой, дебаланс должен быть выше значения порога - 20%
"""
poll_interval = TestConst.BALANCE_ALGORITHM_POLL_INTERVAL
total_wait = TestConst.BALANCE_ALGORITHM_TOTAL_WAIT
end_time = imitator_start_time + timedelta(
seconds=leak.balance_algorithm_leak_waiting_test.offset * 60 + total_wait
)
with allure.step(
f"Подписка и сбор BalanceAlgorithmResults раз в {poll_interval} с, в течение {total_wait} с после начала утечки"
):
await t_utils.connect(
ws_client,
"SubscribeBalanceAlgorithmResultsRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
collected_diagnostic_areas = await t_utils.poll_balance_algorithm_diagnostic_areas(
ws_client,
parser,
imitator_start_time,
end_time,
poll_interval,
)
if collected_diagnostic_areas is not None:
allure.attach(
str(collected_diagnostic_areas),
name="Тестируемый фрагмент ответа с бэка",
attachment_type=allure.attachment_type.TEXT,
)
diagnostic_area_names_with_possible = [
diagnostic_area.name for diagnostic_area in collected_diagnostic_areas if diagnostic_area.isLeakPossible
]
diagnostic_area_possible_leak = next(
(diagnostic_area for diagnostic_area in collected_diagnostic_areas if diagnostic_area.isLeakDetected),
None,
)
is_leak_possible_seen = any(diagnostic_area.isLeakPossible for diagnostic_area in collected_diagnostic_areas)
with SoftAssertions() as soft_failures:
StepCheck(
"Проверка: получен хотя бы один ДУ с подозрением на утечку",
"isLeakPossible",
soft_failures,
).actual(diagnostic_area_names_with_possible).is_not_empty()
StepCheck(
f"Проверка: на ДУ {str(diagnostic_area_names_with_possible)} бы раз за "
f"{TestConst.BALANCE_ALGORITHM_TOTAL_WAIT / TestConst.SEC_PER_MIN} минут приходил"
" статус 'подозрение на утечку': isLeakPossible=True",
"isLeakPossible",
soft_failures,
).actual(is_leak_possible_seen).expected(True).equal_to()
if leak.flow_rate_settings_threshold is not None and diagnostic_area_possible_leak is not None:
threshold = leak.flow_rate_settings_threshold
tolerance = TestConst.DEBALANCE_TOLERANCE
lower_bound = threshold * (1 - tolerance)
StepCheck(
f"Проверка значения дебаланса на ДУ name={diagnostic_area_possible_leak.name} с будущей утечкой"
f" в пределах {int(tolerance * 100)}% снизу от порогового значения по объему: {threshold}).",
"debalance",
soft_failures,
).actual(abs(diagnostic_area_possible_leak.debalance)).is_greater_than(lower_bound)
async def balance_algorithm_leak_detected(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig):
"""
Проверка наличия утечки (isLeakDetected) через BalanceAlgorithmResults.
Логика:
- Подписка на BalanceAlgorithmResultsContent
- Получение первого подходящего сообщения типа BalanceAlgorithmResultsContent
- Проверяем, что на ДУ с утечкой isLeakDetected=True
- Проверяем, что на всех остальных ДУ isLeakDetected=False
- Проверяем, что дебаланс на ДУ с утечкой > FLOW_RATE_SETTINGS_THRESHOLD
"""
with allure.step("Подписка и получение BalanceAlgorithmResultsContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"BalanceAlgorithmResultsContent",
"SubscribeBalanceAlgorithmResultsRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
parsed_payload = parser.parse_balance_algorithm_msg(payload)
reply_content = parsed_payload.replyContent
if not reply_content or not reply_content.flowAreas:
pytest.fail(
"В ответе с бэка в DTO BalanceAlgorithmResults отсутствуют flowAreas, "
"невозможно проверить наличие утечки"
)
all_diagnostic_areas = []
flow_areas = getattr(reply_content, 'flowAreas', [])
for flow_area in flow_areas:
if flow_area.diagnosticAreas:
all_diagnostic_areas.extend(flow_area.diagnosticAreas)
if not all_diagnostic_areas:
pytest.fail(
"В ответе с бэка в DTO BalanceAlgorithmResults во всех flowAreas отсутствуют diagnosticAreas, "
"невозможно проверить наличие утечки"
)
leak_diagnostic_area = next(
(diagnostic_area for diagnostic_area in all_diagnostic_areas if diagnostic_area.isLeakDetected),
None,
)
if leak_diagnostic_area is None:
pytest.fail("Ни одного ДУ с утечкой не найдено в ответе BalanceAlgorithmResultsContent")
leak_diagnostic_area_name = leak_diagnostic_area.name
with SoftAssertions() as soft_failures:
StepCheck(
f"Проверка: на ДУ name={leak_diagnostic_area_name} обнаружена утечка",
"isLeakDetected",
soft_failures,
).actual(leak_diagnostic_area.isLeakDetected).expected(True).equal_to()
foreign_with_detected = [
diagnostic_area
for diagnostic_area in all_diagnostic_areas
if diagnostic_area.name != leak_diagnostic_area_name and diagnostic_area.isLeakDetected
]
if not cfg.has_multiple_leaks:
StepCheck(
"Проверка: на остальных ДУ не обнаружена утечка, "
f" количество ДУ с неправильным статусом: {len(foreign_with_detected)}, "
f"их id: {[diagnostic_area.id for diagnostic_area in foreign_with_detected]})",
"isLeakDetected_without_leak",
soft_failures,
).actual(len(foreign_with_detected)).expected(0).equal_to()
if leak.flow_rate_settings_threshold is not None:
threshold = leak.flow_rate_settings_threshold
StepCheck(
f"Дебаланс на ДУ name={leak_diagnostic_area_name} по модулю больше порога для данного режима МТ:"
f" {threshold}",
"debalance",
soft_failures,
).actual(abs(leak_diagnostic_area.debalance)).is_greater_than(threshold)
async def balance_algorithm_leak_completed(ws_client, cfg: SmokeSuiteConfig):
"""
Проверка отсутствия утечки (isLeakDetected) через BalanceAlgorithmResults.
Логика:
- Подписка на BalanceAlgorithmResultsContent.
- Получение первого подходящего сообщения типа BalanceAlgorithmResultsContent.
- Проверяем, что на всех ДУ флаг isLeakDetected=False.
- Проверяем, что дебаланс на всех ДУ < FLOW_RATE_SETTINGS_THRESHOLD.
"""
with allure.step("Подписка и получение BalanceAlgorithmResultsContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"BalanceAlgorithmResultsContent",
"SubscribeBalanceAlgorithmResultsRequest",
{'tuId': cfg.tu_id, 'additionalProperties': None},
)
parsed_payload = parser.parse_balance_algorithm_msg(payload)
reply_content = parsed_payload.replyContent
if not reply_content or not reply_content.flowAreas:
pytest.fail(
"В ответе с бэка в DTO BalanceAlgorithmResults отсутствуют flowAreas, "
"невозможно проверить наличие/отсутствие утечки"
)
all_diagnostic_areas = []
flow_areas = getattr(reply_content, 'flowAreas', [])
for flow_area in flow_areas:
if flow_area.diagnosticAreas:
all_diagnostic_areas.extend(flow_area.diagnosticAreas)
if not all_diagnostic_areas:
pytest.fail(
"В ответе с бэка в DTO BalanceAlgorithmResults во всех flowAreas отсутствуют diagnosticAreas, "
"невозможно проверить наличие/отсутствие утечки"
)
with SoftAssertions() as soft_failures:
for diagnostic_area in all_diagnostic_areas:
diagnostic_area_name = diagnostic_area.name
StepCheck(
f"Проверка: на ДУ {diagnostic_area_name} не должно быть утечки",
"isLeakDetected_without_leak",
soft_failures,
).actual(diagnostic_area.isLeakDetected).expected(False).equal_to()
async def leak_is_complete_on_kg(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig):
"""
Проверка факта завершения утечки на ЭФ КГ(табличное представление).
Логика:
LeaksContent - проверить, что утечка в статусе завершена
"""
with allure.step("Подключение по ws и получение сообщения об утечке типа: LeaksContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"LeaksContent",
"SubscribeLeaksRequest",
{'tuId': cfg.tu_id},
)
parsed_payload = parser.parse_leaks_content_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
leaks_list_info = getattr(parsed_payload.replyContent, 'leaksListInfo', None)
complete_leak_info = t_utils.find_leak_by_coordinate(leaks_list_info, leak.coordinate_meters)
leak_coordinate_round = round(complete_leak_info.leakCoordinate, cfg.precision)
complete_leak = t_utils.find_object_by_field(
leaks_list_info, "confirmationStatus", ConfirmationStatus.CONFIRMED_AND_LEAK_CLOSED.value
)
leak_algorithm_type = ReservedType(complete_leak_info.type) if complete_leak_info.type else None
leak_confirmation_status = (
ConfirmationStatus(complete_leak.confirmationStatus) if complete_leak.confirmationStatus else None
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка статуса утечки в КГ - завершена", "confirmationStatus", soft_failures).actual(
leak_confirmation_status
).expected(leak.expected_complete_leak_status).equal_to()
StepCheck("Проверка наличия названия участка утечки", "diagnosticAreaName", soft_failures).actual(
complete_leak_info.diagnosticAreaName
).is_not_none()
StepCheck("Проверка источника события (алгоритм)", "type", soft_failures).actual(leak_algorithm_type).expected(
leak.expected_algorithm_type
).equal_to()
StepCheck("Проверка координаты утечки", "leakCoordinate", soft_failures).actual(
leak_coordinate_round
).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
async def leak_is_complete_in_output_signals(ws_client, http_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig):
"""OutputSignalsInfo - нет утечки в выходных сигналах"""
linear_part_id = leak.linear_part_id
with allure.step("Http запрос, получение и обработка ответа типа: GetOutputSignalsRequest"):
request_body = {'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.GET_OUTPUT_SIGNALS_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_output_signals_msg(payload)
# Получение данных линейного участка утечки по id
with allure.step("Извлечение и подготовка данных для проверки"):
linear_part_signals = getattr(parsed_payload.replyContent, 'linearPartSignals', [])
leak_linear_part = t_utils.find_object_by_field(
linear_part_signals,
TestConst.LEAK_LINEAR_PART_ID_KEY,
linear_part_id,
)
StepCheck("Проверка наличия данных выходных сигналов", "linearPartSignals").actual(
leak_linear_part
).is_not_none()
leak_signals_list = leak_linear_part.signals
leak_signal_type = t_utils.find_signal_type_by_address_suffix(
leak_signals_list, TestConst.ADDRESS_SUFFIX_LEAK
)
with allure.step(f"Получение данных выходных сигналов для линейного участка с id: {linear_part_id}"):
with allure.step("Получение сообщения с данными выходных сигналов типа: OutputSignalsInfo"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"OutputSignalsInfo",
"SubscribeOutputSignalsRequest",
{
'objects': {
'linearParts': [{'linearPartId': linear_part_id}],
'controlledSites': [],
},
'signalTypes': 1023,
'tuId': cfg.tu_id,
'additionalProperties': None,
},
)
parsed_payload = parser.parse_output_signals_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
linear_part_signals = getattr(parsed_payload.replyContent, 'linearPartSignals', [])
leak_linear_part = t_utils.find_object_by_field(
linear_part_signals,
TestConst.LEAK_LINEAR_PART_ID_KEY,
linear_part_id,
)
StepCheck("Проверка наличия данных выходных сигналов", "linearPartSignals").actual(
leak_linear_part
).is_not_none()
leak_signals_list = getattr(leak_linear_part, 'signals', [])
leak_value = t_utils.find_signal_val_by_signal_type(leak_signals_list, leak_signal_type)
with SoftAssertions() as soft_failures:
StepCheck(
"Проверка отсутствия времени утечки в выходных сигналах",
TestConst.ADDRESS_SUFFIX_TIME_LEAK,
soft_failures,
).actual(leak_value).expected(TestConst.OUTPUT_IS_NOT_LEAK).equal_to()
StepCheck(
"Проверка отсутствия утечки в выходных сигналах",
TestConst.ADDRESS_SUFFIX_LEAK,
soft_failures,
).actual(leak_value).expected(TestConst.OUTPUT_IS_NOT_LEAK).equal_to()
StepCheck(
"Проверка отсутствия квитирования утечки в выходных сигналах",
TestConst.ADDRESS_SUFFIX_ACK_LEAK,
soft_failures,
).actual(leak_value).expected(TestConst.OUTPUT_IS_NOT_LEAK).equal_to()
StepCheck(
"Проверка отсутствия объема утечки в выходных сигналах",
TestConst.ADDRESS_SUFFIX_Q_LEAK,
soft_failures,
).actual(leak_value).expected(TestConst.OUTPUT_IS_NOT_LEAK).equal_to()
StepCheck(
"Проверка отсутствия координаты утечки в выходных сигналах",
TestConst.ADDRESS_SUFFIX_POINT_LEAK,
soft_failures,
).actual(leak_value).expected(TestConst.OUTPUT_IS_NOT_LEAK).equal_to()
async def complete_tu_leaks_info_content(ws_client, cfg: SmokeSuiteConfig):
"""
TuLeaksInfoContent - проверка отсутствия утечки на схеме
"""
with allure.step("Подключение по ws, получение и обработка сообщения об утечке типа: TuLeaksInfoContent"):
payload = await t_utils.connect_and_subscribe_msg(
ws_client,
"TuLeaksInfoContent",
"subscribeTuLeaksInfoRequest",
{'tuId': cfg.tu_id},
)
parsed_payload = parser.parse_tu_leaks_info_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
leak_on_scheme = getattr(parsed_payload.replyContent, 'leaksInfo', [])
StepCheck("Проверка отсутствия утечки на схеме", "leaksInfo").actual(leak_on_scheme).is_empty()
async def export_leaks_report(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time: datetime):
"""
Сценарий формирования отчёта об утечках.
Этапы:
1. Подписка SubscribeReportsDataExportedRequest на пуш-нотификации.
2. Отправка ExportReportsCommandRequest с фильтром по времени
(start = старт имитатора, end = старт имитатора + offset теста).
3. Ожидание пуш-нотификации ReportDataExportedNotification о готовности отчёта.
4. Лонг-поллинг GetExportedDataListRequest до появления нашего отчёта в списке.
5. Отправка DownloadExportedDataRequest по id отчёта.
6. Получение fileChunk по ответу на скачивание.
7-10. Проверки: формат файла, имя, шапка xlsx, строка утечки.
Скачанный файл удаляется по завершению, прикладывается к Allure только при падении теста.
"""
# Создаем универсальный словарь для сбора фактических результатов ответов с бэка
actual_report_state = ExportLeaksReportState()
with allure.step("Подготовка параметров сценария формирования отчёта об утечках"):
actual_report_state.report_test = leak.export_leaks_report_test
StepCheck("В конфигурации задан export_leaks_report_test", "export_leaks_report_test").actual(
actual_report_state.report_test
).is_not_none()
actual_report_state.period_start = t_utils.localize_as_moscow(imitator_start_time)
actual_report_state.period_end = t_utils.localize_as_moscow(
imitator_start_time + timedelta(minutes=actual_report_state.report_test.offset)
)
actual_report_state.period_start_naive = report_utils.normalize_report_period_naive(
actual_report_state.period_start
)
actual_report_state.period_end_naive = report_utils.normalize_report_period_naive(
actual_report_state.period_end
)
actual_report_state.expected_mt_mode = ReportConst.STATIONARY_STATUS_TO_REPORT_TEXT.get(
leak.expected_report_stationary_status
)
actual_report_state.expected_lds_status_text = LdsStatus.report_text_by_value(
leak.expected_lds_status_in_leaks_report
)
actual_report_state.tu_description_lower = cfg.tu_name.lower()
time_offset_hours = t_utils.report_time_offset_hours()
StepCheck(
f"Смещение timeOffset для запросов отчёта (часовой пояс {TestConst.ZONE_INFO})",
"time_offset_hours",
).actual(time_offset_hours).is_not_none()
actual_report_state.time_offset_hours = time_offset_hours
StepCheck(
"Задан ожидаемый текст режима МТ для отчёта",
"expected_mt_mode",
).actual(actual_report_state.expected_mt_mode).is_not_none()
allure.attach(
f"period.start={actual_report_state.period_start}\n"
f"period.end={actual_report_state.period_end}\n"
f"offset_minutes={actual_report_state.report_test.offset}",
name="Фильтр периода отчёта",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(f"Этап 1. Подписка на пуш-нотификации ({ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST})"):
await t_utils.connect(ws_client, ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST, [])
with allure.step(f"Этап 2. Запрос формирования отчёта ({ReportConst.EXPORT_REPORTS_COMMAND_REQUEST})"):
request_payload = {
"tuId": cfg.tu_id,
"exportedDataTypes": [ExportedDataType.LEAKS_REPORT.value],
"timeOffset": actual_report_state.time_offset_hours,
"period": {
"start": t_utils.datetime_to_msgpack_timestamp(actual_report_state.period_start),
"end": t_utils.datetime_to_msgpack_timestamp(actual_report_state.period_end),
"additionalProperties": {},
},
}
await t_utils.connect(ws_client, ReportConst.EXPORT_REPORTS_COMMAND_REQUEST, request_payload)
with allure.step(
f"Этап 3. Ожидание пуш-нотификации {ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION} о готовности отчёта"
):
actual_report_state.notification = await t_utils.poll_for_report_export_notification(
ws_client=ws_client,
parser=parser,
total_wait_seconds=ReportConst.NOTIFICATION_TIMEOUT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step("Извлечение полей пуш-нотификации"):
notification = actual_report_state.notification
notification_reply_status = notification.replyStatus if notification else None
notification_reply_content = notification.replyContent if notification else None
notification_export_status = notification_reply_content.exportStatus if notification_reply_content else None
notification_error_message = (
(notification_reply_content.errorMessage or "") if notification_reply_content else ""
)
with allure.step(f"Этап 4. Лонг-поллинг {ReportConst.GET_EXPORTED_DATA_LIST_REQUEST} до появления отчёта в списке"):
actual_report_state.report_item = await t_utils.poll_for_exported_file(
ws_client=ws_client,
parser=parser,
list_limit=ReportConst.EXPORTED_DATA_LIST_LIMIT,
expected_data_type=ExportedDataType.LEAKS_REPORT,
name_substring=ReportConst.LEAKS_REPORT_NAME_PART,
tu_name_substring=cfg.tu_name,
period_start=actual_report_state.period_start,
period_end=actual_report_state.period_end,
total_wait_seconds=ReportConst.LIST_POLL_TOTAL_WAIT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step("Подготовка данных найденного отчёта в списке"):
report_item = actual_report_state.report_item
if report_item is not None:
allure.attach(
f"id={report_item.id}, name={report_item.name}, "
f"exportedDataType={report_item.exportedDataType}, "
f"start={t_utils.format_datetime_moscow(report_item.start)}, "
f"end={t_utils.format_datetime_moscow(report_item.end)}",
name="Найденный отчёт в списке",
attachment_type=allure.attachment_type.TEXT,
)
actual_report_state.report_file_name = report_utils.build_export_report_file_name(
cfg.tu_name,
actual_report_state.period_start,
actual_report_state.period_end,
)
with allure.step("Проверка: отчёт найден в списке сформированных файлов"):
StepCheck("Отчёт найден в списке сформированных файлов", "report_item").actual(
actual_report_state.report_item
).is_not_none()
with allure.step(
f"Этап 5. Streaming-вызов {ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST} по "
f"id={actual_report_state.report_item.id}"
):
download_request = {
"exportedDataId": actual_report_state.report_item.id,
"exportedDataType": ExportedDataType.LEAKS_REPORT.to_download_name(),
"additionalProperties": None,
"timeOffset": actual_report_state.time_offset_hours,
}
download_purpose = (
f"скачивание xlsx-отчёта об утечках (exportedDataId={actual_report_state.report_item.id}) "
f"после формирования отчёта и выбора файла в списке GetExportedDataListRequest - "
f"выпадашка уведомлений на UI"
)
await t_utils.connect_stream(
ws_client,
ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
download_request,
purpose=download_purpose,
)
actual_report_state.download_invocation_id = ws_client.invocation_id
with allure.step("Этап 6. Получение fileChunk - скачивание отчёта по утечкам"):
actual_report_state.download_reply = await t_utils.receive_download_exported_data_reply(
ws_client=ws_client,
parser=parser,
invocation_id=actual_report_state.download_invocation_id,
request_name=ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
total_wait_seconds=ReportConst.DOWNLOAD_TIMEOUT_SECONDS,
purpose=download_purpose,
)
with allure.step("Извлечение данных ответа на скачивание"):
download_reply = actual_report_state.download_reply
download_reply_status = download_reply.replyStatus
has_download_reply_content = download_reply.replyContent is not None
actual_report_state.file_bytes = download_reply.replyContent.fileChunk if has_download_reply_content else None
is_xlsx_signature = (
report_utils.is_xlsx_file_bytes(actual_report_state.file_bytes) if actual_report_state.file_bytes else False
)
with allure.step("Проверка ответа на скачивание и формата xlsx"):
StepCheck("Проверка статуса ответа на скачивание", "replyStatus").actual(download_reply_status).expected(
ReplyStatus.OK.value
).equal_to()
StepCheck("Проверка наличия контента ответа на скачивание", "replyContent").actual(
has_download_reply_content
).expected(True).equal_to()
StepCheck("Проверка наличия байт файла", "fileChunk").actual(actual_report_state.file_bytes).is_not_empty()
StepCheck("Проверка xlsx (zip) сигнатуры файла", "file_signature").actual(is_xlsx_signature).expected(
True
).equal_to()
with allure.step("Подготовка данных для проверки имени файла отчёта"):
report_file_name = actual_report_state.report_file_name
report_file_name_lower = report_file_name.lower()
file_name_period_start, file_name_period_end = report_utils.parse_period_from_export_file_name(report_file_name)
period_start_lo, period_start_hi, period_end_lo, period_end_hi = report_utils.report_period_comparison_bounds(
actual_report_state.period_start_naive,
actual_report_state.period_end_naive,
)
has_xlsx_extension = report_utils.is_xlsx_extension(report_file_name)
leaks_report_name_part_lower = ReportConst.LEAKS_REPORT_NAME_PART.lower()
with allure.step("Этап 8. Сохранение, обработка и проверка отчета по утечкам"):
actual_report_state.temp_file_path = report_utils.save_report_bytes_to_temp_file(actual_report_state.file_bytes)
try:
with allure.step("Проверка: временный xlsx файл создан"):
StepCheck("Временный xlsx файл создан", "temp_file_path").actual(
actual_report_state.temp_file_path
).is_not_none()
with allure.step("Этап 9. Открытие xlsx и чтение шапки"):
actual_report_state.worksheet = report_utils.load_report_worksheet(actual_report_state.temp_file_path)
actual_report_state.title_info = report_utils.parse_report_title(
report_utils.get_report_title_cell(actual_report_state.worksheet)
)
allure.attach(
f"Шапка отчёта (raw): {actual_report_state.title_info.raw_title}\n"
f"period_start: {actual_report_state.title_info.period_start}\n"
f"period_end: {actual_report_state.title_info.period_end}",
name="Шапка отчёта (1-я строка)",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных шапки отчёта для проверки"):
title_info = actual_report_state.title_info
report_title_lower = title_info.raw_title.lower()
leaks_report_name_part_lower = ReportConst.LEAKS_REPORT_NAME_PART.lower()
column_headers = report_utils.get_report_column_headers(actual_report_state.worksheet)
period_start_lo, period_start_hi, period_end_lo, period_end_hi = (
report_utils.report_period_comparison_bounds(
actual_report_state.period_start_naive,
actual_report_state.period_end_naive,
)
)
header_period_start = title_info.period_start
header_period_end = title_info.period_end
with allure.step("Этап 10. Извлечение строк данных из отчёта"):
actual_report_state.data_rows = report_utils.iter_report_data_rows(actual_report_state.worksheet)
actual_report_state.target_row = report_utils.find_row_with_object(
actual_report_state.data_rows, cfg.tu_name
)
allure.attach(
"\n".join(f"row#{row.row_index}: {row.cells}" for row in actual_report_state.data_rows),
name="Все строки данных отчёта",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных строки утечки для проверки"):
target_row = actual_report_state.target_row
leak_datetime_value = target_row.datetime_value if target_row else None
object_value_lower = target_row.object_value.lower() if target_row else ""
lds_status_value = target_row.lds_status.strip() if target_row else ""
masking_info_lower = target_row.masking_info.lower() if target_row else ""
leak_coordinate_meters = target_row.coordinate_meters if target_row else None
leak_volume_value = target_row.leak_volume if target_row else None
mt_mode_lower = target_row.mt_mode.lower() if target_row else ""
expected_mt_mode_lower = actual_report_state.expected_mt_mode.lower()
expected_lds_status_lower = actual_report_state.expected_lds_status_text.lower()
masking_not_masked_lower = ReportConst.MASKING_NOT_MASKED_TEXT.lower()
period_start_lo, period_start_hi, period_end_lo, period_end_hi = (
report_utils.report_period_comparison_bounds(
actual_report_state.period_start_naive,
actual_report_state.period_end_naive,
)
)
with allure.step("Проверка содержимого строки утечки"):
StepCheck("В отчёте есть хотя бы одна строка с данными", "data_rows").actual(
actual_report_state.data_rows
).is_not_empty()
StepCheck(
f"Строка с объектом, содержащим '{cfg.tu_name}'",
ReportConst.COL_OBJECT,
).actual(actual_report_state.target_row).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
"Время утечки в диапазоне [старт имитатора, старт + offset теста] (+-1 мин)",
ReportConst.COL_DATETIME,
soft_failures,
).actual(leak_datetime_value).is_between(period_start_lo, period_end_hi)
StepCheck(
f"Колонка '{ReportConst.COL_OBJECT}' содержит '{cfg.tu_name}'",
ReportConst.COL_OBJECT,
soft_failures,
).contains(object_value_lower, actual_report_state.tu_description_lower)
StepCheck(
f"Колонка '{ReportConst.COL_LDS_STATUS}' содержит "
f"'{actual_report_state.expected_lds_status_text}'",
ReportConst.COL_LDS_STATUS,
soft_failures,
).contains(lds_status_value.lower(), expected_lds_status_lower)
StepCheck(
f"Колонка '{ReportConst.COL_MASK_INFO}' содержит '{ReportConst.MASKING_NOT_MASKED_TEXT}'",
ReportConst.COL_MASK_INFO,
soft_failures,
).contains(masking_info_lower, masking_not_masked_lower)
StepCheck(
f"Колонка '{ReportConst.COL_COORDINATE}' (с погрешностью {cfg.allowed_distance_diff_meters} м)",
ReportConst.COL_COORDINATE,
soft_failures,
).actual(leak_coordinate_meters).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
StepCheck(
f"Колонка '{ReportConst.COL_LEAK_VOLUME}' не пустая",
ReportConst.COL_LEAK_VOLUME,
soft_failures,
).actual(leak_volume_value).is_not_none()
StepCheck(
f"Колонка '{ReportConst.COL_MT_MODE}' содержит '{actual_report_state.expected_mt_mode}'",
ReportConst.COL_MT_MODE,
soft_failures,
).contains(mt_mode_lower, expected_mt_mode_lower)
except Exception:
with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
if actual_report_state.temp_file_path and actual_report_state.report_file_name:
report_utils.attach_report_file_to_allure(
actual_report_state.temp_file_path, actual_report_state.report_file_name
)
raise
with allure.step("Проверка имени файла отчёта"):
with SoftAssertions() as soft_failures:
StepCheck(f"Имя файла оканчивается на {ReportConst.XLSX_EXTENSION}", "file_name", soft_failures).actual(
has_xlsx_extension
).expected(True).equal_to()
StepCheck(
f"Имя файла содержит '{ReportConst.LEAKS_REPORT_NAME_PART}'", "file_name", soft_failures
).contains(report_file_name_lower, leaks_report_name_part_lower)
StepCheck(f"Имя файла содержит описание ТУ '{cfg.tu_name}'", "file_name", soft_failures).contains(
report_file_name_lower, actual_report_state.tu_description_lower
)
StepCheck(
"Дата начала периода в имени файла совпадает с фильтром запроса (+-1 мин)",
"period_start_in_file_name",
soft_failures,
).actual(file_name_period_start).is_between(period_start_lo, period_start_hi)
StepCheck(
"Дата конца периода в имени файла совпадает с фильтром запроса (+-1 мин)",
"period_end_in_file_name",
soft_failures,
).actual(file_name_period_end).is_between(period_end_lo, period_end_hi)
with allure.step("Проверка двойной шапки отчёта"):
StepCheck("Лист xlsx открыт", "worksheet").actual(actual_report_state.worksheet).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
f"В шапке отчёта присутствует '{ReportConst.LEAKS_REPORT_NAME_PART}'",
"report_title",
soft_failures,
).contains(report_title_lower, leaks_report_name_part_lower)
StepCheck(
"Время начала периода в шапке совпадает с фильтром запроса (+-1 мин)",
"period_start",
soft_failures,
).actual(header_period_start).is_between(period_start_lo, period_start_hi)
StepCheck(
"Время конца периода в шапке совпадает с фильтром запроса (+-1 мин)",
"period_end",
soft_failures,
).actual(header_period_end).is_between(period_end_lo, period_end_hi)
StepCheck(
"Названия колонок в шапке отчёта",
"column_headers",
soft_failures,
).actual(
column_headers
).expected(ReportConst.EXPECTED_COLUMN_HEADERS).equal_to()
with allure.step("Проверка пуш-нотификации о готовности отчёта"):
with SoftAssertions() as soft_failures:
StepCheck("Получена пуш-нотификация о готовности отчёта", "notification", soft_failures).actual(
actual_report_state.notification
).is_not_none()
StepCheck("Проверка статуса пуш-нотификации", "replyStatus", soft_failures).actual(
notification_reply_status
).expected(ReplyStatus.OK.value).equal_to()
StepCheck("Проверка наличия контента нотификации", "replyContent", soft_failures).actual(
notification_reply_content
).is_not_none()
StepCheck("Проверка exportStatus в нотификации", "exportStatus", soft_failures).actual(
notification_export_status
).expected(ExportStatus.DONE).equal_to()
StepCheck("В нотификации нет текста ошибки", "errorMessage", soft_failures).actual(
notification_error_message
).is_empty()
async def export_lds_status_report(ws_client, cfg: SmokeSuiteConfig, imitator_start_time: datetime):
"""
Сценарий формирования xlsx-отчёта о режиме работы СОУ.
"""
report_state = ExportLdsStatusReportState()
with allure.step("Подготовка параметров сценария формирования отчёта о режиме работы СОУ"):
report_state.report_test = cfg.export_lds_status_report_test
StepCheck("В конфигурации задан export_lds_status_report_test", "export_lds_status_report_test").actual(
report_state.report_test
).is_not_none()
report_state.period_start = t_utils.localize_as_moscow(imitator_start_time)
report_state.period_end = t_utils.localize_as_moscow(
imitator_start_time + timedelta(minutes=report_state.report_test.offset)
)
report_state.period_start_naive = report_utils.normalize_report_period_naive(report_state.period_start)
report_state.period_end_naive = report_utils.normalize_report_period_naive(report_state.period_end)
report_state.tu_description_lower = cfg.tu_name.lower()
time_offset_hours = t_utils.report_time_offset_hours()
StepCheck(
f"Смещение timeOffset для запросов отчёта (часовой пояс {TestConst.ZONE_INFO})",
"time_offset_hours",
).actual(time_offset_hours).is_not_none()
report_state.time_offset_hours = time_offset_hours
allure.attach(
f"period.start={report_state.period_start}\n"
f"period.end={report_state.period_end}\n"
f"offset_minutes={report_state.report_test.offset}\n"
f"sections={LdsReportConst.SECTION_NAMES}",
name="Фильтр периода отчёта о режиме СОУ",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(f"Этап 1. Подписка на пуш-нотификации {ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST}"):
await t_utils.connect(ws_client, ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST, [])
with allure.step(f"Этап 2. Запрос формирования отчёта {ReportConst.EXPORT_REPORTS_COMMAND_REQUEST}"):
request_payload = {
"tuId": cfg.tu_id,
"exportedDataTypes": [ExportedDataType.LDS_STATUS_REPORT.value],
"timeOffset": report_state.time_offset_hours,
"period": {
"start": t_utils.datetime_to_msgpack_timestamp(report_state.period_start),
"end": t_utils.datetime_to_msgpack_timestamp(report_state.period_end),
"additionalProperties": {},
},
}
await t_utils.connect(ws_client, ReportConst.EXPORT_REPORTS_COMMAND_REQUEST, request_payload)
with allure.step(
f"Этап 3. Ожидание пуш-нотификации {ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION} о готовности отчёта"
):
report_state.notification = await t_utils.poll_for_report_export_notification(
ws_client=ws_client,
parser=parser,
total_wait_seconds=ReportConst.NOTIFICATION_TIMEOUT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step(f"Этап 4. Лонг-поллинг {ReportConst.GET_EXPORTED_DATA_LIST_REQUEST} до появления отчёта в списке"):
report_state.report_item = await t_utils.poll_for_exported_file(
ws_client=ws_client,
parser=parser,
list_limit=ReportConst.EXPORTED_DATA_LIST_LIMIT,
expected_data_type=ExportedDataType.LDS_STATUS_REPORT,
name_substring=LdsReportConst.LDS_STATUS_REPORT_NAME_PART,
tu_name_substring=cfg.tu_name,
period_start=report_state.period_start,
period_end=report_state.period_end,
total_wait_seconds=ReportConst.LIST_POLL_TOTAL_WAIT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step("Подготовка данных найденного отчёта в списке"):
report_item = report_state.report_item
if report_item is not None:
allure.attach(
f"id={report_item.id}, name={report_item.name}, "
f"exportedDataType={report_item.exportedDataType}, "
f"start={t_utils.format_datetime_moscow(report_item.start)}, "
f"end={t_utils.format_datetime_moscow(report_item.end)}",
name="Найденный отчёт в списке",
attachment_type=allure.attachment_type.TEXT,
)
report_state.report_file_name = report_utils.build_export_report_file_name(
cfg.tu_name,
report_state.period_start,
report_state.period_end,
LdsReportConst.LDS_STATUS_REPORT_NAME_PART,
". ",
)
with allure.step("Проверка: отчёт найден в списке сформированных файлов"):
StepCheck("Отчёт найден в списке сформированных файлов", "report_item").actual(
report_state.report_item
).is_not_none()
with allure.step(
f"Этап 5. Streaming-вызов {ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST} по id={report_state.report_item.id}"
):
download_request = {
"exportedDataId": report_state.report_item.id,
"exportedDataType": ExportedDataType.LDS_STATUS_REPORT.to_download_name(),
"additionalProperties": None,
"timeOffset": report_state.time_offset_hours,
}
download_purpose = (
f"скачивание xlsx-отчёта о режиме СОУ (exportedDataId={report_state.report_item.id}) "
f"после формирования отчёта и выбора файла в списке GetExportedDataListRequest"
)
await t_utils.connect_stream(
ws_client,
ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
download_request,
purpose=download_purpose,
)
report_state.download_invocation_id = ws_client.invocation_id
with allure.step("Этап 6. Получение fileChunk - скачивание отчёта о режиме СОУ"):
report_state.download_reply = await t_utils.receive_download_exported_data_reply(
ws_client=ws_client,
parser=parser,
invocation_id=report_state.download_invocation_id,
request_name=ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
total_wait_seconds=ReportConst.DOWNLOAD_TIMEOUT_SECONDS,
purpose=download_purpose,
)
with allure.step("Извлечение данных ответа на скачивание"):
download_reply = report_state.download_reply
download_reply_status = download_reply.replyStatus
has_download_reply_content = download_reply.replyContent is not None
report_state.file_bytes = download_reply.replyContent.fileChunk if has_download_reply_content else None
is_xlsx_signature = (
report_utils.is_xlsx_file_bytes(report_state.file_bytes) if report_state.file_bytes else False
)
with allure.step("Проверка ответа на скачивание и формата xlsx"):
StepCheck("Проверка статуса ответа на скачивание", "replyStatus").actual(download_reply_status).expected(
ReplyStatus.OK.value
).equal_to()
StepCheck("Проверка наличия контента ответа на скачивание", "replyContent").actual(
has_download_reply_content
).expected(True).equal_to()
StepCheck("Проверка наличия байт файла", "fileChunk").actual(report_state.file_bytes).is_not_empty()
StepCheck("Проверка xlsx (zip) сигнатуры файла", "file_signature").actual(is_xlsx_signature).expected(
True
).equal_to()
with allure.step("Подготовка данных для проверки имени файла отчёта"):
report_file_name = report_state.report_file_name
report_file_name_lower = report_file_name.lower()
file_name_period_start, file_name_period_end = report_utils.parse_period_from_export_file_name(
report_file_name,
LdsReportConst.REPORT_FILE_NAME_PERIOD_PATTERN,
)
period_start_lo, period_start_hi, period_end_lo, period_end_hi = report_utils.report_period_comparison_bounds(
report_state.period_start_naive,
report_state.period_end_naive,
)
has_xlsx_extension = report_utils.is_xlsx_extension(report_file_name)
lds_report_name_part_lower = LdsReportConst.LDS_STATUS_REPORT_NAME_PART.lower()
try:
with allure.step("Этап 7. Сохранение и разбор xlsx-отчёта о режиме СОУ"):
report_state.temp_file_path = report_utils.save_report_bytes_to_temp_file(
report_state.file_bytes,
prefix="lds_status_report_",
)
StepCheck("Временный xlsx файл создан", "temp_file_path").actual(report_state.temp_file_path).is_not_none()
report_state.worksheet = report_utils.load_report_worksheet(report_state.temp_file_path)
report_state.parsed_report = lds_report_utils.parse_lds_status_report_worksheet(
report_state.worksheet,
LdsReportConst.SECTION_NAMES,
)
allure.attach(
f"Шапка (raw): {report_state.parsed_report.title_info.raw_title}\n"
f"period_start: {report_state.parsed_report.title_info.period_start}\n"
f"period_end: {report_state.parsed_report.title_info.period_end}\n"
f"total_duration: {report_state.parsed_report.total_duration_raw}",
name="Шапка отчёта о режиме СОУ",
attachment_type=allure.attachment_type.TEXT,
)
allure.attach(
lds_report_utils.format_section_rows_for_allure(report_state.parsed_report.section_rows),
name="Строки участков отчёта",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных таблицы отчёта для проверки"):
parsed_report = report_state.parsed_report
expected_section_names = LdsReportConst.SECTION_NAMES
section_rows = parsed_report.section_rows
total_duration_seconds = parsed_report.total_duration_seconds
duration_tolerance = LdsReportConst.TOTAL_DURATION_TOLERANCE_SECONDS
with allure.step("Проверка содержимого таблицы отчёта о режиме СОУ"):
StepCheck("Лист xlsx открыт", "worksheet").actual(report_state.worksheet).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
"Количество строк участков в отчёте",
"section_rows_count",
soft_failures,
).actual(
len(section_rows)
).expected(len(expected_section_names)).equal_to()
for section_index, expected_section_name in enumerate(expected_section_names):
actual_section_name = (
section_rows[section_index].section_name if section_index < len(section_rows) else None
)
StepCheck(
f"Наименование участка #{section_index + 1}",
LdsReportConst.COL_SECTION,
soft_failures,
).actual(actual_section_name).expected(expected_section_name).equal_to()
for section_row in section_rows:
for column_name in LdsReportConst.MODE_DURATION_COLUMNS:
cell_value = section_row.cells.get(column_name)
StepCheck(
f"Длительность '{column_name}' для участка '{section_row.section_name}' заполнена",
column_name,
soft_failures,
).actual(lds_report_utils.is_duration_cell_filled(cell_value)).expected(True).equal_to()
StepCheck(
"В отчёте найдена строка 'Суммарное время работы:'",
"total_work_duration_label",
soft_failures,
).actual(parsed_report.total_label_row_index).is_not_none()
StepCheck(
"Суммарное время работы в отчёте не нулевое",
"total_work_duration",
soft_failures,
).actual(
total_duration_seconds
).is_greater_than(0, LdsReportConst.ZERO_DURATION_TEXT)
for section_row in section_rows:
duration_diff = abs(section_row.modes_sum_seconds - (total_duration_seconds or 0))
StepCheck(
f"Сумма режимов СОУ для '{section_row.section_name}' "
f"совпадает с суммарным временем (+-{duration_tolerance} с)",
"modes_sum_seconds",
soft_failures,
).actual(duration_diff).is_less_than(
duration_tolerance + 1,
f"погрешность {duration_tolerance} с",
)
with allure.step("Подготовка данных шапки отчёта для проверки"):
title_info = parsed_report.title_info
report_title_lower = title_info.raw_title.lower()
column_headers = parsed_report.column_headers
header_period_start = title_info.period_start
header_period_end = title_info.period_end
with allure.step("Проверка двойной шапки отчёта о режиме СОУ"):
with SoftAssertions() as soft_failures:
StepCheck(
f"В шапке отчёта присутствует '{LdsReportConst.LDS_STATUS_REPORT_NAME_PART}'",
"report_title",
soft_failures,
).contains(report_title_lower, lds_report_name_part_lower)
StepCheck(
"Время начала периода в шапке совпадает с фильтром запроса (+-1 мин)",
"period_start",
soft_failures,
).actual(header_period_start).is_between(period_start_lo, period_start_hi)
StepCheck(
"Время конца периода в шапке совпадает с фильтром запроса (+-1 мин)",
"period_end",
soft_failures,
).actual(header_period_end).is_between(period_end_lo, period_end_hi)
StepCheck(
"Названия колонок в шапке отчёта",
"column_headers",
soft_failures,
).actual(
column_headers
).expected(LdsReportConst.EXPECTED_COLUMN_HEADERS).equal_to()
with allure.step("Проверка имени файла отчёта о режиме СОУ"):
with SoftAssertions() as soft_failures:
StepCheck(f"Имя файла оканчивается на {ReportConst.XLSX_EXTENSION}", "file_name", soft_failures).actual(
has_xlsx_extension
).expected(True).equal_to()
StepCheck(
f"Имя файла содержит '{LdsReportConst.LDS_STATUS_REPORT_NAME_PART}'",
"file_name",
soft_failures,
).contains(report_file_name_lower, lds_report_name_part_lower)
StepCheck(
f"Имя файла содержит описание ТУ '{cfg.tu_name}'",
"file_name",
soft_failures,
).contains(report_file_name_lower, report_state.tu_description_lower)
StepCheck(
"Дата начала периода в имени файла совпадает с фильтром запроса (+-1 мин)",
"period_start_in_file_name",
soft_failures,
).actual(file_name_period_start).is_between(period_start_lo, period_start_hi)
StepCheck(
"Дата конца периода в имени файла совпадает с фильтром запроса (+-1 мин)",
"period_end_in_file_name",
soft_failures,
).actual(file_name_period_end).is_between(period_end_lo, period_end_hi)
except Exception:
with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
if report_state.temp_file_path and report_state.report_file_name:
report_utils.attach_report_file_to_allure(report_state.temp_file_path, report_state.report_file_name)
raise
with allure.step("Проверка пуш-нотификации о готовности отчёта"):
notification = report_state.notification
notification_reply_status = notification.replyStatus if notification else None
notification_reply_content = notification.replyContent if notification else None
notification_export_status = notification_reply_content.exportStatus if notification_reply_content else None
notification_error_message = (
(notification_reply_content.errorMessage or "") if notification_reply_content else ""
)
with SoftAssertions() as soft_failures:
StepCheck("Получена пуш-нотификация о готовности отчёта", "notification", soft_failures).actual(
notification
).is_not_none()
StepCheck("Проверка статуса пуш-нотификации", "replyStatus", soft_failures).actual(
notification_reply_status
).expected(ReplyStatus.OK.value).equal_to()
StepCheck("Проверка наличия контента нотификации", "replyContent", soft_failures).actual(
notification_reply_content
).is_not_none()
StepCheck("Проверка exportStatus в нотификации", "exportStatus", soft_failures).actual(
notification_export_status
).expected(ExportStatus.DONE).equal_to()
StepCheck("В нотификации нет текста ошибки", "errorMessage", soft_failures).actual(
notification_error_message
).is_empty()
def stationary_status_in_journal(http_client, cfg: SmokeSuiteConfig, imitator_start_time, test_data: CaseData):
"""
Проверка записей журнала о режиме мт
"""
if not test_data:
pytest.fail("Не заполнены обязательные параметры теста")
exp_mode_part_message, exp_reason_part_message, exp_priority_message = test_data.expected_result
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=PUMPING_STATUS"):
end_time = datetime.now()
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.LIMIT_CONTROLLED_SITES),
filtering=Filtering(messageTypes=int(MessageType.PUMPING_STATUS), objects=FilteringObjects(tuId=cfg.tu_id)),
)
response = http_client.post_request(HttpConst.GET_MESSAGES_URL_PATH, request_body)
payload = t_utils.get_json_from_http_response(response)
parsed_payload = parser.parse_journal_msg(payload)
with allure.step("Извлечение и подготовка данных для проверки"):
messages_info = getattr(parsed_payload.replyContent, 'messagesInfo', [])
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
filter_start_msk = t_utils.localize_as_moscow(imitator_start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
messages_time_filtered = [
msg
for msg in messages_info
if filter_start_msk <= t_utils.ensure_moscow_timezone(msg.time) <= filter_end_msk
]
messages_time_filtered.sort(key=lambda msg: t_utils.ensure_moscow_timezone(msg.time), reverse=True)
# Фильтрация уникальных наименований участков КП-КП
control_points_list = []
for msg in messages_time_filtered:
control_points_list.append(msg.controlPoint)
unique_control_points_list = set(control_points_list)
count_unique_control_points = len(unique_control_points_list)
# Собираю список сообщений с уникальными наименованиями участков КП_КП
already_been = set() # Создание пустого множества для хранения отобранных сообщений
filter_messages_with_unique_control_points = [] # пустой список
for msg in messages_time_filtered:
if msg.controlPoint in unique_control_points_list: # условие1: наименование КП-КП в списке
if msg.controlPoint not in already_been: # условие2: сообщение еще не в already_been
filter_messages_with_unique_control_points.append(
msg
) # действие1: добавляется сообщение в filter_messages_with_unique_control_points
already_been.add(msg.controlPoint) # действие2: добавляется КП-КП во множество, для условия 2
allure.attach(
f" Список: {filter_messages_with_unique_control_points}",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
# Создание и наполнение списков сообщений с фильтром по 'event'
containers = defaultdict(list)
for msg in filter_messages_with_unique_control_points:
event = msg.event
if event in (
TestConst.JOURNAL_MESSAGE_EVENT_STATIONARY,
TestConst.JOURNAL_MESSAGE_EVENT_NOT_STATIONARY,
TestConst.JOURNAL_MESSAGE_EVENT_STOP,
):
containers[event].append(msg)
else:
containers["another"].append(msg)
stationary_status_list = list(containers[TestConst.JOURNAL_MESSAGE_EVENT_STATIONARY])
unstationary_status_list = list(containers[TestConst.JOURNAL_MESSAGE_EVENT_NOT_STATIONARY])
stopped_status_list = list(containers[TestConst.JOURNAL_MESSAGE_EVENT_STOP])
another_event_list = list(containers["another"])
most_long_event_list = max(
[unstationary_status_list, stationary_status_list, stopped_status_list, another_event_list], key=len
)
first_message = next(iter(most_long_event_list)) if most_long_event_list else None
if first_message:
priority_message = MessagePriority(first_message.priority) if first_message.priority else None
mode_part, reason_part = t_utils.parse_event(getattr(first_message, "event", None))
else:
priority_message = None
mode_part, reason_part = None, None
with SoftAssertions() as soft_failures:
StepCheck(
"Проверка результата фильтрации сообщений о режиме МТ",
"Кол-во сообщений о режиме МТ с уникальными наименованиями участков КП-КП",
soft_failures,
).actual(count_unique_control_points).expected(TestConst.COUNT_CONTROLLED_SITES).equal_to()
StepCheck(
"Проверка режима МТ на ЛЧ в наибольшей области связности.",
"Режим МТ",
soft_failures,
).actual(
mode_part
).expected(exp_mode_part_message).equal_to()
if reason_part:
StepCheck(
"Проверка причины режима МТ на ЛЧ в наибольшей области связности.", "Причина режима МТ", soft_failures
).contains(reason_part, exp_reason_part_message)
StepCheck("Проверка значимости сообщения", "Важность", soft_failures).actual(priority_message).expected(
exp_priority_message
).equal_to()
async def export_mt_mode_report(ws_client, cfg: SmokeSuiteConfig, imitator_start_time: datetime):
"""
Сценарий формирования xlsx-отчёта о режиме работы МТ.
Этапы:
1. Подписка SubscribeReportsDataExportedRequest на пуш-нотификации.
2. Отправка ExportReportsCommandRequest (тип StationaryStatusReport, фильтр по периоду).
3. Ожидание ReportDataExportedNotification о готовности отчёта.
4. Лонг-поллинг GetExportedDataListRequest до появления отчёта в списке.
5. DownloadExportedDataRequest и получение fileChunk.
6. Проверка xlsx: участки, длительности режимов МТ, основной режим.
7. Проверка двойной шапки и имени файла.
Скачанный файл удаляется по завершению, прикладывается к Allure только при падении теста.
"""
report_state = ExportMtModeReportState()
with allure.step("Подготовка параметров сценария формирования отчёта о режиме работы МТ"):
report_state.expected_report_test = cfg.export_mt_mode_report_test
StepCheck("В конфигурации задан export_mt_mode_report_test", "export_mt_mode_report_test").actual(
report_state.expected_report_test
).is_not_none()
report_state.expected_period_start = t_utils.localize_as_moscow(imitator_start_time)
report_state.expected_period_end = t_utils.localize_as_moscow(
imitator_start_time + timedelta(minutes=report_state.expected_report_test.offset)
)
report_state.expected_period_start_naive = report_utils.normalize_report_period_naive(
report_state.expected_period_start
)
report_state.expected_period_end_naive = report_utils.normalize_report_period_naive(
report_state.expected_period_end
)
report_state.expected_tu_description_lower = cfg.tu_name.lower()
report_state.expected_section_names = list(MtReportConst.SECTION_NAMES)
report_state.expected_dominant_mode_column = MtReportConst.STATIONARY_STATUS_TO_COLUMN.get(
cfg.expected_report_stationary_status
)
report_state.expected_file_name = report_utils.build_export_report_file_name(
cfg.tu_name,
report_state.expected_period_start,
report_state.expected_period_end,
MtReportConst.MT_MODE_REPORT_NAME_PART,
". ",
)
time_offset_hours = t_utils.report_time_offset_hours()
StepCheck(
f"Смещение timeOffset для запросов отчёта (часовой пояс {TestConst.ZONE_INFO})",
"time_offset_hours",
).actual(time_offset_hours).is_not_none()
report_state.actual_time_offset_hours = time_offset_hours
StepCheck(
"Для набора задан ожидаемый доминирующий режим МТ",
"expected_dominant_mode_column",
).actual(report_state.expected_dominant_mode_column).is_not_none()
allure.attach(
f"period.start={report_state.expected_period_start}\n"
f"period.end={report_state.expected_period_end}\n"
f"offset_minutes={report_state.expected_report_test.offset}\n"
f"sections={report_state.expected_section_names}\n"
f"expected_dominant_mode_column={report_state.expected_dominant_mode_column}",
name="Фильтр периода отчёта о режиме МТ",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(f"Этап 1. Подписка на пуш-нотификации {ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST}"):
await t_utils.connect(ws_client, ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST, [])
with allure.step(f"Этап 2. Запрос формирования отчёта {ReportConst.EXPORT_REPORTS_COMMAND_REQUEST}"):
request_payload = {
"tuId": cfg.tu_id,
"exportedDataTypes": [ExportedDataType.STATIONARY_STATUS_REPORT.value],
"timeOffset": report_state.actual_time_offset_hours,
"period": {
"start": t_utils.datetime_to_msgpack_timestamp(report_state.expected_period_start),
"end": t_utils.datetime_to_msgpack_timestamp(report_state.expected_period_end),
"additionalProperties": {},
},
}
await t_utils.connect(ws_client, ReportConst.EXPORT_REPORTS_COMMAND_REQUEST, request_payload)
with allure.step(
f"Этап 3. Ожидание пуш-нотификации {ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION} о готовности отчёта"
):
report_state.actual_notification = await t_utils.poll_for_report_export_notification(
ws_client=ws_client,
parser=parser,
total_wait_seconds=ReportConst.NOTIFICATION_TIMEOUT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step(f"Этап 4. Лонг-поллинг {ReportConst.GET_EXPORTED_DATA_LIST_REQUEST} до появления отчёта в списке"):
report_state.actual_report_item = await t_utils.poll_for_exported_file(
ws_client=ws_client,
parser=parser,
list_limit=ReportConst.EXPORTED_DATA_LIST_LIMIT,
expected_data_type=ExportedDataType.STATIONARY_STATUS_REPORT,
name_substring=MtReportConst.MT_MODE_REPORT_NAME_PART,
tu_name_substring=cfg.tu_name,
period_start=report_state.expected_period_start,
period_end=report_state.expected_period_end,
total_wait_seconds=ReportConst.LIST_POLL_TOTAL_WAIT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step("Подготовка данных найденного отчёта в списке"):
report_item = report_state.actual_report_item
if report_item is not None:
allure.attach(
f"id={report_item.id}, name={report_item.name}, "
f"exportedDataType={report_item.exportedDataType}, "
f"start={t_utils.format_datetime_moscow(report_item.start)}, "
f"end={t_utils.format_datetime_moscow(report_item.end)}",
name="Найденный отчёт в списке",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Проверка: отчёт найден в списке сформированных файлов"):
StepCheck("Отчёт найден в списке сформированных файлов", "report_item").actual(
report_state.actual_report_item
).is_not_none()
with allure.step(
f"Этап 5. Streaming-вызов {ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST} "
f"по id={report_state.actual_report_item.id}"
):
download_request = {
"exportedDataId": report_state.actual_report_item.id,
"exportedDataType": ExportedDataType.STATIONARY_STATUS_REPORT.to_download_name(),
"additionalProperties": None,
"timeOffset": report_state.actual_time_offset_hours,
}
download_purpose = (
f"скачивание xlsx-отчёта о режиме МТ (exportedDataId={report_state.actual_report_item.id}) "
f"после формирования отчёта и выбора файла в списке GetExportedDataListRequest"
)
await t_utils.connect_stream(
ws_client,
ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
download_request,
purpose=download_purpose,
)
report_state.actual_download_invocation_id = ws_client.invocation_id
with allure.step("Этап 6. Получение fileChunk - скачивание отчёта о режиме МТ"):
report_state.actual_download_reply = await t_utils.receive_download_exported_data_reply(
ws_client=ws_client,
parser=parser,
invocation_id=report_state.actual_download_invocation_id,
request_name=ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
total_wait_seconds=ReportConst.DOWNLOAD_TIMEOUT_SECONDS,
purpose=download_purpose,
)
with allure.step("Извлечение данных ответа на скачивание"):
download_reply = report_state.actual_download_reply
download_reply_status = download_reply.replyStatus
has_download_reply_content = download_reply.replyContent is not None
report_state.actual_file_bytes = download_reply.replyContent.fileChunk if has_download_reply_content else None
is_xlsx_signature = (
report_utils.is_xlsx_file_bytes(report_state.actual_file_bytes) if report_state.actual_file_bytes else False
)
with allure.step("Проверка ответа на скачивание и формата xlsx"):
StepCheck("Проверка статуса ответа на скачивание", "replyStatus").actual(download_reply_status).expected(
ReplyStatus.OK.value
).equal_to()
StepCheck("Проверка наличия контента ответа на скачивание", "replyContent").actual(
has_download_reply_content
).expected(True).equal_to()
StepCheck("Проверка наличия байт файла", "fileChunk").actual(report_state.actual_file_bytes).is_not_empty()
StepCheck("Проверка xlsx (zip) сигнатуры файла", "file_signature").actual(is_xlsx_signature).expected(
True
).equal_to()
with allure.step("Подготовка данных для проверки имени файла отчёта"):
report_file_name = report_state.expected_file_name
report_file_name_lower = report_file_name.lower()
file_name_period_start, file_name_period_end = report_utils.parse_period_from_export_file_name(
report_file_name,
MtReportConst.REPORT_FILE_NAME_PERIOD_PATTERN,
)
period_start_lo, period_start_hi, period_end_lo, period_end_hi = report_utils.report_period_comparison_bounds(
report_state.expected_period_start_naive,
report_state.expected_period_end_naive,
)
has_xlsx_extension = report_utils.is_xlsx_extension(report_file_name)
mt_report_name_part_lower = MtReportConst.MT_MODE_REPORT_NAME_PART.lower()
try:
with allure.step("Этап 7. Сохранение и разбор xlsx-отчёта о режиме МТ"):
report_state.actual_temp_file_path = report_utils.save_report_bytes_to_temp_file(
report_state.actual_file_bytes,
prefix="mt_mode_report_",
)
StepCheck("Временный xlsx файл создан", "temp_file_path").actual(
report_state.actual_temp_file_path
).is_not_none()
report_state.actual_worksheet = report_utils.load_report_worksheet(report_state.actual_temp_file_path)
report_state.actual_parsed_report = mt_report_utils.parse_mt_mode_report_worksheet(
report_state.actual_worksheet,
report_state.expected_section_names,
)
parsed_report = report_state.actual_parsed_report
allure.attach(
f"Шапка (raw): {parsed_report.title_info.raw_title}\n"
f"period_start: {parsed_report.title_info.period_start}\n"
f"period_end: {parsed_report.title_info.period_end}\n"
f"total_duration: {parsed_report.total_duration_raw}",
name="Шапка отчёта о режиме МТ",
attachment_type=allure.attachment_type.TEXT,
)
allure.attach(
lds_report_utils.format_section_rows_for_allure(parsed_report.section_rows),
name="Строки участков отчёта",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных таблицы отчёта для проверки"):
parsed_report = report_state.actual_parsed_report
section_rows = parsed_report.section_rows
total_duration_seconds = parsed_report.total_duration_seconds
duration_tolerance = MtReportConst.TOTAL_DURATION_TOLERANCE_SECONDS
mode_totals = mt_report_utils.sum_duration_columns_across_rows(
section_rows,
MtReportConst.MODE_DURATION_COLUMNS,
)
with allure.step("Проверка содержимого таблицы отчёта о режиме МТ"):
StepCheck("Лист xlsx открыт", "worksheet").actual(report_state.actual_worksheet).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
"Количество строк участков в отчёте",
"section_rows_count",
soft_failures,
).actual(
len(section_rows)
).expected(len(report_state.expected_section_names)).equal_to()
for section_index, expected_section_name in enumerate(report_state.expected_section_names):
actual_section_name = (
section_rows[section_index].section_name if section_index < len(section_rows) else None
)
StepCheck(
f"Наименование участка #{section_index + 1}",
MtReportConst.COL_SECTION,
soft_failures,
).actual(actual_section_name).expected(expected_section_name).equal_to()
for section_row in section_rows:
for column_name in MtReportConst.MODE_DURATION_COLUMNS:
cell_value = section_row.cells.get(column_name)
duration_text = (
section_row.cells.get(column_name) or ""
).strip() or MtReportConst.ZERO_DURATION_TEXT
StepCheck(
f"Участок '{section_row.section_name}': колонка '{column_name}'",
"длительность",
soft_failures,
).actual(mt_report_utils.is_duration_cell_filled(cell_value)).is_true_with_details(
expected_text="указано время в формате H:MM:SS (допускается 0:00:00)",
actual_text=duration_text,
)
total_duration_text = parsed_report.total_duration_raw or lds_report_utils.format_duration_seconds(
total_duration_seconds or 0
)
total_label_row = parsed_report.total_label_row_index
StepCheck(
"В отчёте присутствует строка 'Суммарное время работы:'",
"структура отчёта",
soft_failures,
).actual(total_label_row is not None).is_true_with_details(
expected_text=f"найдена строка с текстом '{MtReportConst.TOTAL_WORK_DURATION_LABEL}'",
actual_text=(
f"строка {total_label_row} содержит '{MtReportConst.TOTAL_WORK_DURATION_LABEL}'"
if total_label_row is not None
else "строка не найдена"
),
)
StepCheck(
"Суммарное время работы в отчёте больше нуля",
"суммарное время",
soft_failures,
).actual((total_duration_seconds or 0) > 0).is_true_with_details(
expected_text="суммарное время больше 0:00:00",
actual_text=total_duration_text,
)
for section_row in section_rows:
duration_diff = abs(section_row.modes_sum_seconds - (total_duration_seconds or 0))
section_sum_text = lds_report_utils.format_duration_seconds(section_row.modes_sum_seconds)
diff_text = lds_report_utils.format_duration_seconds(duration_diff)
sums_match = duration_diff < duration_tolerance + 1
StepCheck(
f"Участок '{section_row.section_name}': сумма режимов МТ совпадает с общим временем "
f"(±{duration_tolerance} с)",
"согласованность длительностей",
soft_failures,
).actual(sums_match).is_true_with_details(
expected_text=(
f"сумма режимов ({section_sum_text}) равна суммарному времени ({total_duration_text}), "
f"погрешность не более {duration_tolerance} с"
),
actual_text=(
f"сумма режимов: {section_sum_text}, суммарное время: {total_duration_text}, "
f"разница: {diff_text} ({duration_diff} с)"
),
)
dominant_column = report_state.expected_dominant_mode_column
dominant_total = mode_totals.get(dominant_column, 0)
max_column = max(mode_totals, key=mode_totals.get)
max_total = mode_totals[max_column]
all_modes_text = ", ".join(
f"'{column_name}': {lds_report_utils.format_duration_seconds(total_seconds)}"
for column_name, total_seconds in mode_totals.items()
)
StepCheck(
f"Режим '{dominant_column}' имеет максимальное суммарное время по всем участкам",
"доминирующий режим МТ",
soft_failures,
).actual(
mt_report_utils.is_expected_dominant_mode_column(mode_totals, dominant_column)
).is_true_with_details(
expected_text=(
f"'{dominant_column}': {lds_report_utils.format_duration_seconds(dominant_total)} - "
f"наибольшее время; по всем участкам: {all_modes_text}"
),
actual_text=(
f"наибольшее время у режима '{max_column}': "
f"{lds_report_utils.format_duration_seconds(max_total)}"
),
)
allure.attach(
"\n".join(
f"{column_name}: {lds_report_utils.format_duration_seconds(total_seconds)}"
for column_name, total_seconds in mode_totals.items()
),
name="Суммарные длительности режимов МТ по всем участкам",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных шапки отчёта для проверки"):
title_info = parsed_report.title_info
report_title_lower = title_info.raw_title.lower()
column_headers = parsed_report.column_headers
header_period_start = title_info.period_start
header_period_end = title_info.period_end
with allure.step("Проверка двойной шапки отчёта о режиме МТ"):
with SoftAssertions() as soft_failures:
StepCheck(
f"В шапке отчёта присутствует '{MtReportConst.MT_MODE_REPORT_NAME_PART}'",
"report_title",
soft_failures,
).contains(report_title_lower, mt_report_name_part_lower)
StepCheck(
"Время начала периода в шапке совпадает с фильтром запроса (+-1 мин)",
"period_start",
soft_failures,
).actual(header_period_start).is_between(period_start_lo, period_start_hi)
StepCheck(
"Время конца периода в шапке совпадает с фильтром запроса (+-1 мин)",
"period_end",
soft_failures,
).actual(header_period_end).is_between(period_end_lo, period_end_hi)
StepCheck(
"Названия колонок в шапке отчёта",
"column_headers",
soft_failures,
).actual(
column_headers
).expected(MtReportConst.EXPECTED_COLUMN_HEADERS).equal_to()
with allure.step("Проверка имени файла отчёта о режиме МТ"):
with SoftAssertions() as soft_failures:
StepCheck(f"Имя файла оканчивается на {ReportConst.XLSX_EXTENSION}", "file_name", soft_failures).actual(
has_xlsx_extension
).expected(True).equal_to()
StepCheck(
f"Имя файла содержит '{MtReportConst.MT_MODE_REPORT_NAME_PART}'",
"file_name",
soft_failures,
).contains(report_file_name_lower, mt_report_name_part_lower)
StepCheck(
f"Имя файла содержит описание ТУ '{cfg.tu_name}'",
"file_name",
soft_failures,
).contains(report_file_name_lower, report_state.expected_tu_description_lower)
StepCheck(
"Дата начала периода в имени файла совпадает с фильтром запроса (+-1 мин)",
"period_start_in_file_name",
soft_failures,
).actual(file_name_period_start).is_between(period_start_lo, period_start_hi)
StepCheck(
"Дата конца периода в имени файла совпадает с фильтром запроса (+-1 мин)",
"period_end_in_file_name",
soft_failures,
).actual(file_name_period_end).is_between(period_end_lo, period_end_hi)
except Exception:
with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
if report_state.actual_temp_file_path and report_state.expected_file_name:
report_utils.attach_report_file_to_allure(
report_state.actual_temp_file_path,
report_state.expected_file_name,
)
raise
with allure.step("Проверка пуш-нотификации о готовности отчёта"):
notification = report_state.actual_notification
notification_reply_status = notification.replyStatus if notification else None
notification_reply_content = notification.replyContent if notification else None
notification_export_status = notification_reply_content.exportStatus if notification_reply_content else None
notification_error_message = (
(notification_reply_content.errorMessage or "") if notification_reply_content else ""
)
with SoftAssertions() as soft_failures:
StepCheck("Получена пуш-нотификация о готовности отчёта", "notification", soft_failures).actual(
notification
).is_not_none()
StepCheck("Проверка статуса пуш-нотификации", "replyStatus", soft_failures).actual(
notification_reply_status
).expected(ReplyStatus.OK.value).equal_to()
StepCheck("Проверка наличия контента нотификации", "replyContent", soft_failures).actual(
notification_reply_content
).is_not_none()
StepCheck("Проверка exportStatus в нотификации", "exportStatus", soft_failures).actual(
notification_export_status
).expected(ExportStatus.DONE).equal_to()
StepCheck("В нотификации нет текста ошибки", "errorMessage", soft_failures).actual(
notification_error_message
).is_empty()
тест лдс статус регрес
"""
Архитектура параметризации:
- SUITE_PARAMS: тесты уровня набора (один раз на набор данных LDSStatusConfig)
Для добавления нового набора:
1. Создать файл в test_config/datasets/ с LDSStatusConfig
2. Добавить маппинг новых тестов в conftest.py -> LDS_STATUS_SUITE_LEVEL_MAPPING
3. Тесты подхватятся автоматически
Запуск:
- Все тесты: pytest tests/test_lds_status_regress.py
- Один набор: pytest tests/test_lds_status_regress.py --suites=Lds_status_regress_in_flow
- Несколько наборов: pytest tests/test_lds_status_regress.py --suites=Lds_status_regress_in_flow,Lds_status_regress_2
"""
import datetime
from typing import Any, List, Optional
import allure
import pytest
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from test_config.datasets import ALL_LDS_STATUS_CONFIGS
from test_config.models_for_tests import CaseMarkers, LDSStatusConfig
from test_scenarios import lds_status_scenarios as scenarios
from test_scenarios import smoke_scenarios
# ===== ГЕНЕРАЦИЯ ПАРАМЕТРОВ =====
def _get_suite_markers(config: LDSStatusConfig) -> List[pytest.MarkDecorator]:
"""Маркеры для тестового набора."""
return [
pytest.mark.test_suite_name(config.suite_name),
pytest.mark.test_suite_data_id(config.suite_data_id),
pytest.mark.test_data_name(config.archive_name),
pytest.mark.tu_id(config.technological_unit.id),
]
def _generate_suite_params() -> List[Any]:
"""
Генерирует параметры для тестов уровня набора данных.
Один параметр на каждый config.
"""
return [
pytest.param(config, id=config.suite_name, marks=_get_suite_markers(config))
for config in ALL_LDS_STATUS_CONFIGS
]
# ===== ПАРАМЕТРЫ ДЛЯ ТЕСТОВ =====
SUITE_PARAMS: List[Any] = _generate_suite_params()
# ===== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =====
def _apply_allure_markers(test_config: CaseMarkers, tag: str, title: str, description: Optional[str] = None) -> None:
"""Применяет allure-маркеры из конфига теста."""
if not test_config:
pytest.skip("Не заполнена конфигурация теста: тест пропущен")
allure.dynamic.tag(tag)
allure.dynamic.tag("REGRESS")
allure.dynamic.title(title)
if description:
allure.dynamic.description(description)
# ===== ТЕСТЫ УРОВНЯ НАБОРА =====
# Запускаются один раз для каждого config
@pytest.mark.parametrize("config", SUITE_PARAMS)
class TestSuiteScenarios:
"""
Тесты уровня набора данных.
Запускаются один раз для каждого конфига.
"""
@pytest.mark.critical_stop
def test_lds_status_basic_info(self, http_client: StandHttpClient, config: LDSStatusConfig) -> None:
"""[BasicInfo] Проверка базовой информации СОУ: список ТУ"""
tag = "BasicInfo"
title = f"[{tag}] Проверка списка ТУ. ЭФ: Главная страница"
_apply_allure_markers(config.lds_status_basic_info_test, tag, title)
smoke_scenarios.basic_info(http_client, config)
@pytest.mark.asyncio
async def test_lds_status_init_accumulation_data(self, ws_client: WebSocketClient, config: LDSStatusConfig) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация,
Причина: Накопление данных
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Инициализация', по причине: 'Накопление данных'. ЭФ: Схема"
_apply_allure_markers(
config.init_accumulation_data_test,
tag,
title,
(
f"Проверка режима работы СОУ на базовых ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_accumulation_data_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Накопление данных"
),
)
test_data = config.init_accumulation_data_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_init_cold_start(self, ws_client: WebSocketClient, config: LDSStatusConfig) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация,
Причина: Холодный пуск
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Инициализация', по причине: 'Холодный пуск'. ЭФ: Схема"
_apply_allure_markers(
config.init_cold_start_test,
tag,
title,
(
f"Проверка режима работы СОУ на базовых ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_cold_start_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n Ожидаемая причина режима работы СОУ: Холодный пуск"
),
)
test_data = config.init_cold_start_test_data
await scenarios.lds_status_check_on_longest_flow_area(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_init_switching_shut_off(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация,
Причина: Переключение запорной арматуры
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Инициализация', "
"по причине: 'Переключение запорной арматуры'. ЭФ: Схема"
)
_apply_allure_markers(
config.init_switching_shut_off_test,
tag,
title,
(
f"Проверка режима работы СОУ на базовых ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_switching_shut_off_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Переключение запорной арматуры"
),
)
test_data = config.init_switching_shut_off_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_cold_start(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после Инициализации, по причине Холодный пуск.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после Иницализации. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_cold_start_test,
tag,
title,
(
"Проверка режима работы СОУ после Инициализации, по причине: Холодный пуск, на показательных ДУ"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.serviceable_after_cold_start_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_additive_injectors_operation(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_additive_injectors_operation_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.technological_unit.description}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_additive_injectors_operation_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_absence_min_pressure_sensors(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_absence_min_pressure_sensors_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_absence_min_pressure_sensors_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_exceeding_distance_between_flow_meters(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_exceeding_distance_between_flow_meters_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.technological_unit.description}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_exceeding_distance_between_flow_meters_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_starting_pumping_out_pumps(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_starting_pumping_out_pumps_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_starting_pumping_out_pumps_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_faulty_pressure_sensors_at_pump(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_faulty_pressure_sensors_at_pump_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_faulty_pressure_sensors_at_pump_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_faulty(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после Инициализации, по причине выхода из Неисправности.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна', после Неисправности. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_faulty_test,
tag,
title,
(
"Проверка режима работы СОУ, после Неисправности, на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.serviceable_after_faulty_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_switching_shut_off(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после Инициализации, по причине Переключение запорной арматуры.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна', после Инициализации. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_switching_shut_off_test,
tag,
title,
(
"Проверка режима работы СОУ, после Инициализации по причине Переключение запорной арматуры"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки: "
f"{config.serviceable_after_switching_shut_off_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_additive_injectors_operation(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Наличие ПТП
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', по причине: 'Наличие ПТП'. ЭФ: Схема"
_apply_allure_markers(
config.deg_additive_injectors_operation_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_additive_injectors_operation_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Наличие ПТП"
),
)
test_data = config.deg_additive_injectors_operation_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_exceeding_distance_between_pressure_sensors(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: расстояние между СИ давления более 50 км
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Расстояние между СИ давления более 50 км'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_exceeding_distance_between_pressure_sensors_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_exceeding_distance_between_pressure_sensors_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Расстояние между СИ давления более 50 км"
),
)
test_data = config.deg_exceeding_distance_between_pressure_sensors_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_absence_min_pressure_sensors(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: менее 4 исправных СИ давления
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'менее 4 исправных СИ давления'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_absence_min_pressure_sensors_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_absence_min_pressure_sensors_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Ухудшение При наличии менее четырех (3, 6) исправных СИ давления "
"на разных КП ЛЧ МТ/НПС на диагностическом участке (кроме случая нахождения трубопровода "
"в режиме остановленной перекачки)"
),
)
test_data = config.deg_absence_min_pressure_sensors_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_faulty_pressure_sensors_at_pump_station(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Отказ СИ давления на входе/выходе НПС
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ давления на входе/выходе НПС'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_faulty_pressure_sensors_at_pump_station_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_faulty_pressure_sensors_at_pump_station_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ давления на входе/выходе НПС"
),
)
test_data = config.deg_faulty_pressure_sensors_at_pump_station_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_pig_sensor_passage(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: прохождение СОД
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', по причине: 'прохождение СОД'"
_apply_allure_markers(
config.deg_pig_sensor_passage_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.deg_pig_sensor_passage_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: прохождение СОД "
),
)
test_data = config.deg_pig_sensor_passage_test_data
await scenarios.lds_status_check_degradation_pig_sensor_passage(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_gravity_section_pumping(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Наличие самотечного участка/участка с неполным сечением
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Наличие самотечного участка/участка с неполным сечением'"
)
_apply_allure_markers(
config.deg_gravity_section_pumping_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.deg_gravity_section_pumping_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Наличие самотечного участка/участка с неполным сечением "
),
)
test_data = config.deg_gravity_section_pumping_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_gravity_section_pumping_in_stopping(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Наличие самотечного участка в режиме остановленной перекачки
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Наличие самотечного участка в режиме остановленной перекачки'"
)
_apply_allure_markers(
config.deg_gravity_section_pumping_in_stopping_test,
tag,
title,
(
f"Проверка режима работы СОУ на всей ЛЧ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_gravity_section_pumping_in_stopping_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Наличие самотечного участка в режиме остановленной перекачки "
),
)
test_data = config.deg_gravity_section_pumping_in_stopping_test_data
await scenarios.lds_status_check_on_longest_flow_area(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_starting_pumping_out_pumps(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ и МТ
Причина: Работа насосов откачки
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режимов работы СОУ и МТ, по причине: 'Работа насосов откачки'. ЭФ: Схема"
_apply_allure_markers(
config.deg_starting_pumping_out_pumps_test,
tag,
title,
(
f"Проверка режимов работы СОУ и МТ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_starting_pumping_out_pumps_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемый режим работы МТ: Нестационарный\n "
"Ожидаемая причина режима работы СОУ: Работа насосов откачки"
"Ожидаемая причина режима работы МТ: Работа насосов откачки"
),
)
test_data = config.deg_starting_pumping_out_pumps_test_data
await scenarios.lds_and_stationary_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_exceeding_distance_between_flow_meters(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Расстояние между СИ расхода на пути перекачки более 200 км
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Расстояние между СИ расхода на пути перекачки более 200 км'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_exceeding_distance_between_flow_meters_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_exceeding_distance_between_flow_meters_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Расстояние между СИ расхода на пути перекачки более 200 км"
),
)
test_data = config.deg_exceeding_distance_between_flow_meters_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_init_exiting_faulty(self, ws_client: WebSocketClient, config: LDSStatusConfig) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация,
Причина: Выход СОУ из режима «Неисправна»
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Инициализация', "
"по причине: 'Выход СОУ из режима «Неисправна»'. ЭФ: Схема"
)
_apply_allure_markers(
config.init_exiting_faulty_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_exiting_faulty_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Выход СОУ из режима «Неисправна»"
),
)
test_data = config.init_exiting_faulty_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_temperature_sensor_on_du_2(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Отказ СИ температуры
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 2: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ температуры'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_temperature_sensor_on_du_2_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 2, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_temperature_sensor_on_du_2_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ температуры"
),
)
test_data = config.deg_rejection_temperature_sensor_on_du_2_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_temperature_sensor_on_du_3(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Отказ СИ температуры
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 3: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ температуры'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_temperature_sensor_on_du_3_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 3, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_temperature_sensor_on_du_3_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ температуры"
),
)
test_data = config.deg_rejection_temperature_sensor_on_du_3_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_temperature_sensor_on_du_5(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Отказ СИ температуры
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 5: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ температуры'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_temperature_sensor_on_du_5_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 5, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_temperature_sensor_on_du_5_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ температуры"
),
)
test_data = config.deg_rejection_temperature_sensor_on_du_5_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_density_and_viscosity_on_du_2(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причины: Отказ СИ плотности и СИ вязкости
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 2: 'Ухудшение характеристик', "
"по причинам: 'Отказ СИ плотности' и 'Отказ СИ вязкости'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_density_and_viscosity_on_du_2_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 2, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_density_and_viscosity_on_du_2_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ плотности"
"Ожидаемая причина режима работы СОУ: Отказ СИ вязкости"
),
)
test_data = config.deg_rejection_density_and_viscosity_on_du_2_test_data
await scenarios.lds_status_check_with_2_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_density_and_viscosity_on_du_3(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причины: Отказ СИ плотности и СИ вязкости
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 3: 'Ухудшение характеристик', "
"по причинам: 'Отказ СИ плотности' и 'Отказ СИ вязкости'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_density_and_viscosity_on_du_3_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 3, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_density_and_viscosity_on_du_3_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ плотности"
"Ожидаемая причина режима работы СОУ: Отказ СИ вязкости"
),
)
test_data = config.deg_rejection_density_and_viscosity_on_du_3_test_data
await scenarios.lds_status_check_with_2_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_density_and_viscosity_on_du_5(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причины: Отказ СИ плотности и СИ вязкости
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 5: 'Ухудшение характеристик', "
"по причинам: 'Отказ СИ плотности' и 'Отказ СИ вязкости'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_density_and_viscosity_on_du_5_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 5, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_density_and_viscosity_on_du_5_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ плотности"
"Ожидаемая причина режима работы СОУ: Отказ СИ вязкости"
),
)
test_data = config.deg_rejection_density_and_viscosity_on_du_5_test_data
await scenarios.lds_status_check_with_2_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_faulty_absence_min_flow_meters(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Неисправна
Причина: Отсутствие минимального количества СИ Расхода
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Неисправна', "
"по причине: 'Отсутствие минимального количества СИ Расхода'. ЭФ: Схема"
)
_apply_allure_markers(
config.faulty_absence_min_flow_meters_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_flow_meters_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Неисправна\n "
"Ожидаемая причина режима работы СОУ: При одновременном выполнении следующих условий:"
"- отсутствие достоверных показаний граничного на диагностическом участке СИ расхода (кроме "
"отсеченных от рассматриваемого участка СИ расхода для трубопровода в режиме остановленной перекачки);"
"- отсутствие смежного с данным СИ расхода диагностического участка "
"с достоверными показаниями СИ расхода"
),
)
test_data = config.faulty_absence_min_flow_meters_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_faulty_absence_min_flow_meters_continuous(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Неисправна
Причина: Отсутствие минимального количества СИ Расхода
После устранения причины неисправности
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Неисправна', "
"по причине: 'Отсутствие минимального количества СИ Расхода'. "
"После устранения причины неисправности. ЭФ: Схема"
)
_apply_allure_markers(
config.faulty_absence_min_flow_meters_continuous_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_flow_meters_continuous_test.offset} мин.\n"
"После устранения причины неисправности (Заменяет Инициализацию при непродолжительной неисправности)\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Неисправна\n "
"Ожидаемая причина режима работы СОУ: При одновременном выполнении следующих условий:"
"- отсутствие достоверных показаний граничного на диагностическом участке СИ расхода (кроме "
"отсеченных от рассматриваемого участка СИ расхода для трубопровода в режиме остановленной перекачки);"
"- отсутствие смежного с данным СИ расхода диагностического участка "
"с достоверными показаниями СИ расхода"
),
)
test_data = config.faulty_absence_min_flow_meters_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_faulty_absence_min_pressure_sensors(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Неисправна
Причина: Менее 4 КП с достоверными СИ давления
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Неисправна', "
"по причине: 'Менее 4 КП с достоверными СИ давления'. ЭФ: Схема"
)
_apply_allure_markers(
config.faulty_absence_min_pressure_sensors_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_pressure_sensors_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Неисправна\n "
"Ожидаемая причина режима работы СОУ: Менее 4 КП с достоверными СИ давления"
),
)
test_data = config.faulty_absence_min_pressure_sensors_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
def test_lds_status_init_accumulation_data_in_journal(
self, http_client: StandHttpClient, config: LDSStatusConfig
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Инициализация,
Причина: Накопление данных
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Инициализация', по причине: 'Накопление данных'. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.init_accumulation_data_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_accumulation_data_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Инициализация\n Ожидаемая причина режима работы СОУ: Накопление данных"
),
)
test_data = config.init_accumulation_data_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
def test_lds_status_init_cold_start_in_journal(
self, http_client: StandHttpClient, config: LDSStatusConfig, imitator_start_time: datetime
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Инициализация,
Причина: Холодный пуск
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Инициализация', по причине: 'Холодный пуск'. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.init_cold_start_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_cold_start_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Инициализация\n Ожидаемая причина режима работы СОУ: Холодный пуск"
),
)
smoke_scenarios.lds_status_init_in_journal(http_client, config, imitator_start_time)
def test_lds_status_init_switching_shut_off_in_journal(
self, http_client: StandHttpClient, config: LDSStatusConfig
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Инициализация,
Причина: Переключение запорной арматуры
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Инициализация', по причине: 'Переключение запорной арматуры'. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.init_switching_shut_off_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.init_switching_shut_off_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Переключение запорной арматуры"
),
)
test_data = config.init_switching_shut_off_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_serviceable_after_cold_start_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Исправна.
Установка режима Исправна после Инициализации, по причине Холодный пуск.
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Исправна', после Инициализации. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.serviceable_after_cold_start_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.serviceable_after_cold_start_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Исправна\n"
),
)
test_data = config.serviceable_all_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_serviceable_after_switching_shut_off_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Исправна.
Установка режима Исправна после Инициализации, по причине Переключение запорной арматуры.
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Исправна', после Инициализации. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.serviceable_after_switching_shut_off_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.serviceable_after_switching_shut_off_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Исправна\n"
),
)
test_data = config.serviceable_all_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Исправна', после режима Ухудшение характеристик. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Исправна\n"
),
)
test_data = config.serviceable_all_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_degradation_gravity_section_pumping_in_stopping_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение характеристик
Причина: Наличие самотечного участка в режиме остановленной перекачки
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Наличие самотечного участка в режиме остановленной перекачки'"
)
_apply_allure_markers(
config.deg_gravity_section_pumping_in_stopping_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_gravity_section_pumping_in_stopping_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Наличие самотечного участка в режиме остановленной перекачки "
),
)
test_data = config.deg_gravity_section_pumping_in_stopping_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_degradation_exceeding_distance_between_pressure_sensors_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение характеристик
Причина: расстояние между СИ давления более 50 км
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Расстояние между СИ давления более 50 км'"
)
_apply_allure_markers(
config.deg_exceeding_distance_between_pressure_sensors_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_exceeding_distance_between_pressure_sensors_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Расстояние между СИ давления более 50 км "
),
)
test_data = config.deg_exceeding_distance_between_pressure_sensors_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_degradation_faulty_pressure_sensors_at_pump_station_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение характеристик
Причина: Отказ СИ давления на входе/выходе НПС
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ давления на входе/выходе НПС'"
)
_apply_allure_markers(
config.deg_faulty_pressure_sensors_at_pump_station_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_faulty_pressure_sensors_at_pump_station_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ давления на входе/выходе НПС "
),
)
test_data = config.deg_faulty_pressure_sensors_at_pump_station_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_faulty_absence_min_pressure_sensors_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Неисправна
Причина: Менее 4 КП с достоверными СИ давления
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: 'Неисправна', "
"по причине: 'Менее 4 КП с достоверными СИ давления'"
)
_apply_allure_markers(
config.faulty_absence_min_pressure_sensors_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_pressure_sensors_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Неисправна\n "
"Ожидаемая причина режима работы СОУ: Менее 4 КП с достоверными СИ давления "
),
)
test_data = config.faulty_absence_min_pressure_sensors_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
тест смоук
"""
Единый тестовый модуль для всех наборов данных (single-leak и multi-leak).
Архитектура параметризации:
- SUITE_PARAMS: тесты уровня набора (один раз на набор данных)
- LEAK_PARAMS: тесты уровня утечки (один раз на каждую утечку)
Для добавления нового набора:
1. Создать файл в test_config/datasets/
2. Тесты подхватятся автоматически
Запуск:
- Все тесты: pytest tests/test_smoke.py
- Один набор: pytest tests/test_smoke.py --suites=select_4
- Несколько наборов: pytest tests/test_smoke.py --suites=imitative_17,select_19_20
"""
import datetime
from typing import Any, List, Optional
import allure
import pytest
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from constants.test_constants import BaseTN3Constants as Base_const
from test_config.datasets import ALL_SMOKE_CONFIGS
from test_config.models_for_tests import CaseMarkers, LeakTestConfig, SmokeSuiteConfig
from test_scenarios import lds_status_scenarios
from test_scenarios import smoke_scenarios as scenarios
# ===== ГЕНЕРАЦИЯ ПАРАМЕТРОВ =====
def _get_suite_markers(config: SmokeSuiteConfig) -> List[pytest.MarkDecorator]:
"""Маркеры для тестового набора."""
return [
pytest.mark.test_suite_name(config.suite_name),
pytest.mark.test_suite_data_id(config.suite_data_id),
pytest.mark.test_data_name(config.archive_name),
pytest.mark.tu_id(config.technological_unit.id),
]
def _generate_suite_params() -> List[Any]:
"""
Генерирует параметры для тестов уровня набора данных.
Один параметр на каждый config.
"""
return [
pytest.param(config, id=config.suite_name, marks=_get_suite_markers(config)) for config in ALL_SMOKE_CONFIGS
]
def _generate_leak_params() -> List[Any]:
"""
Генерирует параметры для тестов уровня утечки.
Для single-leak конфигов: один параметр (config, leak, 1)
Для multi-leak конфигов: N параметров (config, leak_n, n) для каждой утечки
"""
params: List[Any] = []
for config in ALL_SMOKE_CONFIGS:
# Собираем все утечки из конфига
if config.has_multiple_leaks:
leaks = config.leaks
elif config.leak:
leaks = [config.leak]
else:
continue # Нет утечек в конфиге
# Создаём параметр для каждой утечки
for index, leak in enumerate(leaks):
leak_number = index + 1
# ID для Allure/pytest: select_4 или select_19_20_leak_1
if len(leaks) > 1:
param_id = f"{config.suite_name}_leak_{leak_number}"
else:
param_id = config.suite_name
params.append(
pytest.param(
config,
leak,
leak_number,
id=param_id,
marks=_get_suite_markers(config),
)
)
return params
# ===== ПАРАМЕТРЫ ДЛЯ ТЕСТОВ =====
SUITE_PARAMS: List[Any] = _generate_suite_params()
LEAK_PARAMS: List[Any] = _generate_leak_params()
# ===== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =====
def _apply_allure_markers(test_config: CaseMarkers, tag: str, title: str, description: Optional[str] = None) -> None:
"""Применяет allure-маркеры из конфига теста."""
if not test_config:
pytest.skip("Не заполнена конфигурация теста: тест пропущен")
allure.dynamic.tag(tag)
allure.dynamic.tag("SMOKE")
allure.dynamic.title(title)
if description:
allure.dynamic.description(description)
# ===== ТЕСТЫ УРОВНЯ НАБОРА =====
# Запускаются один раз для каждого config
@pytest.mark.parametrize("config", SUITE_PARAMS)
class TestSuiteScenarios:
"""
Тесты уровня набора данных.
Запускаются один раз для каждого конфига.
"""
@pytest.mark.critical_stop
def test_basic_info(self, http_client: StandHttpClient, config: SmokeSuiteConfig) -> None:
"""[BasicInfo] Проверка базовой информации СОУ: список ТУ"""
tag = "BasicInfo"
title = f"[{tag}] Проверка списка ТУ. ЭФ: Главная страница"
_apply_allure_markers(config.basic_info_test, tag, title)
scenarios.basic_info(http_client, config)
def test_journal_info(self, http_client: StandHttpClient, config: SmokeSuiteConfig) -> None:
"""[MessagesInfo] Проверка наличия сообщений в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка наличия сообщений в журнале. ЭФ: Журнал.Реальное время"
_apply_allure_markers(
config.journal_info_test,
tag,
title,
"Проверка наличия сообщений в журнале.\n" "Синхронный запрос типа: MessagesInfo",
)
scenarios.messages_exist_in_journal(http_client)
@pytest.mark.asyncio
async def test_lds_status_initialization(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[CommonScheme] Проверка режима работы СОУ: Инициализация"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Инициализация'. ЭФ: Схема"
_apply_allure_markers(
config.lds_status_initialization_test,
tag,
title,
(
f"Проверка режима работы СОУ на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.lds_status_initialization_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация"
),
)
await scenarios.lds_status_initialization(ws_client, config)
@pytest.mark.asyncio
async def test_diagnostics_of_signals_after_initialization(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig
) -> None:
"""[OutputSignalsInfo] Проверка выходных сигналов после выхода из инициализации"""
tag = "OutputSignalsInfo"
title = f"[{tag}] Проверка выходных сигналов после выхода из инициализации. ЭФ Диагностика сигналов"
_apply_allure_markers(
config.diagnostics_of_signals_after_initialization_test,
tag,
title,
(
f"Проверка состояния СОУ и перекачки на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.diagnostics_of_signals_after_initialization_test.offset} мин.\n"
),
)
test_data = config.controlled_sites_with_segment
await scenarios.diagnostics_of_signals_after_initialization(ws_client, config, test_data)
@pytest.mark.asyncio
def test_lds_status_init_in_journal(
self, http_client: StandHttpClient, config: SmokeSuiteConfig, imitator_start_time: datetime
) -> None:
"""[MessagesInfo] Проверка записи в журнале о входе СОУ в Инициализацию"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка записи в журнале: СОУ в Инициализации. ЭФ: Журнал. Реальное время"
description = (
f"Проверка записи в журнале о входе СОУ в Инициализацию на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.lds_status_init_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
)
_apply_allure_markers(config.lds_status_init_in_journal_test, tag, title, description)
scenarios.lds_status_init_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_main_page_info(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[MainPageInfo] Проверка установки режима МТ"""
tag = "MainPageInfo"
title = f"[{tag}] Проверка установки режима работы МТ: стационарный. ЭФ: Главная страница.Контент таблица по ТУ"
_apply_allure_markers(
config.main_page_info_test,
tag,
title,
(
f"Проверка установки режима работы МТ: стационарный на данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.main_page_info_test.offset} мин.\n"
"Подписка на сообщения типа: MainPageInfo\n"
"Ожидаемый режим работы МТ: Стационарный"
),
)
await scenarios.main_page_info(ws_client, config)
@pytest.mark.asyncio
async def test_main_page_info_signals(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[MainPageSignalsInfo] Проверка счетчиков состояния сигналов"""
tag = "MainPageSignalsInfo"
title = f"[{tag}] Проверка счетчиков состояния сигналов. ЭФ: Главная страница.Контент таблица по ТУ"
_apply_allure_markers(
config.main_page_info_signals_test,
tag,
title,
(
f"Проверка счетчиков состояния сигналов на данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.main_page_info_signals_test.offset} мин.\n"
"Подписка на сообщения типа: MainPageSignalsInfo\n"
),
)
await scenarios.main_page_info_signals(ws_client, config)
@pytest.mark.asyncio
async def test_imitate_flowmeter_signal(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
"""[ImitateSignal] Проверка имитации расходомера"""
tag = "ImitateSignal"
title = f"[{tag}] Проверка имитации сигнала расходомера. ЭФ: Схема, Входные сигналы, Журнал"
_apply_allure_markers(
config.imitate_flowmeter_signal_test,
tag,
title,
(
"Проверка работы имитации и снятия имитации c расходомера "
f"на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.imitate_flowmeter_signal_test.offset} мин.\n"
"Синхронные запросы типа: ImitateSignalRequest, UnImitateSignalRequest\n"
"Подписка на сообщения типа: InputSignalsContent\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос имитации.\n"
"Значения в полях isImitated, quality и imitation.value расходомера "
"во входных сигналах после имитации.\n"
"Статус-код ответа на синхронный запрос снятия имитации.\n"
"Значение в поле isImitated расходомера во входных сигналах после снятия имитации.\n"
"Примечание: что бы не повлиять на проверки утечек, тест на имитацию выполняется во время "
"инициализации"
),
)
test_data = config.imitate_flowmeter_signal_test_data
await scenarios.imitate_sensor_signal(ws_client, http_client, config, test_data, imitator_start_time)
@pytest.mark.asyncio
async def test_imitate_pressure_sensor_signal(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
"""[ImitateSignal] Проверка имитации датчика давления"""
tag = "ImitateSignal"
title = f"[{tag}] Проверка имитации сигнала датчика давления. ЭФ: Схема, Входные сигналы, Журнал"
_apply_allure_markers(
config.imitate_pressure_sensor_signal_test,
tag,
title,
(
"Проверка работы имитации и снятия имитации c датчика давления "
f"на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.imitate_pressure_sensor_signal_test.offset} мин.\n"
"Синхронные запросы типа: ImitateSignalRequest, UnImitateSignalRequest\n"
"Подписка на сообщения типа: InputSignalsContent\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос имитации.\n"
"Значения в полях isImitated, quality и imitation.value датчика давления "
"во входных сигналах после имитации.\n"
"Статус-код ответа на синхронный запрос снятия имитации.\n"
"Значение в поле isImitated датчика давления во входных сигналах после снятия имитации.\n"
"Примечание: что бы не повлиять на проверки утечек, тест на имитацию выполняется во время "
"инициализации"
),
)
test_data = config.imitate_pressure_sensor_signal_test_data
await scenarios.imitate_sensor_signal(ws_client, http_client, config, test_data, imitator_start_time)
@pytest.mark.asyncio
async def test_mask_signal(
self, ws_client: WebSocketClient, http_client: StandHttpClient, config: SmokeSuiteConfig
) -> None:
"""[MaskSignal] Проверка маскирования датчиков"""
tag = "MaskSignal"
title = f"[{tag}] проверка маскирования датчиков. ЭФ: Схема"
_apply_allure_markers(
config.mask_signal_test,
tag,
title,
(
f"Проверка работы маскирования и снятия маскирования на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.mask_signal_test.offset} мин.\n"
"Синхронные запросы типа: MaskSignalRequest, UnmaskSignalRequest\n"
"Подписка на сообщения типа: InputSignalsContent\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос маскирования.\n"
"Значение в поле isMasked сигнала во входных сигналах после маскирования.\n"
"Статус-код ответа на синхронный запрос снятия маскирования.\n"
"Значение в поле isMasked сигнала во входных сигналах после снятия маскирования.\n"
"Примечание: что бы не повлиять на проверки утечек, тест на маскирование выполняется во время "
"инициализации"
),
)
test_data = config.mask_signal_test_data
await scenarios.mask_signal_test(ws_client, http_client, config, test_data)
def test_mask_info_in_journal(
self, http_client: StandHttpClient, config: SmokeSuiteConfig, imitator_start_time: datetime
) -> None:
"""[MessagesInfo] Проверка записей журнала о маскировании и размаскировании"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка записей журнала о маскировании и размаскировании. ЭФ: Журнал.Реальное время"
description = (
f"Проверка записей журнала о маскировании и размаскировании на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.mask_info_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo\n"
"Проверки: структура, фильтрация по времени, совпадение тегов между маскированием и снятием.\n"
"Примечание: тест выполняется после теста маскирования, чтобы записи успели попасть в журнал"
)
_apply_allure_markers(config.mask_info_in_journal_test, tag, title, description)
scenarios.mask_info_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_mask_du_on_mini_scheme(
self, ws_client: WebSocketClient, http_client: StandHttpClient, config: SmokeSuiteConfig
) -> None:
"""Проверка маскирования ДУ"""
tag = "MaskDuOnMiniScheme"
title = f"[{tag}] проверка маскирования линейного участка. ЭФ: Схема"
description = (
f"Проверка маскирования линейного участка, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.mask_signal_test.offset} мин.\n"
"Синхронные запросы типа: MaskLdsCommandRequest\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос маскирования.\n"
"Проверка маскирования ДУ в выходных сигналах.\n"
"Проверка маскирования на схеме.\n"
"Проверка сообщения о маскировании в журнале.\n"
)
_apply_allure_markers(config.mask_du_on_mini_scheme_test, tag, title, description)
await scenarios.mask_du_on_mini_scheme(ws_client, http_client, config)
@pytest.mark.asyncio
async def test_unmask_du_on_mini_scheme(
self, ws_client: WebSocketClient, http_client: StandHttpClient, config: SmokeSuiteConfig
) -> None:
"""Проверка размаскирования ДУ"""
tag = "UnmaskDuOnMiniScheme"
title = f"[{tag}] проверка снятия маскирования линейному участку. ЭФ: Схема"
description = (
f"Проверка снятия маскирования линейному участку, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.unmask_du_on_mini_scheme_test.offset} мин.\n"
"Синхронные запросы типа: UnmaskDuOnMiniScheme\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос снятия маски.\n"
"Проверка тега с маской в выходных сигналах.\n"
"Проверка сообщения о снятии маскирования в журнале.\n"
)
_apply_allure_markers(config.unmask_du_on_mini_scheme_test, tag, title, description)
await scenarios.unmask_du_on_mini_scheme(ws_client, http_client, config)
@pytest.mark.asyncio
async def test_lds_status_initialization_out(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[CommonScheme] Проверка выхода СОУ из Инициализации"""
tag = "CommonScheme"
title = f"[{tag}] Проверка выхода СОУ из Инициализации. ЭФ: Схема"
_apply_allure_markers(
config.lds_status_initialization_out_test,
tag,
title,
(
f"Проверка выхода СОУ из Инициализации на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.lds_status_initialization_out_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: не Инициализация"
),
)
await scenarios.lds_status_initialization_out(ws_client, config)
def test_lds_status_init_out_in_journal(
self, http_client: StandHttpClient, config: SmokeSuiteConfig, imitator_start_time: datetime
) -> None:
"""[MessagesInfo] Проверка записи в журнале о выходе СОУ из Инициализации"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка записи в журнале: выход СОУ из Инициализации. ЭФ: Журнал. Реальное время"
description = (
f"Проверка записи в журнале о выходе СОУ из Инициализации на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.lds_status_init_out_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
)
_apply_allure_markers(config.lds_status_init_out_in_journal_test, tag, title, description)
scenarios.lds_status_init_out_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_main_page_info_unstationary(self, ws_client: WebSocketClient, config: SmokeSuiteConfig) -> None:
"""[MainPageInfo] Проверка установки режима Нестационарный (для multi-leak)"""
tag = "MainPageInfo"
title = (
f"[{tag}] Проверка установки режима работы МТ: нестационарный. ЭФ: Главная страница. Контент таблица по ТУ"
)
_apply_allure_markers(
config.main_page_info_unstationary_test,
tag,
title,
(
f"Проверка установки режима работы МТ: нестационарный на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.main_page_info_unstationary_test.offset} мин.\n"
"Подписка на сообщения типа: MainPageInfo\n"
"Ожидаемый режим работы МТ: Нестационарный"
),
)
await scenarios.main_page_info_unstationary(ws_client, config)
def test_mode_mt_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка наличия сообщения о режиме МТ в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка сообщений о режиме МТ. ЭФ Журнал."
description = (
f"Проверка сообщений о режиме МТ на наборе данных {config.suite_name}, \n"
f"Время проведения проверки {config.mode_mt_in_journal_test.offset}"
)
_apply_allure_markers(config.mode_mt_in_journal_test, tag, title, description)
test_data = config.exp_mode_mt_message
scenarios.stationary_status_in_journal(http_client, config, imitator_start_time, test_data)
@pytest.mark.skip("Не готов после перехода на REST API, включить в задаче LDS-14843")
@pytest.mark.asyncio
async def test_export_lds_status_report(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
tag = "ExportReports"
title = f"[{tag}] Проверка формирования отчёта о режиме работы СОУ. ЭФ: Выпадашка отчётов"
_apply_allure_markers(
config.export_lds_status_report_test,
tag,
title,
(
f"Проверка формирования и содержимого xlsx-отчёта о режиме работы СОУ на наборе данных "
f"{config.suite_name},\n"
f"на технологическом участке {config.tu_name}\n"
f"Период отчёта: от старта имитатора до старта + сдвиг теста "
f"{config.export_lds_status_report_test.offset} мин.\n"
"Этапы сценария:\n"
"1) SubscribeReportsDataExportedRequest - подписка на пуш-нотификации\n"
"2) ExportReportsCommandRequest - запрос формирования отчёта (тип LdsStateReport, фильтр по периоду)\n"
"3) Ожидание ReportDataExportedNotification\n"
"4) Лонг-поллинг GetExportedDataListRequest - поиск отчёта в списке\n"
"5) DownloadExportedDataRequest (StreamInvocation) - скачивание по exportedDataId\n"
"6) Проверка xlsx: участки, длительности режимов СОУ, суммарное время работы\n"
"7) Проверка двойной шапки и названий колонок\n"
"8) Проверка имени файла (.xlsx, название отчёта, ТУ, период +-1 мин)\n"
"Во вложениях Allure xlsx прикладывается только при падении теста"
),
)
await scenarios.export_lds_status_report(ws_client, config, imitator_start_time)
@pytest.mark.skip("Не готов после перехода на REST API, включить в задаче LDS-14843")
@pytest.mark.asyncio
async def test_export_mt_mode_report(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
imitator_start_time: datetime,
) -> None:
tag = "ExportReports"
title = f"[{tag}] Проверка формирования отчёта о режиме работы МТ. ЭФ: Выпадашка отчётов"
_apply_allure_markers(
config.export_mt_mode_report_test,
tag,
title,
(
f"Проверка формирования и содержимого xlsx-отчёта о режиме работы МТ на наборе данных "
f"{config.suite_name},\n"
f"на технологическом участке {config.tu_name}\n"
f"Период отчёта: от старта имитатора до старта + сдвиг теста "
f"{config.export_mt_mode_report_test.offset} мин.\n"
"Этапы сценария:\n"
"1) SubscribeReportsDataExportedRequest - подписка на пуш-нотификации\n"
"2) ExportReportsCommandRequest - запрос формирования отчёта (тип StationaryStatusReport)\n"
"3) Ожидание ReportDataExportedNotification\n"
"4) Лонг-поллинг GetExportedDataListRequest - поиск отчёта в списке\n"
"5) DownloadExportedDataRequest (StreamInvocation) - скачивание по exportedDataId\n"
"6) Проверка xlsx: участки, длительности режимов МТ, суммарное время, доминирующий режим\n"
"7) Проверка двойной шапки и названий колонок\n"
"8) Проверка имени файла (.xlsx, название отчёта, ТУ, период +-1 мин)\n"
"Во вложениях Allure xlsx прикладывается только при падении теста"
),
)
await scenarios.export_mt_mode_report(ws_client, config, imitator_start_time)
# ===== ТЕСТЫ УРОВНЯ УТЕЧКИ =====
# Запускаются для каждой утечки в конфиге
@pytest.mark.parametrize("config, leak, leak_number", LEAK_PARAMS)
class TestLeakScenarios:
"""
Тесты уровня утечки.
Для single-leak: запускаются один раз.
Для multi-leak: запускаются для каждой утечки отдельно.
"""
@pytest.mark.asyncio
async def test_all_leaks_info(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[AllLeaksInfo] Проверка утечки в пуше"""
tag = "AllLeaksInfo"
title = f"[{tag}] Проверка сообщения об утечке. ЭФ: Пуш-сообщение об утечке на всех ЭФ"
_apply_allure_markers(
leak.all_leaks_info_test,
tag,
title,
(
f"Проверка пуш-сообщения об утечке на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.all_leaks_info_test.offset} мин.\n"
"Подписка на сообщения типа: AllLeaksInfo\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: тесты сообщений об утечке должны выполняться раньше теста на квитирование"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.all_leaks_info(ws_client, config, leak, imitator_start_time)
@pytest.mark.asyncio
async def test_all_leaks_is_empty(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""[AllLeaksInfo] Проверка отсутствия информации об утечке"""
tag = "AllLeaksInfo"
title = f"[{tag}] Проверка отсутствия информации об утечке в сообщении. ЭФ: Пуш-сообщение об утечке"
_apply_allure_markers(
leak.all_leaks_is_empty_test,
tag,
title,
(
"Проверка отсутствия информации об утечке в сообщении, \n"
f"Время проведения проверки: {leak.all_leaks_is_empty_test.offset} мин.\n"
"Подписка на сообщения типа: AllLeaksInfo\n"
"Примечание: проверяем, когда в КГ табличный вид список пуст"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.all_leaks_is_empty(ws_client, config)
@pytest.mark.asyncio
async def test_leaks_content(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[LeaksContent] Проверка сообщения об утечке в таблице КГ"""
tag = "LeaksContent"
title = f"[{tag}] Проверка сообщения об утечке. ЭФ: КГ.Табличное представление"
_apply_allure_markers(
leak.leaks_content_test,
tag,
title,
(
f"Проверка сообщения об утечке в таблице КГ на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leaks_content_test.offset} мин.\n"
"Подписка на сообщения типа: LeaksContent\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: тесты сообщений об утечке должны выполняться раньше теста на квитирование"
),
)
# Добавляем номер утечки в title для multi-leak
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leaks_content(ws_client, config, leak, imitator_start_time)
def test_leak_info_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка сообщения об утечке в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка сообщения об утечке в журнале. ЭФ: Журнал.Реальное время"
_apply_allure_markers(
leak.leak_info_in_journal,
tag,
title,
(
f"Проверка сообщения об утечке в журнале на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leak_info_in_journal.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: тесты сообщений об утечке должны выполняться раньше теста на квитирование"
),
)
# Добавляем номер утечки в title для multi-leak
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
scenarios.leak_info_in_journal(http_client, config, leak, imitator_start_time)
def test_completed_leak_info_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка сообщения о завершенной утечке в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка сообщения о завершенной утечке в журнале. ЭФ: Журнал.Реальное время"
description = (
f"Проверка сообщения о завершенной утечке в журнале на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.completed_leak_info_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo\n"
)
_apply_allure_markers(
leak.completed_leak_info_in_journal_test,
tag,
title,
description,
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
scenarios.completed_leak_info_in_journal(http_client, config, leak, imitator_start_time)
def test_possible_leak_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка наличия сообщения 'Возможна утечка' в журнале"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка сообщения 'Возможна утечка' в журнале. ЭФ: Журнал. Реальное время"
description = (
f"Проверка наличия сообщения 'Возможна утечка' в журнале на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.possible_leak_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo"
)
_apply_allure_markers(leak.possible_leak_in_journal_test, tag, title, description)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
scenarios.possible_leak_in_journal(http_client, config, imitator_start_time)
@pytest.mark.asyncio
async def test_tu_leaks_info(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[TuLeaksInfo] Проверка утечки на ТУ"""
tag = "TuLeaksInfo"
title = f"[{tag}] Проверка сообщения об утечке. Сообщения на ЭФ: Схема, Минисхема, Гидроуклон"
_apply_allure_markers(
leak.tu_leaks_info_test,
tag,
title,
(
f"Проверка сообщения об утечке на схемах и гидроуклоне на наборе данных {config.suite_name}, \n"
f"на технологическом участке {leak.diagnostic_area_name}\n"
f"Время проведения проверки: {leak.tu_leaks_info_test.offset} мин.\n"
"Подписка на сообщения типа: TuLeaksInfo\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: тесты сообщений об утечке должны выполняться раньше теста на квитирование"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.tu_leaks_info(ws_client, config, leak, imitator_start_time)
@pytest.mark.asyncio
async def test_lds_status_during_leak(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""[CommonScheme] Проверка режима работы СОУ во время утечки"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ во время утечки. ЭФ: Схема"
_apply_allure_markers(
leak.lds_status_during_leak_test,
tag,
title,
(
f"Проверка режима работы СОУ во время утечки на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.lds_status_during_leak_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Примечание: проверка режимов СОУ во время утечки должна выполняться раньше теста на квитирование\n"
"В рамках данного теста проверяется режим СОУ на ДУ с утечкой и на соседних ДУ"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.lds_status_during_leak(ws_client, config, leak)
@pytest.mark.skip("Не готов после перехода на REST API, включить в задаче LDS-14845")
@pytest.mark.asyncio
async def test_acknowledge_leak_info(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""[AcknowledgeLeak] Проверка квитирования утечки"""
tag = "AcknowledgeLeak"
title = f"[{tag}] Проверка квитирования утечки. Отсутствие Пуш-сообщений об утечке на всех ЭФ"
_apply_allure_markers(
leak.acknowledge_leak_test,
tag,
title,
(
"Проверка квитирования утечки на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.acknowledge_leak_test.offset} мин.\n"
"Синхронный запрос типа: AcknowledgeLeak\n"
"Подписка на сообщения типа: TuLeaksInfo, AllLeaksInfo\n"
"Проверки:\n"
"Статус-код ответа на синхронный запрос о квитировании,\n"
"Отсутствие пуш-сообщений об утечках после квитирования"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.acknowledge_leak_info(ws_client, http_client, config, leak)
@pytest.mark.skip("Не готов после перехода на REST API, включить в задаче LDS-14845")
def test_acknowledge_leak_in_journal(
self,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка записи в журнале о квитировании утечки"""
tag = "MessagesInfo"
title = f"[{tag}] Проверка записи в журнале о квитировании утечки. ЭФ: Журнал. Реальное время"
description = (
f"Проверка записи о квитировании утечки в журнале на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.acknowledge_leak_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром userActions=LEAK_ACK\n"
)
_apply_allure_markers(leak.acknowledge_leak_in_journal_test, tag, title, description)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
scenarios.acknowledge_leak_in_journal(http_client, config, imitator_start_time)
@pytest.mark.skip("Не готов после перехода на REST API, включить в задаче LDS-14845")
@pytest.mark.asyncio
async def test_output_signals(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[OutputSignalsInfo] Проверка данных об утечке в выходных сигналах"""
tag = "OutputSignalsInfo"
title = (
f"[{tag}] Проверка наличия данных об утечке в выходных сигналах. ЭФ: Диагностика сигналов.Выходные "
f"сигналы"
)
_apply_allure_markers(
leak.output_signals_test,
tag,
title,
(
f"Проверка наличия данных об утечке в выходных сигналах на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.output_signals_test.offset} мин.\n"
"Синхронный запрос типа: GetOutputSignals\n"
"Подписка на сообщения типа: SubscribeOutputSignals\n"
f"Допустимое время обнаружения {leak.allowed_time_diff_minutes} мин. с момента начала утечки, "
f"т к для данных {config.suite_name} интенсивность утечки {leak.leak_rate_percentages}%.\n"
"Примечание: "
"В offset указано время проверок сообщения выходных сигналов + 1 минута "
"для корректной отработки проверок.\n"
"Данный тест так же проверяет квитирование, offset выставлять после запуска теста на квитирование "
"утечки"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.output_signals(ws_client, http_client, config, leak, imitator_start_time)
@pytest.mark.asyncio
async def test_balance_algorithm_leak_waiting(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
"""[BalanceAlgorithmResultsContent] Проверка сообщения 'подозрение на утечку' в таблице КГ"""
tag = "BalanceAlgorithmResultsContent"
title = f"[{tag}] Проверка сообщения 'подозрение на утечку'. ЭФ: КГ.Графическое представление"
_apply_allure_markers(
leak.balance_algorithm_leak_waiting_test,
tag,
title,
(
f"Проверка сообщения 'подозрение на утечку' в таблице КГ, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: "
f"{leak.balance_algorithm_leak_waiting_test.offset} - "
f"{leak.balance_algorithm_leak_waiting_test.offset + Base_const.BALANCE_ALGORITHM_TOTAL_WAIT / 60} "
f"мин.\n"
"Подписка на сообщения типа: BalanceAlgorithmResultsContent\n"
),
)
# Добавляем номер утечки в title для multi-leak
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.balance_algorithm_leak_waiting(ws_client, config, leak, imitator_start_time)
@pytest.mark.asyncio
async def test_balance_algorithm_leak_detected(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""[BalanceAlgorithmResultsContent] Проверка наличия утечки (isLeakDetected) в графическом представлении КГ"""
tag = "BalanceAlgorithmResultsContent"
title = f"[{tag}] Проверка наличия утечки (isLeakDetected). ЭФ: КГ Графическое представление"
_apply_allure_markers(
leak.balance_algorithm_leak_detected_test,
tag,
title,
(
f"Проверка наличия утечки (isLeakDetected) в графическом представлении КГ, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.balance_algorithm_leak_detected_test.offset} мин.\n"
"Подписка на сообщения типа: BalanceAlgorithmResultsContent\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.balance_algorithm_leak_detected(ws_client, config, leak)
@pytest.mark.asyncio
async def test_balance_algorithm_leak_completed(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""
[BalanceAlgorithmResultsContent] Проверка отсутствия утечки (isLeakDetected) в графическом представлении КГ
"""
tag = "BalanceAlgorithmResultsContent"
title = (
f"[{tag}] Проверка отсутствия утечки (isLeakDetected) после завершения утечки. "
f"ЭФ: КГ Графическое представление"
)
_apply_allure_markers(
leak.balance_algorithm_leak_detected_test,
tag,
title,
(
f"Проверка отсутствия утечки (isLeakDetected) в графическом представлении КГ, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.balance_algorithm_leak_detected_test.offset} мин.\n"
"Подписка на сообщения типа: BalanceAlgorithmResultsContent\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.balance_algorithm_leak_completed(ws_client, config)
@pytest.mark.asyncio
async def test_lds_status_after_confirming_leak(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig, leak: LeakTestConfig, leak_number: int
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация
Причина: Накопление данных
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: Инициализация, по причине: Накопление данных. ЭФ: Схема"
_apply_allure_markers(
leak.lds_status_after_confirming_leak_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{leak.lds_status_after_confirming_leak_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Накопление данных"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
test_data = leak.lds_status_after_confirming_leak_data
await lds_status_scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
await lds_status_scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_completed_leak(
self, ws_client: WebSocketClient, config: SmokeSuiteConfig, leak: LeakTestConfig, leak_number: int
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ после завершения утечки: Инициализация
Причина: Накопление данных
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ после завершения утечки: Инициализация"
"Причина: Накопление данных. ЭФ: Схема"
)
_apply_allure_markers(
leak.lds_status_after_confirming_leak_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{leak.lds_status_completed_leak_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Накопление данных"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
test_data = leak.lds_status_after_completed_leak_data
await lds_status_scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_leak_is_complete_on_main_page(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка отсутствия утечки на ЭФ Состояние МТ"""
tag = "MainPageInfoContent"
title = f"[{tag}] Проверка отсутствия утечки на ЭФ Состояние МТ."
_apply_allure_markers(
leak.leak_is_complete_on_main_page_test,
tag,
title,
(
f"Проверка отсутствия утечки на ЭФ Состояние МТ\n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leak_is_complete_on_main_page_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leak_is_complete_on_main_page(ws_client, config)
@pytest.mark.asyncio
async def test_leak_is_confirm_on_main_page(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка подтвержденной утечки на ЭФ Состояние МТ"""
tag = "MainPageInfoContent"
title = f"[{tag}] Проверка подтвержденной утечки на ЭФ Состояние МТ"
_apply_allure_markers(
leak.leak_is_confirm_on_main_page_test,
tag,
title,
(
f"Проверка наличия утечки на ЭФ Состояние МТ \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leak_is_confirm_on_main_page_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leak_is_confirm_on_main_page(ws_client, config)
@pytest.mark.asyncio
async def test_the_leak_is_complete_on_kg(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка события завершения утечки в контроле герметичности"""
tag = "LeaksContent"
title = f"[{tag}] Проверка cобытия завершения утечки ЭФ КГ табличный вид."
_apply_allure_markers(
leak.the_leak_is_complete_on_kg_test,
tag,
title,
(
f"Проверка ЭФ КГ табличное представление, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.the_leak_is_complete_on_kg_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leak_is_complete_on_kg(ws_client, config, leak)
@pytest.mark.asyncio
async def test_leak_is_complete_in_output_signals(
self,
ws_client: WebSocketClient,
http_client: StandHttpClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка отсутствия утечки в выходных сигналах"""
tag = "OutputSignalsInfo"
title = f"[{tag}] Проверка отсутствия утечки в выходных сигналах."
_apply_allure_markers(
leak.leak_is_complete_in_output_signals_test,
tag,
title,
(
f"Проверка ЭФ Выходные сигналы \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.leak_is_complete_in_output_signals_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.leak_is_complete_in_output_signals(ws_client, http_client, config, leak)
@pytest.mark.asyncio
async def test_complete_tu_leaks_info_content(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
) -> None:
"""Проверка сообщения в журнале о завершении утечки"""
tag = "TuLeaksInfoContent"
title = f"[{tag}] Проверка отсутствия утечки на схеме"
_apply_allure_markers(
leak.complete_tu_leaks_info_content_test,
tag,
title,
(
f"Проверка ЭФ Схема, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {leak.complete_tu_leaks_info_content_test.offset} мин.\n"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.complete_tu_leaks_info_content(ws_client, config)
@pytest.mark.skip("Не готов после перехода на REST API, включить в задаче LDS-14843")
@pytest.mark.asyncio
async def test_export_leaks_report(
self,
ws_client: WebSocketClient,
config: SmokeSuiteConfig,
leak: LeakTestConfig,
leak_number: int,
imitator_start_time: datetime,
) -> None:
tag = "ExportReports"
title = f"[{tag}] Проверка формирования отчёта об утечках. ЭФ: Выпадашка отчётов"
_apply_allure_markers(
leak.export_leaks_report_test,
tag,
title,
(
f"Проверка формирования и содержимого xlsx-отчёта об утечках на наборе данных "
f"{config.suite_name},\n"
f"на технологическом участке {config.tu_name}\n"
f"Период отчёта: от старта имитатора до старта + сдвиг теста"
f"{leak.export_leaks_report_test.offset} мин.\n"
"Этапы сценария:\n"
"1) SubscribeReportsDataExportedRequest - подписка на пуш-нотификации\n"
"2) ExportReportsCommandRequest - запрос формирования отчёта (тип LeaksReport, фильтр по периоду)\n"
"3) Ожидание ReportDataExportedNotification - пуш-нотификация о формировании отчёта\n"
"4) Лонг-поллинг GetExportedDataListRequest - поиск отчёта в списке "
"(имя, ТУ, период start/end с погрешностью +-1 мин)\n"
"5) DownloadExportedDataRequest (StreamInvocation) - скачивание по exportedDataId\n"
"6) Приём fileChunk, проверка формата xlsx (zip-сигнатура, контент существует)\n"
"7) Проверка имени файла: .xlsx, 'Отчет об утечках', название ТУ, "
"диапазон дат в имени (+-1 мин)\n"
"8) Проверка шапки xlsx: заголовок с периодом (+-1 мин), названия колонок\n"
"9) Проверка строки утечки: дата в периоде, объект, режим СОУ, маскирование, "
"координата, объём, режим работы МТ\n"
"Во вложениях Allure xlsx прикладывается только при падении теста"
),
)
if config.has_multiple_leaks:
allure.dynamic.title(f"{title} (утечка #{leak_number})")
await scenarios.export_leaks_report(ws_client, config, leak, imitator_start_time)
test_stationary_status_regress.py
"""
Архитектура параметризации:
- SUITE_PARAMS: тесты уровня набора (один раз на набор данных LDSStatusConfig)
Для добавления нового набора:
1. Создать файл в test_config/datasets/ с LDSStatusConfig
2. Добавить маппинг новых тестов в conftest.py -> LDS_STATUS_SUITE_LEVEL_MAPPING
3. Тесты подхватятся автоматически
Запуск:
- Все тесты: pytest tests/test_lds_status_regress.py
- Один набор: pytest tests/test_lds_status_regress.py --suites=Lds_status_regress_in_flow
- Несколько наборов: pytest tests/test_lds_status_regress.py --suites=Lds_status_regress_in_flow,Lds_status_regress_2
"""
from typing import Any, List, Optional
import allure
import pytest
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from test_config.datasets import ALL_STATIONARY_STATUS_CONFIGS
from test_config.models_for_tests import CaseMarkers, StationaryStatusConfig
from test_scenarios import smoke_scenarios
from test_scenarios import stationary_status_scenarios as scenarios
# ===== ГЕНЕРАЦИЯ ПАРАМЕТРОВ =====
def _get_suite_markers(config: StationaryStatusConfig) -> List[pytest.MarkDecorator]:
"""Маркеры для тестового набора."""
return [
pytest.mark.test_suite_name(config.suite_name),
pytest.mark.test_suite_data_id(config.suite_data_id),
pytest.mark.test_data_name(config.archive_name),
pytest.mark.tu_id(config.technological_unit.id),
]
def _generate_suite_params() -> List[Any]:
"""
Генерирует параметры для тестов уровня набора данных.
Один параметр на каждый config.
"""
return [
pytest.param(config, id=config.suite_name, marks=_get_suite_markers(config))
for config in ALL_STATIONARY_STATUS_CONFIGS
]
# ===== ПАРАМЕТРЫ ДЛЯ ТЕСТОВ =====
SUITE_PARAMS: List[Any] = _generate_suite_params()
# ===== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =====
def _apply_allure_markers(test_config: CaseMarkers, tag: str, title: str, description: Optional[str] = None) -> None:
"""Применяет allure-маркеры из конфига теста."""
if not test_config:
pytest.skip("Не заполнена конфигурация теста: тест пропущен")
allure.dynamic.tag(tag)
allure.dynamic.tag("REGRESS")
allure.dynamic.title(title)
if description:
allure.dynamic.description(description)
# ===== ТЕСТЫ УРОВНЯ НАБОРА =====
# Запускаются один раз для каждого config
@pytest.mark.parametrize("config", SUITE_PARAMS)
class TestSuiteScenarios:
"""
Тесты уровня набора данных.
Запускаются один раз для каждого конфига.
"""
@pytest.mark.critical_stop
def test_stationary_status_basic_info(self, http_client: StandHttpClient, config: StationaryStatusConfig) -> None:
"""[BasicInfo] Проверка базовой информации СОУ: список ТУ"""
tag = "BasicInfo"
title = f"[{tag}] Проверка списка ТУ. ЭФ: Главная страница"
_apply_allure_markers(config.stationary_status_basic_info_test, tag, title)
smoke_scenarios.basic_info(http_client, config)
@pytest.mark.asyncio
async def test_stationary_status_check_with_reasons(
self, ws_client: WebSocketClient, config: StationaryStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы МТ и причины режима работы МТ на ЭФ: Схема
"""
test_data = config.stationary_status_check_with_reasons_test_data
expected_stationary_status, expected_stationary_status_reasons = test_data.expected_result
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы МТ: {expected_stationary_status},"
f" по причине: {expected_stationary_status_reasons}. ЭФ: Схема"
)
_apply_allure_markers(
config.stationary_status_check_with_reasons_test,
tag,
title,
(
f"Проверка режима работы МТ на одном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.stationary_status_check_with_reasons_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
f"Ожидаемый режим работы СОУ: {expected_stationary_status}\n "
f"Ожидаемая причина режима работы СОУ: {expected_stationary_status_reasons}"
),
)
await scenarios.stationary_status_check_with_reasons(ws_client, config, test_data)
def test_stationary_status_in_journal(self, http_client: StandHttpClient, config: StationaryStatusConfig) -> None:
"""
[MessagesInfo] Проверка режима работы МТ и причины режима работы МТ на ЭФ: Журнал
"""
test_data = config.stationary_status_in_journal_test_data
expected_stationary_status, expected_stationary_status_reasons = test_data.expected_result
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка режима работы МТ: {expected_stationary_status},"
f" по причине: {expected_stationary_status_reasons}. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.stationary_status_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.stationary_status_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=PUMPING_STATUS\n"
f"Ожидаемый режим работы СОУ: {expected_stationary_status}\n "
f"Ожидаемая причина режима работы СОУ: {expected_stationary_status_reasons}"
),
)
scenarios.stationary_status_in_journal(http_client, config, test_data)