https://pastein.ru/t/Tvh
Загрузка данных
"""
Тесты отбраковки сигналов с датчиков давления и расходомеров.
Параметризация:
SUITE_PARAMS генерирует (config, rejection_case, case_number) для каждого RejectionTestCase
из каждого IsRejectedConfig.
Запуск:
pytest tests/test_is_rejected_regress.py --suites=is_rejected_regress
"""
from datetime import datetime
from typing import Any, List, Optional
import allure
import pytest
from clients.websocket_client import WebSocketClient
from test_config.datasets import ALL_IS_REJECTED_CONFIGS
from test_config.models_for_tests import CaseMarkers, IsRejectedConfig, RejectionTestCase
from test_scenarios import rejected_scenarios as scenarios
# ===== ГЕНЕРАЦИЯ ПАРАМЕТРОВ =====
def _get_suite_markers(config: IsRejectedConfig) -> 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_rejection_params() -> List[Any]:
"""
Генерирует параметры для тестов отбраковки.
Один параметр на каждый (config, rejection_case, case_number).
"""
params = []
for config in ALL_IS_REJECTED_CONFIGS:
for rejection_case in config.rejection_cases:
param_id = f"{config.suite_name}_{rejection_case.name}"
params.append(
pytest.param(
config,
rejection_case,
id=param_id,
marks=_get_suite_markers(config),
)
)
return params
# ===== ПАРАМЕТРЫ ДЛЯ ТЕСТОВ =====
REJECTION_PARAMS: List[Any] = _generate_rejection_params()
# ===== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =====
def _apply_allure_markers(
test_config: Optional[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)
# ===== ТЕСТЫ ОТБРАКОВКИ =====
@pytest.mark.parametrize("config, rejection_case", REJECTION_PARAMS)
class TestIsRejectedScenarios:
"""
Тесты отбраковки сигналов.
Для каждого RejectionTestCase из конфига запускается 4 теста по одному на подписку.
"""
@pytest.mark.asyncio
async def test_rejection_input_signals(
self,
ws_client: WebSocketClient,
config: IsRejectedConfig,
rejection_case: RejectionTestCase,
) -> None:
"""[InputSignals] Проверка отбраковки во входных сигналах по подписке SubscribeInputSignalsRequest"""
sensor = rejection_case.sensor
tag = "InputSignals"
title = (
f"[{tag}] Проверка: {rejection_case.expected_event} для {sensor.description} (id={sensor.id}) "
f"({rejection_case.name}). ЭФ: Входные сигналы"
)
_apply_allure_markers(
rejection_case.rejection_input_signals_test,
tag,
title,
(
f"Проверка отбраковки сигнала {sensor.description} (id={sensor.id}), "
f"на наборе данных {config.suite_name},\n"
f"на технологическом участке {config.technological_unit.description}\n"
f"Время проведения проверки: {rejection_case.rejection_input_signals_test.offset} мин.\n"
f"Тип отбраковки: {rejection_case.name}\n"
"Подписка: SubscribeInputSignalsRequest"
),
)
await scenarios.rejection_input_signals(ws_client, config, rejection_case)
@pytest.mark.asyncio
async def test_rejection_journal(
self,
ws_client: WebSocketClient,
config: IsRejectedConfig,
rejection_case: RejectionTestCase,
imitator_start_time: datetime,
) -> None:
"""[MessagesInfo] Проверка записи об отбраковке в журнале"""
sensor = rejection_case.sensor
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка: {rejection_case.expected_event} для {sensor.description} (id={sensor.id}) "
f"({rejection_case.name}) в журнале. ЭФ: Журнал"
)
_apply_allure_markers(
rejection_case.rejection_journal_test,
tag,
title,
(
f"Проверка записи в журнале об отбраковке сигнала {sensor.description} (id={sensor.id}), "
f"на наборе данных {config.suite_name},\n"
f"на технологическом участке {config.technological_unit.description}\n"
f"Время проведения проверки: {rejection_case.rejection_journal_test.offset} мин.\n"
f"Тип отбраковки: {rejection_case.name}\n"
"Синхронный запрос типа: GetMessagesRequest с фильтром messageTypes=REJECTION"
),
)
await scenarios.rejection_journal(ws_client, config, rejection_case, imitator_start_time)
@pytest.mark.asyncio
async def test_rejection_main_page(
self,
ws_client: WebSocketClient,
config: IsRejectedConfig,
rejection_case: RejectionTestCase,
) -> None:
"""[MainPageSignalsInfo] Проверка счетчика отбраковки на состоянии МТ"""
tag = "MainPageSignalsInfo"
title = (
f"[{tag}] Проверка: {rejection_case.expected_event} - "
f"счетчик отбраковки на состоянии МТ показывает отбраковку"
f"({rejection_case.name}). ЭФ: Состояние МТ"
)
_apply_allure_markers(
rejection_case.rejection_main_page_test,
tag,
title,
(
f"Проверка количества отбракованных сигналов > 0 при отбраковке {rejection_case.name}, "
f"на наборе данных {config.suite_name},\n"
f"на технологическом участке {config.technological_unit.description}\n"
f"Время проведения проверки: {rejection_case.rejection_main_page_test.offset} мин.\n"
"Подписка: subscribeMainPageSignalsInfoRequest"
),
)
await scenarios.rejection_main_page(ws_client, config)
@pytest.mark.asyncio
async def test_rejection_scheme_signals_state(
self,
ws_client: WebSocketClient,
config: IsRejectedConfig,
rejection_case: RejectionTestCase,
) -> None:
"""[SchemeSignalsState] Проверка отбраковки сигнала на схеме по подписке SubscribeSchemeSignalsStateRequest"""
sensor = rejection_case.sensor
tag = "SchemeSignalsState"
title = (
f"[{tag}] Проверка: {rejection_case.expected_event} для {sensor.description} (id={sensor.id}) "
f"({rejection_case.name}). ЭФ: Схема"
)
_apply_allure_markers(
rejection_case.rejection_scheme_signals_state_test,
tag,
title,
(
f"Проверка отбраковки сигнала {sensor.description} (id={sensor.id}), "
f"на наборе данных {config.suite_name},\n"
f"на технологическом участке {config.technological_unit.description}\n"
f"Время проведения проверки: {rejection_case.rejection_scheme_signals_state_test.offset} мин.\n"
f"Тип отбраковки: {rejection_case.name}\n"
f"Ожидаемый criteriaNames: {rejection_case.expected_criteria_names}\n"
"Подписка: SubscribeSchemeSignalsStateRequest"
),
)
await scenarios.rejection_scheme_signals_state(ws_client, config, rejection_case)
"""
Сценарии тестов - функции-обёртки без 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=True для указанного датчика.
"""
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"Проверка отбраковки датчика {sensor.description} (id={sensor.id})", "isRejected", soft_failures
).actual(target_signal.isRejected).expected(True).equal_to()
if 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
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=rejection_case.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}' и событием {rejection_case.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 rejection_case.expected_event:
StepCheck("Проверка event", "event", soft_failures).actual(
(target_msg.event.rstrip() or "").strip()
).expected(rejection_case.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(True).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_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()
"""
Конфигурация тестового набора is_rejected_regress
Особенности набора:
- Проверка отбраковки сигналов с датчиков давления и расходомеров
- Типы отбраковки: empty, quality, VTOR, nearbySensors,
diagnosticInfo, constantSignal, range, discharge
Слои данных:
- emptyFilterSettings: 0-111 с (расход), 0-174 с (давление)
- qualityFilterSettings: 234-294 с (расход), 294-354 с (давление)
- vtorFilterSettings: 354-414 с (расход), 414-474 с (давление)
- nearbySensorsFilterSettings: 474--534 с (давление Pin+Pout)
- diagnosticInfoFilterSettings: 534-594 с (расход)
- constantSignalFilter: 655-713 с (расход), 717-777 с (давление)
- rangeFilterSettings upper: 779-839 с (расход), 1015-1075 с (давление)
- rangeFilterSettings lower: 899-957 с (расход), 1133-1196 cс (давление)
- dischargeFilterSettings: ~1194 с (расход), ~1270 с (давление)
"""
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 = "Значение давления"
# Из-за расхождений в данных и результатах работы бэка сдвинул все отбраковки на 1 минуту вперед по времени
# ===== Конфигурация набора =====
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, # 287 - снята
rejection_input_signals_test=CaseMarkers(test_case_id="189", offset=3),
# Проверка отбраковки датчика AK.CHTN.NPS_KRIM_P.UZR_1.Vmom (id=30157) Ожидаемый результат:
# isRejected = True Фактический результат: isRejected = False
rejection_main_page_test=CaseMarkers(test_case_id="189", offset=3), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="189", offset=3),
# Проверка isRejected для AK.CHTN.NPS_KRIM_P.UZR_1.Vmom (id=30157) Ожидаемый результат:
# isRejected = True Фактический результат: isRejected = False
),
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, # 581
rejection_input_signals_test=CaseMarkers(test_case_id="190", offset=4), # passed
rejection_main_page_test=CaseMarkers(test_case_id="190", offset=6), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="190", offset=5), # passed
),
# ===== 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, # 668
time_range_end_s=840, # 910
rejection_input_signals_test=CaseMarkers(test_case_id="191", offset=12),
# Проверка отбраковки датчика AK.CHTN.NPS_KRIM_P.UZR_1.Vmom (id=30157) Ожидаемый результат:
# isRejected = True Фактический результат: isRejected = False
rejection_journal_test=CaseMarkers(test_case_id="191", offset=15), # passed
rejection_main_page_test=CaseMarkers(test_case_id="191", offset=14), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="191", offset=13),
# Проверка isRejected для AK.CHTN.NPS_KRIM_P.UZR_1.Vmom (id=30157) Ожидаемый результат:
# isRejected = True Фактический результат: isRejected = False
),
RejectionTestCase(
name="quality_pressure",
sensor=PRESSURE_VELKRIM,
expected_event="Отбраковка по качеству",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.QUALITY,
time_range_start_s=900, # 941 сек
time_range_end_s=1140, # 1181 сек
rejection_input_signals_test=CaseMarkers(test_case_id="", offset=17),
# Фактический результат: criteriaNames = emptyRejection (4)] Expected <emptyRejection (4)>
# to be equal to <qualityRejection (1)>, but was not.
rejection_journal_test=CaseMarkers(test_case_id="", offset=20), # passed
rejection_main_page_test=CaseMarkers(test_case_id="", offset=19), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="", offset=18),
# Фактический результат: criteriaNames = emptyRejection (4)] Expected <emptyRejection (4)>
# to be equal to <qualityRejection (1)>, but was not.
),
# ===== 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, # 1231 sec
time_range_end_s=1440, # 1481 sec
rejection_input_signals_test=CaseMarkers(test_case_id="", offset=22), # fail
# [1, {}, None, 'InputSignalsContent',
# [{'replyStatus': 200, 'replyErrors': None, 'replyContent': {'tuId': 3, 'inputSignals': []}}], []]
rejection_journal_test=CaseMarkers(test_case_id="", offset=25), # fail - пришло на 6 8 сек позже диапазона
rejection_main_page_test=CaseMarkers(test_case_id="", offset=24), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="", offset=23), # fail
# Failed: Сигнал с id=30145 (AK.CHTN.NPS_TIH_5.UZR_1.Vmom) не найден среди 2363 полученных сигналов
),
RejectionTestCase(
name="vtor_pressure",
sensor=PRESSURE_KP7,
expected_event="Отбраковка по ВТОР сигналу",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.VTOR,
time_range_start_s=1500, # 1541 sec
time_range_end_s=1740, # 1781 sec
rejection_input_signals_test=CaseMarkers(test_case_id="", offset=27),
# Фактический результат: criteriaNames = emptyRejection (4)] Expected <emptyRejection (4)>
# to be equal to <VTORRejection (128)>
rejection_journal_test=CaseMarkers(test_case_id="", offset=30), # должно пофикситься
rejection_main_page_test=CaseMarkers(test_case_id="", offset=29), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="", offset=28),
# Фактический результат: criteriaNames = emptyRejection (4)] Expected <emptyRejection (4)>
# to be equal to <VTORRejection (128)>
),
# ===== 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, # 1870 sec
time_range_end_s=2040, # 2083 sec
rejection_input_signals_test=CaseMarkers(test_case_id="192", offset=32),
# Проверка rejection.criteriaNames для AK.CHTN.LU_TIHVEL.KP_8.SW_8-3.Pin (id=31439)
# Ожидаемый результат: criteriaNames = nearbyRejection (256) Фактический результат:
# criteriaNames = emptyRejection (4)
rejection_journal_test=CaseMarkers(test_case_id="192", offset=35), # passed
rejection_main_page_test=CaseMarkers(test_case_id="192", offset=34), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="192", offset=33),
# Проверка rejection.criteriaNames для AK.CHTN.LU_TIHVEL.KP_8.SW_8-3.Pin (id=31439)
# Ожидаемый результат: criteriaNames = nearbyRejection (256) Фактический результат:
# criteriaNames = emptyRejection (4)
),
RejectionTestCase(
name="nearby_pressure_pout", # ставлю время 4 минуты раньше чем в описании данных
sensor=PRESSURE_KP8_POUT,
expected_event="Отбраковка по разнице показаний СИ давления на КП",
expected_signal_name=SIGNAL_PRESSURE,
expected_criteria_names=RejectionCriteria.NEARBY,
time_range_start_s=2100, # first 1870 sec | second 2770 sec
time_range_end_s=2340, # first 2083 sec | second 3084 sec
rejection_input_signals_test=CaseMarkers(test_case_id="193", offset=32), # Кривой временной диапазон
rejection_journal_test=CaseMarkers(
test_case_id="193", offset=35
), # failed должно пофиксится из-за end по datetime
rejection_main_page_test=CaseMarkers(test_case_id="193", offset=34), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="193", offset=33), # Кривой временной диапазон
),
# ===== 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, # 2153 сек
time_range_end_s=2340, # 3042 сек
rejection_input_signals_test=CaseMarkers(test_case_id="194", offset=37),
# [1, {}, None, 'InputSignalsContent', [{'replyStatus': 200, 'replyErrors': None, 'replyContent':
# {'tuId': 3, 'inputSignals': []}}], []]
rejection_journal_test=CaseMarkers(test_case_id="194", offset=40), # тут заведена бага LDS-12394
rejection_main_page_test=CaseMarkers(test_case_id="194", offset=39), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="194", offset=38),
# Failed: Сигнал с id=30145 (AK.CHTN.NPS_TIH_5.UZR_1.Vmom) не найден среди 2363 полученных сигналов
),
# ===== constantSignalFilter =====
RejectionTestCase(
name="constant_signal_flow", # ставлю время 3 минуты позже чем в описании данных
sensor=FLOW_TIH,
expected_event="Отбраковка по постоянному сигналу",
expected_signal_name=SIGNAL_FLOW,
expected_criteria_names=RejectionCriteria.CONSTANT_SIGNAL,
time_range_start_s=2400, # 2581 сек
time_range_end_s=2640, # 2682 сек
rejection_input_signals_test=CaseMarkers(test_case_id="", offset=44),
# [1, {}, None, 'InputSignalsContent', [{'replyStatus': 200, 'replyErrors': None, 'replyContent':
# {'tuId': 3, 'inputSignals': []}}], []]
rejection_journal_test=CaseMarkers(test_case_id="", offset=45), # pass
rejection_main_page_test=CaseMarkers(test_case_id="", offset=44), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="", offset=44),
# Failed: Сигнал с id=30145 (AK.CHTN.NPS_TIH_5.UZR_1.Vmom) не найден среди 2363 полученных сигналов
),
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, # 2770=nearby 2950=constant_signal
time_range_end_s=2940, # 2984=nearby=constant_signal
rejection_input_signals_test=CaseMarkers(test_case_id="", offset=47),
# также отбраковка пришла очень поздно и не попала в диапазон журнала
rejection_journal_test=CaseMarkers(
test_case_id="", offset=50
), # Отбраковка одновременно приосходит и по nearby и по constant_signal,
# также отбраковка пришла очень поздно и не попала в диапазон журнала
rejection_main_page_test=CaseMarkers(test_case_id="", offset=49), # passed
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="", offset=48),
# также отбраковка пришла очень поздно и не попала в диапазон журнала
),
# ===== 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, # 3041
time_range_end_s=3240, # 3284
rejection_input_signals_test=CaseMarkers(test_case_id="195", offset=52),
# [1, {}, None, 'InputSignalsContent', [{'replyStatus': 200, 'replyErrors': None, 'replyContent':
# {'tuId': 3, 'inputSignals': []}}], []]
rejection_journal_test=CaseMarkers(test_case_id="195", offset=55), # pass
rejection_main_page_test=CaseMarkers(test_case_id="195", offset=54), # pass
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="195", offset=53),
# Failed: Сигнал с id=30145 (AK.CHTN.NPS_TIH_5.UZR_1.Vmom) не найден среди 2363 полученных сигналов
),
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, # 3359
time_range_end_s=3540, # 3581
rejection_input_signals_test=CaseMarkers(test_case_id="197", offset=57),
# [1, {}, None, 'InputSignalsContent', [{'replyStatus': 200, 'replyErrors': None, 'replyContent':
# {'tuId': 3, 'inputSignals': []}}], []]
rejection_journal_test=CaseMarkers(test_case_id="197", offset=60), # pass
rejection_main_page_test=CaseMarkers(test_case_id="197", offset=59), # pass
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="197", offset=58),
# Failed: Сигнал с id=30145 (AK.CHTN.NPS_TIH_5.UZR_1.Vmom) не найден среди 2363 полученных сигналов
),
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, # по факту 3641 сек
time_range_end_s=3840, # по факту 3883 сек
rejection_input_signals_test=CaseMarkers(test_case_id="196", offset=62), # fail
# Фактический результат: criteriaNames = emptyRejection (4)] Expected <emptyRejection (4)>
# to be equal to <rangeRejection (2)>
rejection_journal_test=CaseMarkers(test_case_id="196", offset=65), # pass
rejection_main_page_test=CaseMarkers(test_case_id="196", offset=64), # pass
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="196", offset=63), # fail
# Фактический результат: criteriaNames = emptyRejection (4)] Expected <emptyRejection (4)>
# to be equal to <rangeRejection (2)>
),
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, # 3943
time_range_end_s=4140, # 4183
rejection_input_signals_test=CaseMarkers(test_case_id="198", offset=67), # fail
# Фактический результат: criteriaNames = emptyRejection (4)] Expected <emptyRejection (4)>
# to be equal to <rangeRejection (2)>
rejection_journal_test=CaseMarkers(test_case_id="198", offset=70), # pass
rejection_main_page_test=CaseMarkers(test_case_id="198", offset=69), # pass
rejection_scheme_signals_state_test=CaseMarkers(test_case_id="198", offset=68), # fail
# Фактический результат: criteriaNames = emptyRejection (4)] Expected <emptyRejection (4)>
# to be equal to <rangeRejection (2)>
),
],
)
