https://pastein.ru/t/Ag8
Загрузка данных
датасет
"""
Конфигурация тестового набора is_rejected_regress
Особенности набора:
- Проверка отбраковки сигналов с датчиков давления и расходомеров
- Типы отбраковки: empty, quality, VTOR, nearbySensors,
diagnosticInfo, constantSignal, range
- Для каждого типа - отдельный RejectionTestCase на снятие отбраковки
"""
from constants.enums import TU, RejectionCriteria, RejectionSensorTag
from test_config.models_for_tests import CaseMarkers, IsRejectedConfig, RejectionTestCase
# ===== Константы набора =====
SUITE_NAME = "is_rejected_regress"
SUITE_DATA_ID = 183
ARCHIVE_NAME = f"{SUITE_NAME}.tar.gz"
TECHNOLOGICAL_UNIT = TU.TIKHORETSK_NOVOROSSIYSK_3
MAIN_PIPELINE = "МН Тихорецк-Новороссийск-3"
# ===== Тегированные датчики =====
FLOW_KRIM = RejectionSensorTag.NPS_KRIM_P_Vmom
PRESSURE_VELKRIM = RejectionSensorTag.KP_209_1_Pin
FLOW_TIH = RejectionSensorTag.NPS_TIH_5_Vmom
PRESSURE_KP7 = RejectionSensorTag.KP_7_Pin
PRESSURE_KP8_PIN = RejectionSensorTag.KP_8_Pin
PRESSURE_KP8_POUT = RejectionSensorTag.KP_8_Pout
# ===== Ожидаемые signalName =====
SIGNAL_FLOW = "Расход"
SIGNAL_PRESSURE = "Значение давления"
# ===== Конфигурация набора =====
IS_REJECTED_REGRESS_CONFIG = IsRejectedConfig(
# ----- Метаданные -----
suite_name=SUITE_NAME,
suite_data_id=SUITE_DATA_ID,
archive_name=ARCHIVE_NAME,
technological_unit=TECHNOLOGICAL_UNIT,
main_pipeline=MAIN_PIPELINE,
rejection_cases=[
# ===== emptyFilterSettings =====
RejectionTestCase(
name="empty_flow",
sensor=FLOW_KRIM,
expected_event="Отбраковка по отсутствию значения",
expected_signal_name=SIGNAL_FLOW,
expected_criteria_names=RejectionCriteria.EMPTY,
time_range_start_s=0,
time_range_end_s=240,
rejection_input_signals_test=CaseMarkers(test_case_id="189", offset=3),
rejection_main_page_test=CaseMarkers(test_case_id="189", offset=3),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="189", offset=3),
),
RejectionTestCase(
name="empty_flow_clearance",
sensor=FLOW_KRIM,
expected_event="Отбраковка по отсутствию значения снята",
expected_signal_name=SIGNAL_FLOW,
expected_is_rejected=False,
time_range_start_s=0,
time_range_end_s=240,
rejection_input_signals_test=CaseMarkers(test_case_id="189", offset=5),
rejection_journal_test=CaseMarkers(test_case_id="189", offset=5),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="189", offset=5),
),
RejectionTestCase(
name="empty_pressure",
sensor=PRESSURE_VELKRIM,
expected_event="Отбраковка по отсутствию значения",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.EMPTY,
time_range_start_s=0,
time_range_end_s=540,
rejection_input_signals_test=CaseMarkers(test_case_id="190", offset=4),
rejection_main_page_test=CaseMarkers(test_case_id="190", offset=6),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="190", offset=5),
),
RejectionTestCase(
name="empty_pressure_clearance",
sensor=PRESSURE_VELKRIM,
expected_event="Отбраковка по отсутствию значения снята",
expected_signal_name=SIGNAL_PRESSURE,
expected_is_rejected=False,
time_range_start_s=0,
time_range_end_s=540,
rejection_input_signals_test=CaseMarkers(test_case_id="190", offset=7),
rejection_journal_test=CaseMarkers(test_case_id="190", offset=7),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="190", offset=7),
),
# ===== qualityFilterSettings =====
RejectionTestCase(
name="quality_flow",
sensor=FLOW_KRIM,
expected_event="Отбраковка по качеству",
expected_signal_name=SIGNAL_FLOW,
expected_criteria_names=RejectionCriteria.QUALITY,
time_range_start_s=600,
time_range_end_s=840,
rejection_input_signals_test=CaseMarkers(test_case_id="191", offset=13),
rejection_journal_test=CaseMarkers(test_case_id="191", offset=14),
rejection_main_page_test=CaseMarkers(test_case_id="191", offset=11),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="191", offset=12),
),
RejectionTestCase(
name="quality_flow_clearance",
sensor=FLOW_KRIM,
expected_event="Отбраковка по качеству снята",
expected_signal_name=SIGNAL_FLOW,
expected_is_rejected=False,
time_range_start_s=600,
time_range_end_s=840,
rejection_input_signals_test=CaseMarkers(test_case_id="191", offset=15),
rejection_journal_test=CaseMarkers(test_case_id="191", offset=15),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="191", offset=15),
),
RejectionTestCase(
name="quality_pressure",
sensor=PRESSURE_VELKRIM,
expected_event="Отбраковка по качеству",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.QUALITY,
time_range_start_s=900,
time_range_end_s=1140,
rejection_input_signals_test=CaseMarkers(test_case_id="205", offset=18),
rejection_journal_test=CaseMarkers(test_case_id="205", offset=19),
rejection_main_page_test=CaseMarkers(test_case_id="205", offset=16),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="", offset=17),
),
RejectionTestCase(
name="quality_pressure_clearance",
sensor=PRESSURE_VELKRIM,
expected_event="Отбраковка по качеству снята",
expected_signal_name=SIGNAL_PRESSURE,
expected_is_rejected=False,
time_range_start_s=900,
time_range_end_s=1140,
rejection_input_signals_test=CaseMarkers(test_case_id="205", offset=20),
rejection_journal_test=CaseMarkers(test_case_id="205", offset=20),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="205", offset=20),
),
# ===== vtorFilterSettings =====
RejectionTestCase(
name="vtor_flow",
sensor=FLOW_TIH,
expected_event="Отбраковка по сигналу ВТОР",
expected_signal_name=SIGNAL_FLOW,
expected_criteria_names=RejectionCriteria.VTOR,
time_range_start_s=1200,
time_range_end_s=1440,
rejection_input_signals_test=CaseMarkers(test_case_id="206", offset=23),
rejection_journal_test=CaseMarkers(test_case_id="206", offset=24),
rejection_main_page_test=CaseMarkers(test_case_id="206", offset=21),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="206", offset=22),
),
RejectionTestCase(
name="vtor_flow_clearance",
sensor=FLOW_TIH,
expected_event="Отбраковка по сигналу ВТОР снята",
expected_signal_name=SIGNAL_FLOW,
expected_is_rejected=False,
time_range_start_s=1200,
time_range_end_s=1440,
rejection_input_signals_test=CaseMarkers(test_case_id="206", offset=25),
rejection_journal_test=CaseMarkers(test_case_id="206", offset=25),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="206", offset=25),
),
RejectionTestCase(
name="vtor_pressure",
sensor=PRESSURE_KP7,
expected_event="Отбраковка по сигналу ВТОР",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.VTOR,
time_range_start_s=1500,
time_range_end_s=1740,
rejection_input_signals_test=CaseMarkers(test_case_id="207", offset=28),
rejection_journal_test=CaseMarkers(test_case_id="207", offset=29),
rejection_main_page_test=CaseMarkers(test_case_id="207", offset=26),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="207", offset=27),
),
RejectionTestCase(
name="vtor_pressure_clearance",
sensor=PRESSURE_KP7,
expected_event="Отбраковка по сигналу ВТОР снята",
expected_signal_name=SIGNAL_PRESSURE,
expected_is_rejected=False,
time_range_start_s=1500,
time_range_end_s=1740,
rejection_input_signals_test=CaseMarkers(test_case_id="207", offset=30),
rejection_journal_test=CaseMarkers(test_case_id="207", offset=30),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="207", offset=30),
),
# ===== nearbySensorsFilterSettings =====
RejectionTestCase(
name="nearby_pressure_pin",
sensor=PRESSURE_KP8_PIN,
expected_event="Отбраковка по разнице показаний СИ давления на КП",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.NEARBY,
time_range_start_s=1800,
time_range_end_s=2040,
rejection_input_signals_test=CaseMarkers(test_case_id="192", offset=32),
rejection_journal_test=CaseMarkers(test_case_id="192", offset=32.5),
rejection_main_page_test=CaseMarkers(test_case_id="192", offset=31),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="192", offset=31.5),
),
RejectionTestCase(
name="nearby_pressure_pin_clearance",
sensor=PRESSURE_KP8_PIN,
expected_event="Отбраковка по разнице показаний СИ давления на КП снята",
expected_signal_name=SIGNAL_PRESSURE,
expected_is_rejected=False,
time_range_start_s=1800,
time_range_end_s=2040,
rejection_input_signals_test=CaseMarkers(test_case_id="192", offset=33.5),
rejection_journal_test=CaseMarkers(test_case_id="192", offset=33.5),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="192", offset=33.5),
),
RejectionTestCase(
name="nearby_pressure_pout",
sensor=PRESSURE_KP8_POUT,
expected_event="Отбраковка по разнице показаний СИ давления на КП",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.NEARBY,
time_range_start_s=1800,
time_range_end_s=2040,
rejection_input_signals_test=CaseMarkers(test_case_id="193", offset=34),
rejection_journal_test=CaseMarkers(test_case_id="193", offset=34.5),
rejection_main_page_test=CaseMarkers(test_case_id="193", offset=33),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="193", offset=33.5),
),
RejectionTestCase(
name="nearby_pressure_pout_clearance",
sensor=PRESSURE_KP8_POUT,
expected_event="Отбраковка по разнице показаний СИ давления на КП снята",
expected_signal_name=SIGNAL_PRESSURE,
expected_is_rejected=False,
time_range_start_s=1800,
time_range_end_s=2040,
rejection_input_signals_test=CaseMarkers(test_case_id="193", offset=35.5),
rejection_journal_test=CaseMarkers(test_case_id="193", offset=35.5),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="193", offset=35.5),
),
# ===== diagnosticInfoFilterSettings =====
RejectionTestCase(
name="diagnostic_info_flow",
sensor=FLOW_TIH,
expected_event="Отбраковка по диагностической информации",
expected_signal_name=SIGNAL_FLOW,
expected_criteria_names=RejectionCriteria.DIAGNOSTIC_INFO,
time_range_start_s=2100,
time_range_end_s=2340,
rejection_input_signals_test=CaseMarkers(test_case_id="194", offset=38),
# rejection_journal_test=CaseMarkers(test_case_id="194", offset=39), # blocked by LDS-12394
rejection_main_page_test=CaseMarkers(test_case_id="194", offset=36),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="194", offset=37),
),
RejectionTestCase(
name="diagnostic_info_flow_clearance",
sensor=FLOW_TIH,
expected_event="Отбраковка по диагностической информации снята",
expected_signal_name=SIGNAL_FLOW,
expected_is_rejected=False,
time_range_start_s=2100,
time_range_end_s=2340,
rejection_input_signals_test=CaseMarkers(test_case_id="194", offset=40),
rejection_journal_test=CaseMarkers(test_case_id="194", offset=40),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="194", offset=40),
),
# ===== constantSignalFilter =====
RejectionTestCase(
name="constant_signal_flow",
sensor=FLOW_TIH,
expected_event="Отбраковка по постоянному сигналу",
expected_signal_name=SIGNAL_FLOW,
expected_criteria_names=RejectionCriteria.CONSTANT_SIGNAL,
time_range_start_s=2400,
time_range_end_s=2640,
rejection_input_signals_test=CaseMarkers(test_case_id="208", offset=43),
rejection_journal_test=CaseMarkers(test_case_id="208", offset=44),
rejection_main_page_test=CaseMarkers(test_case_id="208", offset=41),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="208", offset=42),
),
RejectionTestCase(
name="constant_signal_flow_clearance",
sensor=FLOW_TIH,
expected_event="Отбраковка по постоянному сигналу снята",
expected_signal_name=SIGNAL_FLOW,
expected_is_rejected=False,
time_range_start_s=2400,
time_range_end_s=2640,
rejection_input_signals_test=CaseMarkers(test_case_id="208", offset=45),
rejection_journal_test=CaseMarkers(test_case_id="208", offset=45),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="208", offset=45),
),
RejectionTestCase(
name="constant_signal_pressure",
sensor=PRESSURE_KP8_PIN,
expected_event="Отбраковка по постоянному сигналу",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.CONSTANT_SIGNAL,
time_range_start_s=2700,
time_range_end_s=2940,
rejection_input_signals_test=CaseMarkers(test_case_id="209", offset=48),
rejection_journal_test=CaseMarkers(test_case_id="209", offset=49),
rejection_main_page_test=CaseMarkers(test_case_id="209", offset=46),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="209", offset=47),
),
RejectionTestCase(
name="constant_signal_pressure_clearance",
sensor=PRESSURE_KP8_PIN,
expected_event="Отбраковка по постоянному сигналу снята",
expected_signal_name=SIGNAL_PRESSURE,
expected_is_rejected=False,
time_range_start_s=2700,
time_range_end_s=2940,
rejection_input_signals_test=CaseMarkers(test_case_id="209", offset=50),
rejection_journal_test=CaseMarkers(test_case_id="209", offset=50),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="209", offset=50),
),
# ===== rangeFilterSettings =====
RejectionTestCase(
name="range_upper_flow",
sensor=FLOW_TIH,
expected_event="Отбраковка по допустимому диапазону",
expected_signal_name=SIGNAL_FLOW,
expected_criteria_names=RejectionCriteria.RANGE,
time_range_start_s=3000,
time_range_end_s=3240,
rejection_input_signals_test=CaseMarkers(test_case_id="195", offset=53),
rejection_journal_test=CaseMarkers(test_case_id="195", offset=54),
rejection_main_page_test=CaseMarkers(test_case_id="195", offset=51),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="195", offset=52),
),
RejectionTestCase(
name="range_upper_flow_clearance",
sensor=FLOW_TIH,
expected_event="Отбраковка по допустимому диапазону снята",
expected_signal_name=SIGNAL_FLOW,
expected_is_rejected=False,
time_range_start_s=3000,
time_range_end_s=3240,
rejection_input_signals_test=CaseMarkers(test_case_id="195", offset=55),
rejection_journal_test=CaseMarkers(test_case_id="195", offset=55),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="195", offset=55),
),
RejectionTestCase(
name="range_lower_flow",
sensor=FLOW_TIH,
expected_event="Отбраковка по допустимому диапазону",
expected_signal_name=SIGNAL_FLOW,
expected_criteria_names=RejectionCriteria.RANGE,
time_range_start_s=3300,
time_range_end_s=3540,
rejection_input_signals_test=CaseMarkers(test_case_id="197", offset=58),
rejection_journal_test=CaseMarkers(test_case_id="197", offset=59),
rejection_main_page_test=CaseMarkers(test_case_id="197", offset=56),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="197", offset=57),
),
RejectionTestCase(
name="range_lower_flow_clearance",
sensor=FLOW_TIH,
expected_event="Отбраковка по допустимому диапазону снята",
expected_signal_name=SIGNAL_FLOW,
expected_is_rejected=False,
time_range_start_s=3300,
time_range_end_s=3540,
rejection_input_signals_test=CaseMarkers(test_case_id="197", offset=60),
rejection_journal_test=CaseMarkers(test_case_id="197", offset=60),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="197", offset=60),
),
RejectionTestCase(
name="range_upper_pressure",
sensor=PRESSURE_KP8_PIN,
expected_event="Отбраковка по допустимому диапазону",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.RANGE,
time_range_start_s=3600,
time_range_end_s=3840,
rejection_input_signals_test=CaseMarkers(test_case_id="196", offset=63),
rejection_journal_test=CaseMarkers(test_case_id="196", offset=64),
rejection_main_page_test=CaseMarkers(test_case_id="196", offset=61),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="196", offset=62),
),
RejectionTestCase(
name="range_upper_pressure_clearance",
sensor=PRESSURE_KP8_PIN,
expected_event="Отбраковка по допустимому диапазону снята",
expected_signal_name=SIGNAL_PRESSURE,
expected_is_rejected=False,
time_range_start_s=3600,
time_range_end_s=3840,
rejection_input_signals_test=CaseMarkers(test_case_id="196", offset=65),
rejection_journal_test=CaseMarkers(test_case_id="196", offset=65),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="196", offset=65),
),
RejectionTestCase(
name="range_lower_pressure",
sensor=PRESSURE_KP8_PIN,
expected_event="Отбраковка по допустимому диапазону",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.RANGE,
time_range_start_s=3900,
time_range_end_s=4140,
rejection_input_signals_test=CaseMarkers(test_case_id="198", offset=68),
rejection_journal_test=CaseMarkers(test_case_id="198", offset=69),
rejection_main_page_test=CaseMarkers(test_case_id="198", offset=66),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="198", offset=67),
),
RejectionTestCase(
name="range_lower_pressure_clearance",
sensor=PRESSURE_KP8_PIN,
expected_event="Отбраковка по допустимому диапазону снята",
expected_signal_name=SIGNAL_PRESSURE,
expected_is_rejected=False,
time_range_start_s=3900,
time_range_end_s=4140,
rejection_input_signals_test=CaseMarkers(test_case_id="198", offset=70),
rejection_journal_test=CaseMarkers(test_case_id="198", offset=70),
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="198", offset=70),
),
],
)
модели
@dataclass
class RejectionTestCase:
"""
Описание одного события отбраковки для тестирования.
Содержит:
- Тег и id датчика (из RejectionSensorTag)
- Ожидаемые значения для проверок журнала и схемы
- Маркеры (offset и test_case_id)
"""
name: str = ""
sensor: RejectionSensorTag = ""
expected_event: str = ""
expected_signal_name: str = ""
expected_criteria_names: Optional[RejectionCriteria] = None
expected_is_rejected: bool = True
time_range_start_s: float = 0
time_range_end_s: float = 0
rejection_input_signals_test: Optional[CaseMarkers] = None
rejection_journal_test: Optional[CaseMarkers] = None
rejection_main_page_test: Optional[CaseMarkers] = None
rejection_scheme_signals_state_test: Optional[CaseMarkers] = None
@dataclass
class IsRejectedConfig(BaseSuiteConfig):
"""
Конфигурация тестового набора отбраковки сигналов.
Структура:
1. Название МН
2. Список случаев отбраковки (RejectionTestCase)
"""
сцен
"""
Сценарии тестов - функции-обёртки без pytest маркеров.
Каждая функция содержит логику одного теста.
Pytest маркеры и allure декораторы применяются в тестовых файлах.
"""
import allure
import pytest
from constants.enums import Direction, MessageType, RejectionCriteria
from constants.test_constants import BaseTN3Constants as TestConst
from models.get_messages_model import Filtering, FilteringObjects, Pagination
from test_config.models_for_tests import IsRejectedConfig, RejectionTestCase
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
# ===== Сценарии отбраковки сигналов =====
async def rejection_input_signals(ws_client, cfg: IsRejectedConfig, rejection_case: RejectionTestCase):
"""
Проверка отбраковки сигнала по подписке SubscribeInputSignalsRequest.
Ожидание isRejected задаётся в RejectionTestCase.expected_is_rejected.
"""
sensor = rejection_case.sensor
with allure.step(
f"Подключение по ws, получение данных InputSignalsContent для датчика {sensor.description} (id={sensor.id})"
):
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)
sensor_data = parsed_payload.replyContent.inputSignals
target_signal = t_utils.find_object_by_field(sensor_data, "id", sensor.id)
with SoftAssertions() as soft_failures:
StepCheck(
f"Проверка isRejected={rejection_case.expected_is_rejected} "
f"для датчика {sensor.description} (id={sensor.id})",
"isRejected",
soft_failures,
).actual(target_signal.isRejected).expected(rejection_case.expected_is_rejected).equal_to()
if rejection_case.expected_is_rejected and rejection_case.expected_criteria_names:
raw_criteria = (
target_signal.rejection.get(TestConst.CRITERIA_NAMES_FIELD)
if isinstance(target_signal.rejection, dict)
else None
)
criteria = RejectionCriteria(raw_criteria) if raw_criteria is not None else None
StepCheck(
f"Проверка rejection.criteriaNames для {sensor.description} (id={sensor.id})",
TestConst.CRITERIA_NAMES_FIELD,
soft_failures,
).actual(criteria).expected(rejection_case.expected_criteria_names).equal_to()
async def rejection_journal(
ws_client, cfg: IsRejectedConfig, rejection_case: RejectionTestCase, imitator_start_time
):
"""
Проверка наличия записи об отбраковке в журнале по GetMessagesRequest.
"""
sensor = rejection_case.sensor
expected_event = rejection_case.expected_event
with allure.step("Подготовка запроса и ожидаемого диапазона времени"):
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_REJECT_LIMIT, direction=Direction.FIRST.value),
filtering=Filtering(
messageTypes=int(MessageType.REJECTION),
objects=FilteringObjects(tuId=cfg.tu_id),
),
)
range_start, range_end = t_utils.get_rejection_time_window(
imitator_start_time=imitator_start_time,
start_seconds=rejection_case.time_range_start_s,
reserve_seconds=TestConst.SEC_PER_MIN,
)
with allure.step("Получение сообщений журнала с фильтром messageTypes=REJECTION"):
payload = await t_utils.connect_and_get_msg(ws_client, "GetMessagesRequest", request_body)
parsed_payload = parser.parse_journal_msg(payload)
messages_info = parsed_payload.replyContent.messagesInfo
with allure.step("Проверка наличия сообщений в журнале"):
StepCheck("Проверка наличия сообщений в журнале", "messagesInfo").actual(messages_info).is_not_empty()
with allure.step(
f"Подготовка сообщений к проверке по диапазону слоя данных "
f"({rejection_case.time_range_start_s - TestConst.SEC_PER_MIN}-"
f"{rejection_case.time_range_end_s + TestConst.SEC_PER_MIN} с от старта имитатора)"
):
time_filtered, target_msg = t_utils.find_rejection_journal_message(
messages_info=messages_info,
tag=sensor.description,
range_start=range_start,
range_end=range_end,
technological_section=cfg.tu_name,
expected_event=expected_event,
)
allure.attach(
f"Всего получено сообщений: {len(messages_info)}\n"
f"Диапазон фильтрации: {range_start} - {range_end}\n"
f"После фильтрации по tag='{sensor.description}' и времени: {len(time_filtered)}\n"
f"Найдено ли сообщение с technologicalSection='{cfg.tu_name}' и событием {expected_event}: "
f"{'True' if target_msg else 'False'}",
name="Результат фильтрации сообщений журнала",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(
f"Проверка: найдено ли сообщение с tag='{sensor.description}' (id={sensor.id}) "
f"в диапазоне {range_start}-{range_end} с"
):
if target_msg is None:
pytest.fail(
f"Сообщение с tag='{sensor.description}' (id={sensor.id}) "
f"и technologicalSection='{cfg.tu_name}' не найдено в диапазоне "
f"{range_start} - {range_end} "
f"(всего сообщений: {len(messages_info)}, после фильтрации: {len(time_filtered)})"
)
with SoftAssertions() as soft_failures:
StepCheck("Проверка mainPipeline", "mainPipeline", soft_failures).actual(target_msg.mainPipeline).expected(
cfg.main_pipeline
).equal_to()
StepCheck("Проверка messageType", "messageType", soft_failures).actual(target_msg.messageType).expected(
TestConst.JOURNAL_MESSAGE_TYPE_REJECTION
).equal_to()
StepCheck("Проверка technologicalSection не пустой", "technologicalSection", soft_failures).actual(
target_msg.technologicalSection
).is_not_none()
StepCheck("Проверка technologicalObject не пустой", "technologicalObject", soft_failures).actual(
target_msg.technologicalObject
).is_not_none()
StepCheck(f"Проверка tag для {sensor.description} (id={sensor.id})", "tag", soft_failures).actual(
target_msg.tag
).expected(sensor.description).equal_to()
if rejection_case.expected_signal_name:
StepCheck("Проверка signalName", "signalName", soft_failures).actual(target_msg.signalName).expected(
rejection_case.expected_signal_name
).equal_to()
if expected_event:
StepCheck("Проверка event", "event", soft_failures).actual(
(target_msg.event.rstrip() or "").strip()
).expected(expected_event).equal_to()
async def rejection_main_page(ws_client, cfg: IsRejectedConfig):
"""
Проверка numberOfRejectedSignals > 0 по подписке subscribeMainPageSignalsInfoRequest.
"""
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 SoftAssertions() as soft_failures:
StepCheck("Проверка id полученного ТУ", "tu_id", soft_failures).actual(
parsed_payload.replyContent.tuId
).expected(cfg.tu_id).equal_to()
StepCheck(
f"Проверка numberOfRejectedSignals > 0 для ТУ {cfg.tu_name}",
"numberOfRejectedSignals",
soft_failures,
).actual(parsed_payload.replyContent.signalsInfo.numberOfRejectedSignals).is_greater_than(0)
async def rejection_scheme_signals_state(ws_client, cfg: IsRejectedConfig, rejection_case: RejectionTestCase):
"""
Проверка отбраковки сигнала по подписке SubscribeSchemeSignalsStateRequest.
Проверяет isRejected, isMasked, isImitated и rejection.criteriaNames.
Логирование больших ответов подавляется suppress_recv_logging.
"""
sensor = rejection_case.sensor
ws_client.suppress_recv_logging = True
parser.suppress_recv_logging = True
try:
with allure.step(
f"Подключение по ws, получение данных SchemeSignalsStateContent "
f"для датчика {sensor.description} (id={sensor.id})"
):
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)
signals = parsed_payload.replyContent.signalsStates
target_signal = next(
(signal for signal in signals if signal.id == sensor.id),
None,
)
allure.attach(
f"Всего сигналов получено: {len(signals)}\n"
f"Поиск сигнала с id={sensor.id} ({sensor.description}): "
f"{'Найден' if target_signal else 'Не найден'}",
name="Результат поиска сигнала в SchemeSignalsState",
attachment_type=allure.attachment_type.TEXT,
)
if target_signal is not None:
allure.attach(
str(target_signal),
name=f"Тестируемый фрагмент ответа с бэка: сигнал id={sensor.id} ({sensor.description})",
attachment_type=allure.attachment_type.TEXT,
)
finally:
ws_client.suppress_recv_logging = False
parser.suppress_recv_logging = False
with allure.step(f"Проверка: найден ли сигнал с id={sensor.id} ({sensor.description})"):
if target_signal is None:
pytest.fail(
f"Сигнал с id={sensor.id} ({sensor.description}) " f"не найден среди {len(signals)} полученных сигналов"
)
with SoftAssertions() as soft_failures:
StepCheck(f"Проверка isRejected для {sensor.description} (id={sensor.id})", "isRejected", soft_failures).actual(
target_signal.isRejected
).expected(rejection_case.expected_is_rejected).equal_to()
StepCheck(f"Проверка isMasked для {sensor.description} (id={sensor.id})", "isMasked", soft_failures).actual(
target_signal.isMasked
).expected(False).equal_to()
StepCheck(f"Проверка isImitated для {sensor.description} (id={sensor.id})", "isImitated", soft_failures).actual(
target_signal.isImitated
).expected(False).equal_to()
if rejection_case.expected_is_rejected and rejection_case.expected_criteria_names and target_signal.rejection is not None:
raw_criteria = (
target_signal.rejection.get(TestConst.CRITERIA_NAMES_FIELD)
if isinstance(target_signal.rejection, dict)
else None
)
criteria = RejectionCriteria(raw_criteria) if raw_criteria is not None else None
StepCheck(
f"Проверка rejection.criteriaNames для {sensor.description} (id={sensor.id})",
TestConst.CRITERIA_NAMES_FIELD,
soft_failures,
).actual(criteria).expected(rejection_case.expected_criteria_names).equal_to()
