https://pastein.ru/t/D20
Загрузка данных
lds_status_regress_bik.py lfnfctn
"""
Конфигурация тестового набора Lds_status_regress_in_flow
Особенности набора:
- В течении
- Проверка режимов работы СОУ
- Проверка группы причин режимов работы СОУ, связанных с БИК
"""
from constants.enums import TU, AdminTU, LdsStatus, LdsStatusDegradation, LdsStatusInitialization, MeasureConversionRule
from constants.test_constants import BaseTN3Constants as TestConst
from test_config.models_for_tests import CaseData, CaseMarkers, LDSStatusConfig
# ===== Константы набора =====
SUITE_NAME = "Lds_status_regress_bik"
SUITE_DATA_ID = 18
ARCHIVE_NAME = f"{SUITE_NAME}.tar.gz"
DIAGNOSTIC_AREA_2_PIPE_ID = 1463
DIAGNOSTIC_AREA_3_PIPE_ID = 474
DIAGNOSTIC_AREA_5_PIPE_ID = 1474
# Технологический участок
TECHNOLOGICAL_UNIT = TU.TIKHORETSK_NOVOROSSIYSK_3
# Название МН
MAIN_PIPELINE = "МН Тихорецк-Новороссийск-3"
# ===== Конфигурация набора =====
LDS_STATUS_BIK_CONFIG = LDSStatusConfig(
# ----- Метаданные -----
ost_name=TestConst.CHTN_OST_NAME,
suite_name=SUITE_NAME,
suite_data_id=SUITE_DATA_ID,
archive_name=ARCHIVE_NAME,
technological_unit=TECHNOLOGICAL_UNIT,
measure_conversion_rules=MeasureConversionRule.KG_CM_MEASURE,
# ===== LDS Configurator =====
use_lds_configurator=True,
admin_tu=AdminTU.TIKHORETSK_NOVOROSSIYSK_3_AUTOTEST_DATA_ABSENCE_FALSE,
main_pipeline=MAIN_PIPELINE,
init_cold_start_test_data=CaseData(
expected_result=(LdsStatus.INITIALIZATION, LdsStatusInitialization.COLD_START_OF_SERVERS),
),
deg_rejection_temperature_sensor_on_du_2_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_2_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.REJECTION_TEMPERATURE_SENSOR,
),
),
deg_rejection_temperature_sensor_on_du_3_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_3_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.REJECTION_TEMPERATURE_SENSOR,
),
),
deg_rejection_temperature_sensor_on_du_5_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_5_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.REJECTION_TEMPERATURE_SENSOR,
),
),
deg_rejection_density_and_viscosity_on_du_2_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_2_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.REJECTION_DENSITY_SENSOR,
LdsStatusDegradation.REJECTION_VISCOSITY_SENSOR,
),
),
deg_rejection_density_and_viscosity_on_du_3_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_3_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.REJECTION_DENSITY_SENSOR,
LdsStatusDegradation.REJECTION_VISCOSITY_SENSOR,
),
),
deg_rejection_density_and_viscosity_on_du_5_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_5_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.REJECTION_DENSITY_SENSOR,
LdsStatusDegradation.REJECTION_VISCOSITY_SENSOR,
),
),
# ===== ТЕСТЫ =====
lds_status_basic_info_test=CaseMarkers(test_case_id="1", offset=5),
init_cold_start_test=CaseMarkers(test_case_id="18", offset=7),
init_cold_start_in_journal_test=CaseMarkers(test_case_id="228", offset=7),
deg_rejection_temperature_sensor_on_du_2_test=CaseMarkers(test_case_id="179", offset=25),
deg_rejection_density_and_viscosity_on_du_2_test=CaseMarkers(test_case_id="181", offset=35),
deg_rejection_temperature_sensor_on_du_3_test=CaseMarkers(test_case_id="179", offset=58),
deg_rejection_density_and_viscosity_on_du_3_test=CaseMarkers(test_case_id="181", offset=68),
)
lds_status_regress_in_flow
"""
Конфигурация тестового набора Lds_status_regress_in_flow
Особенности набора:
- В течении
- Проверка режимов работы СОУ
- Проверка причин режимов работы СОУ
"""
from constants.enums import (
TU,
AdminTU,
LdsStatus,
LdsStatusDegradation,
LdsStatusFaulty,
LdsStatusInitialization,
MeasureConversionRule,
StationaryStatus,
UnStationaryReason,
)
from constants.test_constants import BaseTN3Constants as TestConst
from test_config.models_for_tests import CaseData, CaseMarkers, LDSStatusConfig
# ===== Константы набора =====
SUITE_NAME = "Lds_status_regress_in_flow"
SUITE_DATA_ID = 17
ARCHIVE_NAME = f"{SUITE_NAME}.tar.gz"
DIAGNOSTIC_AREA_2_PIPE_ID = 1463
DIAGNOSTIC_AREA_3_PIPE_ID = 474
DIAGNOSTIC_AREA_5_PIPE_ID = 1478
PIG_TRAP_ID = 344
# Технологический участок
TECHNOLOGICAL_UNIT = TU.TIKHORETSK_NOVOROSSIYSK_3
# Название МН
MAIN_PIPELINE = "МН Тихорецк-Новороссийск-3"
# ===== Конфигурация набора =====
LDS_STATUS_INFLOW_CONFIG = LDSStatusConfig(
# ----- Метаданные -----
ost_name=TestConst.CHTN_OST_NAME,
suite_name=SUITE_NAME,
suite_data_id=SUITE_DATA_ID,
archive_name=ARCHIVE_NAME,
main_pipeline=MAIN_PIPELINE,
technological_unit=TECHNOLOGICAL_UNIT,
measure_conversion_rules=MeasureConversionRule.KG_CM_MEASURE,
# ===== LDS Configurator =====
use_lds_configurator=True,
admin_tu=AdminTU.TIKHORETSK_NOVOROSSIYSK_3_AUTOTEST_DATA_ABSENCE_FALSE,
init_cold_start_test_data=CaseData(
expected_result=(LdsStatus.INITIALIZATION, LdsStatusInitialization.COLD_START_OF_SERVERS),
),
serviceable_all_in_journal_test_data=CaseData(
expected_result=(LdsStatus.INITIALIZATION, LdsStatusInitialization.COLD_START_OF_SERVERS),
),
serviceable_all_test_data=CaseData(
params={"pipe_ids": [DIAGNOSTIC_AREA_2_PIPE_ID, DIAGNOSTIC_AREA_3_PIPE_ID]},
expected_result=LdsStatus.SERVICEABLE,
),
deg_additive_injectors_operation_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_3_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.ADDITIVE_INJECTORS_OPERATION,
),
),
deg_exceeding_distance_between_pressure_sensors_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_3_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.EXCEEDING_DISTANCE_BETWEEN_SERVICEABLE_PRESSURE_SENSORS,
),
),
deg_absence_min_pressure_sensors_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_5_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.ABSENCE_MIN_PRESSURE_SENSORS_REQUIRED_NUMBER,
),
),
deg_faulty_pressure_sensors_at_pump_station_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_2_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.FAULTY_PRESSURE_SENSORS_AT_PUMP_STATION_NODES,
),
),
deg_gravity_section_pumping_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_5_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.GRAVITY_SECTION_IN_PUMPING_MODE,
),
),
deg_pig_sensor_passage_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_2_PIPE_ID, "pig_trap_id": PIG_TRAP_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.PIG_SENSOR_PASSAGE,
),
),
deg_starting_pumping_out_pumps_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_2_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.STARTING_PUMPING_OUT_PUMPS,
StationaryStatus.UNSTATIONARY,
UnStationaryReason.CHANGING_WORKING_OF_PUMPING_OUT_PUMPS,
),
),
faulty_absence_min_flow_meters_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_5_PIPE_ID},
expected_result=(
LdsStatus.FAULTY,
LdsStatusFaulty.ABSENCE_MIN_FLOW_METERS_REQUIRED_NUMBER,
),
),
deg_exceeding_distance_between_flow_meters_test_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_3_PIPE_ID},
expected_result=(
LdsStatus.DEGRADATION,
LdsStatusDegradation.EXCEEDING_DISTANCE_BETWEEN_FLOW_METERS,
),
),
# ===== ТЕСТЫ =====
lds_status_basic_info_test=CaseMarkers(test_case_id="1", offset=5),
init_cold_start_test=CaseMarkers(test_case_id="17", offset=7),
init_cold_start_in_journal_test=CaseMarkers(test_case_id="228", offset=7),
serviceable_after_cold_start_test=CaseMarkers(test_case_id="186", offset=22),
deg_exceeding_distance_between_pressure_sensors_test=CaseMarkers(test_case_id="164", offset=32),
deg_gravity_section_pumping_test=CaseMarkers(test_case_id="167", offset=34),
deg_absence_min_pressure_sensors_test=CaseMarkers(test_case_id="166", offset=38.5),
serviceable_after_deg_absence_min_pressure_sensors_test=CaseMarkers(test_case_id="186", offset=43),
deg_starting_pumping_out_pumps_test=CaseMarkers(test_case_id="174", offset=46),
serviceable_after_deg_starting_pumping_out_pumps_test=CaseMarkers(test_case_id="186", offset=52.5),
faulty_absence_min_flow_meters_test=CaseMarkers(test_case_id="170", offset=60),
faulty_absence_min_flow_meters_continuous_test=CaseMarkers(test_case_id="170", offset=73),
serviceable_after_faulty_test=CaseMarkers(test_case_id="186", offset=76),
deg_exceeding_distance_between_flow_meters_test=CaseMarkers(test_case_id="169", offset=107),
# serviceable_after_deg_exceeding_distance_between_flow_meters_test=CaseMarkers(test_case_id="186", offset=114.5),
# TODO разобраться в LDS-14444
deg_faulty_pressure_sensors_at_pump_station_test=CaseMarkers(test_case_id="184", offset=122),
serviceable_after_deg_faulty_pressure_sensors_at_pump_test=CaseMarkers(test_case_id="186", offset=126),
deg_additive_injectors_operation_test=CaseMarkers(test_case_id="188", offset=132),
deg_pig_sensor_passage_test=CaseMarkers(test_case_id="165", offset=140),
)
models_for_tests
"""
Датаклассы для конфигурации тестовых наборов.
Архитектура:
- TestSuiteConfig - главный конфиг набора, содержит всё для запуска тестов
- LeakTestConfig - конфиг утечки с параметрами и тест-кейсами
- TestCaseMarkers - маркеры для allure и pytest
Принцип: один файл конфига select_xx.py в папке datasets = один набор данных.
"""
from dataclasses import asdict, dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Optional
from constants.enums import (
TU,
AdminTU,
ConfirmationStatus,
LdsStatus,
MeasureConversionRule,
RejectionCriteria,
RejectionSensorTag,
ReservedType,
StationaryStatus,
)
from constants.test_constants import BaseTN3Constants
from models.export_reports_model import ReportDataExportedNotification
from models.get_exported_files_list_model import ExportedDataItem
from models.subscribe_main_page_signals_info_model import SignalsInfo
from models.upload_exported_file_model import DownloadExportedDataReply
from utils.helpers.report_xlsx_utils import LeakReportRow, ReportTitleInfo
@dataclass
class CaseMarkers:
"""
Маркеры тест-кейса для pytest и allure.
"""
test_case_id: str
offset: float
class SuiteTuIdentityMixin:
"""
Идентификация ТУ для WS-тестов и legacy-стенда.
Поля technological_unit, use_lds_configurator, admin_tu, resolved_tu_id
объявляются в BaseSuiteConfig.
"""
technological_unit: TU
use_lds_configurator: bool
admin_tu: Optional[AdminTU]
resolved_tu_id: Optional[int]
@property
def admin_tu_name(self) -> str:
"""tuName для поиска ТУ в Администрировании."""
if self.admin_tu is None:
return ""
return self.admin_tu.admin_name
@property
def legacy_tu_id(self) -> int:
"""Legacy id ТУ для имитатора, tags.txt и конфигурации стенда."""
return self.technological_unit.id
@property
def legacy_tu_name(self) -> str:
"""Legacy имя ТУ из enum TU."""
return self.technological_unit.description
@property
def tu_name(self) -> str:
"""Имя ТУ для WS-тестов и отчётов: admin при configurator, иначе legacy."""
if self.use_lds_configurator:
if self.admin_tu is None:
return ""
return self.admin_tu.admin_name
return self.technological_unit.description
@property
def tu_id(self) -> int:
"""ID ТУ для WS-тестов: resolved из Администрирования при configurator, иначе legacy."""
if self.use_lds_configurator:
if self.resolved_tu_id is None:
raise RuntimeError(
"resolved_tu_id не установлен - выполнить lds_configurator_admin_setup "
"перед операциями Администрирования"
)
return self.resolved_tu_id
return self.technological_unit.id
@property
def configurator_tu_id(self) -> int:
"""tuId из Администрирования после configurator setup (LaunchLds/StopLds)."""
if self.resolved_tu_id is None:
raise RuntimeError(
"resolved_tu_id не установлен - выполнить lds_configurator_admin_setup "
"перед операциями Администрирования"
)
return self.resolved_tu_id
@property
def infra_tu_id(self) -> int:
"""Алиас legacy_tu_id (обратная совместимость)."""
return self.technological_unit.id
@dataclass
class BaseSuiteConfig(SuiteTuIdentityMixin):
"""
Структура:
1. Метаданные набора (имя, id, архив)
2. Технологический участок (из enum TU)
"""
# ===== Метаданные набора =====
ost_name: str
suite_name: str
suite_data_id: int
archive_name: str = "" # Автоматически вычисляется из suite_name
# ===== Технологический участок =====
technological_unit: TU = TU.TIKHORETSK_NOVOROSSIYSK_3
# ===== LDS Configurator (Администрирование) =====
use_lds_configurator: bool = False
admin_tu: Optional[AdminTU] = None
resolved_tu_id: Optional[int] = None
# ===== Название Магистрального Нефтепровода =====
main_pipeline: Optional[str] = None
# ===== Правила конвертации единиц измерения давления на стенде =====
measure_conversion_rules: Optional[MeasureConversionRule] = None
# ===== Общие константы (можно переопределить) =====
allowed_distance_diff_meters: int = BaseTN3Constants.ALLOWED_DISTANCE_DIFF_METERS
precision: int = BaseTN3Constants.PRECISION
basic_message_timeout: float = BaseTN3Constants.BASIC_MESSAGE_TIMEOUT
mask_message_timeout: float = BaseTN3Constants.MASK_MESSAGE_TIMEOUT
mask_du_name: Optional[str] = None
main_pipe_line: Optional[str] = None
mask_du_event: Optional[str] = None
unmask_du_event: Optional[str] = None
def __post_init__(self) -> None:
self.validate_lds_configurator_config()
def validate_lds_configurator_config(self) -> None:
"""Проверяет согласованность полей LDS Configurator в датасете."""
if not self.use_lds_configurator:
return
if self.admin_tu is None:
raise ValueError(f"Набор '{self.suite_name}': admin_tu обязателен при use_lds_configurator=True")
if self.admin_tu.legacy_tu != self.technological_unit:
raise ValueError(
f"Набор '{self.suite_name}': admin_tu.legacy_tu ({self.admin_tu.legacy_tu}) "
f"не совпадает с technological_unit ({self.technological_unit})"
)
# ===== Свойства для удобства =====
@property
def has_multiple_leaks(self) -> bool:
return False
@dataclass
class CaseData:
"""
Данные тест-кейса.
"""
name: str = ""
params: Optional[Dict[str, Any]] = None
expected_result: Optional[Any] = None
description: str = ""
@dataclass
class DiagnosticAreaStatusConfig:
"""
Конфигурация ожидаемых статусов СОУ для диагностического участка.
Используется в тесте lds_status_during_leak.
"""
leak_diagnostic_area_name: str
leak_du_expected_lds_status: Any
neighbors_du_expected_lds_status: Any
leak_diagnostic_area_id: Optional[int] = None
@dataclass
class LeakTestConfig:
"""
Полная конфигурация утечки со всеми параметрами и тест-кейсами.
Все данные для тестов утечки:
- Параметры утечки (координата, объём)
- Временные интервалы
- Ожидаемые значения
- Маркеры тестов (AllLeaksInfo, TuLeaksInfo, и т.д.)
"""
# ===== Идентификаторы =====
control_site_id: Optional[int] = None
diagnostic_area_id: Optional[int] = None
diagnostic_area_name: Optional[str] = None
linear_part_id: Optional[int] = None
technological_object: Optional[str] = None
message_event_leak_completion: Optional[str] = None
# ===== Параметры утечки =====
coordinate_meters: float = None
volume_m3: float = None
max_pumping_m3: int = 2500 # Производительность(максимальная перекачка)
flow_rate_settings_threshold: Optional[float] = None # Порог объема дебаланса для текущего ДУ в текущем режиме
# ===== Временные интервалы (секунды) =====
leak_start_interval_seconds: int = BaseTN3Constants.LEAK_START_INTERVAL
allowed_time_diff_seconds: int = 0 # Допустимое время обнаружения
output_test_delay_seconds: int = BaseTN3Constants.OUTPUT_TEST_DELAY
# ===== Ожидаемые статусы =====
expected_lds_status: Any = LdsStatus.SERVICEABLE
# Режим СОУ в xlsx export_leaks_report (колонка 'Режим работы СОУ')
expected_lds_status_in_leaks_report: Optional[int] = None
expected_report_stationary_status: int = StationaryStatus.STATIONARY.value
expected_stationary_status: Any = StationaryStatus.STATIONARY
expected_algorithm_type: Any = ReservedType.STATIONARY_FLOW
expected_leak_status: Any = ConfirmationStatus.CONFIRMED
expected_complete_leak_status: Any = ConfirmationStatus.CONFIRMED_AND_LEAK_CLOSED
# ===== Конфигурация статусов СОУ во время утечки =====
lds_status_during_leak_config: Optional[DiagnosticAreaStatusConfig] = None
# ===== Данные тест-кейсов =====
lds_status_after_confirming_leak_data: Optional[CaseData] = None
lds_status_after_completed_leak_data: Optional[CaseData] = None
# ===== Тест-кейсы для этой утечки =====
balance_algorithm_leak_waiting_test: Optional[CaseMarkers] = None
balance_algorithm_leak_detected_test: Optional[CaseMarkers] = None
leaks_content_test: Optional[CaseMarkers] = None
all_leaks_info_test: Optional[CaseMarkers] = None
all_leaks_is_empty_test: Optional[CaseMarkers] = None
tu_leaks_info_test: Optional[CaseMarkers] = None
leak_info_in_journal: Optional[CaseMarkers] = None
possible_leak_in_journal_test: Optional[CaseMarkers] = None
acknowledge_leak_test: Optional[CaseMarkers] = None
acknowledge_leak_in_journal_test: Optional[CaseMarkers] = None
output_signals_test: Optional[CaseMarkers] = None
lds_status_during_leak_test: Optional[CaseMarkers] = None
lds_status_after_confirming_leak_test: Optional[CaseMarkers] = None
lds_status_completed_leak_test: Optional[CaseMarkers] = None
the_leak_is_complete_on_kg_test: Optional[CaseMarkers] = None
leak_is_complete_in_output_signals_test: Optional[CaseMarkers] = None
leak_is_complete_on_main_page_test: Optional[CaseMarkers] = None
leak_is_confirm_on_main_page_test: Optional[CaseMarkers] = None
complete_tu_leaks_info_content_test: Optional[CaseMarkers] = None
completed_leak_info_in_journal_test: Optional[CaseMarkers] = None
balance_algorithm_leak_completed_test: Optional[CaseMarkers] = None
export_leaks_report_test: Optional[CaseMarkers] = None
@property
def leak_diagnostic_area_id(self) -> Optional[int]:
"""ID диагностического участка с утечкой из lds_status_during_leak_config"""
if self.lds_status_during_leak_config is not None:
return self.lds_status_during_leak_config.leak_diagnostic_area_id
return None
@property
def allowed_volume_m3(self) -> float:
"""Допустимая погрешность объёма"""
return self.volume_m3 * BaseTN3Constants.ALLOWED_VOLUME_DIFF
@property
def leak_rate_percentages(self) -> float:
"""Интенсивность утечки в процентах"""
return round((self.volume_m3 / self.max_pumping_m3) * 100, 2)
@property
def allowed_time_diff_minutes(self) -> float:
"""Допустимое время обнаружения утечки в минутах"""
return round(self.allowed_time_diff_seconds / 60, 2)
@property
def output_allowed_time_diff_seconds(self) -> int:
"""Допустимое время для теста выходных сигналов"""
return self.allowed_time_diff_seconds + self.output_test_delay_seconds
@dataclass
class SmokeSuiteConfig(BaseSuiteConfig):
"""
Полная конфигурация тестового набора.
Один конфиг = один набор данных = один файл в test_config/datasets/
Структура:
1. Базовые тесты с маркерами
2. Конфигурации утечек (LeakTestConfig)
"""
# ===== Ожидаемый статусы для main_page_info =====
expected_stationary_status: Any = StationaryStatus.STATIONARY
expected_main_page_signals: dict = field(default_factory=lambda: asdict(SignalsInfo()))
# ===== Ожидаемые переменные при маскировании ДУ =====
mask_reason: Optional[str] = None
unmask_reason: Optional[str] = None
mask_one_du: Optional[int] = None
not_mask_du: Optional[int] = None
linear_part_identifier_for_mask: Optional[int] = None
technological_section: Optional[str] = None
imitate_flowmeter_signal_test_data: Optional[CaseData] = None
imitate_pressure_sensor_signal_test_data: Optional[CaseData] = None
exp_mode_mt_message: Optional[CaseData] = None
# Режим МТ в xlsx export_leaks_report
expected_report_stationary_status: int = StationaryStatus.STATIONARY.value
# ===== Данные участков КП-КП =====
controlled_sites_with_segment: Optional[CaseData] = None
# дефолтные значения для датчиков маскирования
mask_signal_test_data: Optional[CaseData] = CaseData(
params={
"pressure_sensor_address": BaseTN3Constants.PRESSURE_SENSOR_ADDRESS,
"flowmeter_address": BaseTN3Constants.FLOWMETER_ADDRESS,
}
)
# ===== Базовые тесты =====
basic_info_test: Optional[CaseMarkers] = None
imitate_flowmeter_signal_test: Optional[CaseMarkers] = None
imitate_pressure_sensor_signal_test: Optional[CaseMarkers] = None
journal_info_test: Optional[CaseMarkers] = None
lds_status_initialization_test: Optional[CaseMarkers] = None
lds_status_init_in_journal_test: Optional[CaseMarkers] = None
main_page_info_test: Optional[CaseMarkers] = None
main_page_info_signals_test: Optional[CaseMarkers] = None
mask_signal_test: Optional[CaseMarkers] = None
mask_info_in_journal_test: Optional[CaseMarkers] = None
lds_status_initialization_out_test: Optional[CaseMarkers] = None
lds_status_init_out_in_journal_test: Optional[CaseMarkers] = None
mask_du_on_mini_scheme_test: Optional[CaseMarkers] = None
unmask_du_on_mini_scheme_test: Optional[CaseMarkers] = None
diagnostics_of_signals_after_initialization_test: Optional[CaseMarkers] = None
mode_mt_in_journal_test: Optional[CaseMarkers] = None
export_lds_status_report_test: Optional[CaseMarkers] = None
export_mt_mode_report_test: Optional[CaseMarkers] = None
# ===== Конфигурации утечек =====
# Для наборов с одной утечкой
leak: Optional[LeakTestConfig] = None
# Для наборов с несколькими утечками (select_19_20)
leaks: list[LeakTestConfig] = field(default_factory=list)
# Участки в xlsx-отчёте о режиме работы СОУ (export_lds_status_report)
lds_status_report_section_names: list[str] = field(default_factory=list)
# ===== Дополнительные тесты для двух утечек =====
main_page_info_unstationary_test: Optional[CaseMarkers] = None
def get_leak(self, index: int = 0) -> Optional[LeakTestConfig]:
"""Получить конфигурацию утечки по индексу"""
if self.leak and index == 0:
return self.leak
if self.leaks and index < len(self.leaks):
return self.leaks[index]
return None
@property
def has_multiple_leaks(self) -> bool:
"""Проверить, есть ли несколько утечек"""
return len(self.leaks) > 1
@property
def allowed_volume_diff(self) -> float:
"""Относительная погрешность по объёму"""
return BaseTN3Constants.ALLOWED_VOLUME_DIFF
@dataclass
class LDSStatusConfig(BaseSuiteConfig):
"""
Полная конфигурация тестового набора.
Один конфиг = один набор данных = один файл в test_config/datasets/
Структура:
1. Данные для тестов(параметры и ожидаемый результат)
2. Тесты с маркерами
"""
# ===== Данные для тестов =====
init_accumulation_data_test_data: Optional[CaseData] = None
init_accumulation_data_in_journal_test_data: Optional[CaseData] = None
init_cold_start_test_data: Optional[CaseData] = None
init_exiting_faulty_test_data: Optional[CaseData] = None
init_switching_shut_off_test_data: Optional[CaseData] = None
init_switching_shut_off_in_journal_test_data: Optional[CaseData] = None
serviceable_all_test_data: Optional[CaseData] = None
serviceable_all_in_journal_test_data: Optional[CaseData] = None
serviceable_after_switching_shut_off_test_data: Optional[CaseData] = None
serviceable_after_switching_shut_off_in_journal_test_data: Optional[CaseData] = None
serviceable_after_deg_faulty_pressure_sensors_at_pump_test_data: Optional[CaseData] = None
serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test_data: Optional[CaseData] = None
deg_faulty_pressure_sensors_at_pump_station_test_data: Optional[CaseData] = None
deg_faulty_pressure_sensors_at_pump_station_in_journal_test_data: Optional[CaseData] = None
deg_additive_injectors_operation_test_data: Optional[CaseData] = None
deg_absence_min_pressure_sensors_test_data: Optional[CaseData] = None
deg_exceeding_distance_between_pressure_sensors_test_data: Optional[CaseData] = None
deg_exceeding_distance_between_pressure_sensors_in_journal_test_data: Optional[CaseData] = None
deg_gravity_section_pumping_test_data: Optional[CaseData] = None
deg_gravity_section_pumping_in_stopping_test_data: Optional[CaseData] = None
deg_gravity_section_pumping_in_stopping_in_journal_test_data: Optional[CaseData] = None
deg_pig_sensor_passage_test_data: Optional[CaseData] = None
deg_starting_pumping_out_pumps_test_data: Optional[CaseData] = None
deg_exceeding_distance_between_flow_meters_test_data: Optional[CaseData] = None
deg_rejection_temperature_sensor_on_du_2_test_data: Optional[CaseData] = None
deg_rejection_temperature_sensor_on_du_3_test_data: Optional[CaseData] = None
deg_rejection_temperature_sensor_on_du_5_test_data: Optional[CaseData] = None
deg_rejection_density_and_viscosity_on_du_2_test_data: Optional[CaseData] = None
deg_rejection_density_and_viscosity_on_du_3_test_data: Optional[CaseData] = None
deg_rejection_density_and_viscosity_on_du_5_test_data: Optional[CaseData] = None
faulty_absence_min_flow_meters_test_data: Optional[CaseData] = None
faulty_absence_min_pressure_sensors_test_data: Optional[CaseData] = None
faulty_absence_min_pressure_sensors_in_journal_test_data: Optional[CaseData] = None
# ===== Тесты =====
lds_status_basic_info_test: Optional[CaseMarkers] = None
init_accumulation_data_test: Optional[CaseMarkers] = None
init_accumulation_data_in_journal_test: Optional[CaseMarkers] = None
init_cold_start_test: Optional[CaseMarkers] = None
init_cold_start_in_journal_test: Optional[CaseMarkers] = None
init_exiting_faulty_test: Optional[CaseMarkers] = None
init_switching_shut_off_test: Optional[CaseMarkers] = None
init_switching_shut_off_in_journal_test: Optional[CaseMarkers] = None
serviceable_after_cold_start_test: Optional[CaseMarkers] = None
serviceable_after_cold_start_in_journal_test: Optional[CaseMarkers] = None
serviceable_after_switching_shut_off_test: Optional[CaseMarkers] = None
serviceable_after_switching_shut_off_in_journal_test: Optional[CaseMarkers] = None
serviceable_after_deg_absence_min_pressure_sensors_test: Optional[CaseMarkers] = None
serviceable_after_deg_additive_injectors_operation_test: Optional[CaseMarkers] = None
serviceable_after_deg_starting_pumping_out_pumps_test: Optional[CaseMarkers] = None
serviceable_after_deg_faulty_pressure_sensors_at_pump_test: Optional[CaseMarkers] = None
serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test: Optional[CaseMarkers] = None
serviceable_after_deg_exceeding_distance_between_flow_meters_test: Optional[CaseMarkers] = None
serviceable_after_faulty_test: Optional[CaseMarkers] = None
deg_additive_injectors_operation_test: Optional[CaseMarkers] = None
deg_exceeding_distance_between_pressure_sensors_test: Optional[CaseMarkers] = None
deg_exceeding_distance_between_pressure_sensors_in_journal_test: Optional[CaseMarkers] = None
deg_absence_min_pressure_sensors_test: Optional[CaseMarkers] = None
deg_faulty_pressure_sensors_at_pump_station_test: Optional[CaseMarkers] = None
deg_faulty_pressure_sensors_at_pump_station_in_journal_test: Optional[CaseMarkers] = None
deg_gravity_section_pumping_test: Optional[CaseMarkers] = None
deg_gravity_section_pumping_in_stopping_test: Optional[CaseMarkers] = None
deg_gravity_section_pumping_in_stopping_in_journal_test: Optional[CaseMarkers] = None
deg_pig_sensor_passage_test: Optional[CaseMarkers] = None
deg_starting_pumping_out_pumps_test: Optional[CaseMarkers] = None
deg_exceeding_distance_between_flow_meters_test: Optional[CaseMarkers] = None
deg_rejection_temperature_sensor_on_du_2_test: Optional[CaseMarkers] = None
deg_rejection_temperature_sensor_on_du_3_test: Optional[CaseMarkers] = None
deg_rejection_temperature_sensor_on_du_5_test: Optional[CaseMarkers] = None
deg_rejection_density_and_viscosity_on_du_2_test: Optional[CaseMarkers] = None
deg_rejection_density_and_viscosity_on_du_3_test: Optional[CaseMarkers] = None
deg_rejection_density_and_viscosity_on_du_5_test: Optional[CaseMarkers] = None
faulty_absence_min_flow_meters_test: Optional[CaseMarkers] = None
faulty_absence_min_flow_meters_continuous_test: Optional[CaseMarkers] = None
faulty_absence_min_pressure_sensors_test: Optional[CaseMarkers] = None
faulty_absence_min_pressure_sensors_in_journal_test: Optional[CaseMarkers] = None
@dataclass
class StationaryStatusConfig(BaseSuiteConfig):
# ===== Данные для тестов =====
stationary_status_with_reason_test_data: Optional[CaseData] = None
stationary_status_journal_test_data: Optional[CaseData] = None
# ===== Тесты =====
stationary_status_basic_info_test: Optional[CaseMarkers] = None
stationary_status_journal_test: Optional[CaseMarkers] = None
stationary_status_common_scheme_test: Optional[CaseMarkers] = None
stationary_status_main_page_info_test: Optional[CaseMarkers] = None
@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 RejectionReportRow:
"""Разобранная строка отчёта об отбракованных входных данных."""
row_index: int
datetime_value: Optional[datetime] = None
object_value: str = ""
event_value: str = ""
value_text: str = ""
duration_seconds: int = 0
tag_value: str = ""
@dataclass
class IsRejectedConfig(BaseSuiteConfig):
"""
Конфигурация тестового набора отбраковки сигналов.
Структура:
1. Название МН
2. Список случаев отбраковки (RejectionTestCase)
"""
rejection_cases: list[RejectionTestCase] = field(default_factory=list)
rejection_report_test: Optional[CaseMarkers] = None
@dataclass
class ExportLeaksReportState:
"""
Состояние сценария формирования xlsx-отчёта об утечках между allure-шагами.
Заполняется по ходу export_leaks_report в smoke_scenarios.
"""
report_test: Optional[CaseMarkers] = None
period_start: Optional[datetime] = None
period_end: Optional[datetime] = None
period_start_naive: Optional[datetime] = None
period_end_naive: Optional[datetime] = None
expected_mt_mode: Optional[str] = None
expected_lds_status_text: Optional[str] = None
time_offset_hours: Optional[int] = None
tu_description_lower: str = ""
notification: Optional[ReportDataExportedNotification] = None
report_item: Optional[ExportedDataItem] = None
report_file_name: str = ""
download_invocation_id: Optional[str] = None
download_payload: Optional[list] = None
download_reply: Optional[DownloadExportedDataReply] = None
file_bytes: Optional[bytes] = None
temp_file_path: Optional[Path] = None
worksheet: Any = None
title_info: Optional[ReportTitleInfo] = None
data_rows: list[LeakReportRow] = field(default_factory=list)
target_row: Optional[LeakReportRow] = None
@dataclass
class ExportLdsStatusReportState:
"""Состояние сценария формирования xlsx-отчёта о режиме работы СОУ."""
report_test: Optional[CaseMarkers] = None
period_start: Optional[datetime] = None
period_end: Optional[datetime] = None
period_start_naive: Optional[datetime] = None
period_end_naive: Optional[datetime] = None
time_offset_hours: Optional[int] = None
tu_description_lower: str = ""
notification: Optional[ReportDataExportedNotification] = None
report_item: Optional[ExportedDataItem] = None
report_file_name: str = ""
download_invocation_id: Optional[str] = None
download_reply: Optional[DownloadExportedDataReply] = None
file_bytes: Optional[bytes] = None
temp_file_path: Optional[Path] = None
worksheet: Any = None
parsed_report: Any = None
@dataclass
class ExportRejectedReportState:
"""
Состояние сценария формирования xlsx-отчёта об отбракованных входных данных.
Поля с префиксом expected_ - из конфигурации теста, actual_ - из ответов бэка и разбора xlsx.
"""
# --- expected: конфигурация теста и расчётные ожидания ---
expected_report_test: Optional[CaseMarkers] = None
expected_period_start: Optional[datetime] = None
expected_period_end: Optional[datetime] = None
expected_period_start_naive: Optional[datetime] = None
expected_period_end_naive: Optional[datetime] = None
expected_tu_description_lower: str = ""
expected_file_name: str = ""
# --- actual: ответы бэка ---
actual_time_offset_hours: Optional[int] = None
actual_notification: Optional[ReportDataExportedNotification] = None
actual_report_item: Optional[ExportedDataItem] = None
actual_download_invocation_id: Optional[str] = None
actual_download_reply: Optional[DownloadExportedDataReply] = None
actual_file_bytes: Optional[bytes] = None
# --- actual: разбор xlsx ---
actual_temp_file_path: Optional[Path] = None
actual_worksheet: Any = None
actual_title_info: Optional[ReportTitleInfo] = None
actual_data_rows: list[RejectionReportRow] = field(default_factory=list)
actual_monitored_tag_rows: list[RejectionReportRow] = field(default_factory=list)
actual_header_column_headers: list[str] = field(default_factory=list)
actual_header_period_start: Optional[datetime] = None
actual_header_period_end: Optional[datetime] = None
actual_header_contains_expected_title: bool = False
actual_case_checks: list = field(default_factory=list)
@dataclass
class ExportMtModeReportState:
"""
Состояние сценария формирования xlsx-отчёта о режиме работы МТ.
Поля с префиксом expected_ - из конфигурации теста, actual_ - из ответов бэка и разбора xlsx.
"""
# --- expected: конфигурация теста и расчётные ожидания ---
expected_report_test: Optional[CaseMarkers] = None
expected_period_start: Optional[datetime] = None
expected_period_end: Optional[datetime] = None
expected_period_start_naive: Optional[datetime] = None
expected_period_end_naive: Optional[datetime] = None
expected_tu_description_lower: str = ""
expected_file_name: str = ""
expected_section_names: list[str] = field(default_factory=list)
expected_dominant_mode_column: str = ""
# --- actual: ответы бэка ---
actual_time_offset_hours: Optional[int] = None
actual_notification: Optional[ReportDataExportedNotification] = None
actual_report_item: Optional[ExportedDataItem] = None
actual_download_invocation_id: Optional[str] = None
actual_download_reply: Optional[DownloadExportedDataReply] = None
actual_file_bytes: Optional[bytes] = None
# --- actual: разбор xlsx ---
actual_temp_file_path: Optional[Path] = None
actual_worksheet: Any = None
actual_parsed_report: Any = None
lds_status_scenarios
"""
Сценарии тестов по режимам СОУ - функции-обёртки без pytest маркеров.
Каждая функция содержит логику одного теста.
Pytest маркеры и allure декораторы применяются в тестовых файлах.
"""
import time
from datetime import datetime
import allure
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.enums import Direction, LdsStatus, MessageType, ReplyStatus, StationaryStatus
from constants.test_constants import BaseTN3Constants as TestConst
from models.get_messages_model import Filtering, FilteringObjects, Pagination
from test_config.models_for_tests import CaseData, LDSStatusConfig, SmokeSuiteConfig
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 lds_status_check_on_longest_flow_area(ws_client, cfg: LDSStatusConfig, test_data: CaseData):
"""
Проверка Инициализации и причины инициализации СОУ на самом протяженном участке карты течений
"""
# Распаковка данных для теста
expected_lds_status, expected_lds_status_reasons = test_data.expected_result
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_poll_subscribed_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()
for diagnostic_area in diagnostic_areas:
diagnostic_area_lds_status = LdsStatus(diagnostic_area.ldsStatus) if diagnostic_area.ldsStatus else None
StepCheck(f"Проверка режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatus").actual(
diagnostic_area_lds_status
).expected(expected_lds_status).equal_to()
lds_status_reasons = t_utils.parse_lds_status_reasons(
diagnostic_area.ldsStatus, diagnostic_area.ldsStatusReasons
)
StepCheck(f"Проверка причины режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatusReasons").contains(
lds_status_reasons, expected_lds_status_reasons
)
async def lds_status_check_with_reasons(ws_client, cfg: SmokeSuiteConfig | LDSStatusConfig, test_data: CaseData):
"""
Проверка режима работы и причины режима СОУ на заданном ДУ
"""
# Распаковка данных для теста
pipe_id = test_data.params.get("pipe_id")
expected_lds_status, expected_lds_status_reasons = test_data.expected_result
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_poll_subscribed_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', [])
# Получает ДУ
diagnostic_area = t_utils.find_diagnostic_area_by_pipe_id(flow_areas, pipe_id)
StepCheck("Проверка наличия данных диагностического участка", "diagnosticAreas").actual(
diagnostic_area
).is_not_none()
diagnostic_area_lds_status = LdsStatus(diagnostic_area.ldsStatus) if diagnostic_area.ldsStatus else None
StepCheck(f"Проверка режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatus").actual(
diagnostic_area_lds_status
).expected(expected_lds_status).equal_to()
if diagnostic_area_lds_status != LdsStatus.SERVICEABLE:
lds_status_reasons = t_utils.parse_lds_status_reasons(
diagnostic_area.ldsStatus, diagnostic_area.ldsStatusReasons
)
StepCheck(f"Проверка причины режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatusReasons").contains(
lds_status_reasons, expected_lds_status_reasons
)
async def lds_status_check_with_2_reasons(ws_client, cfg: SmokeSuiteConfig | LDSStatusConfig, test_data: CaseData):
"""
Проверка режима работы СОУ и двух причин режима работы СОУ на заданном ДУ
"""
# Распаковка данных для теста
pipe_id = test_data.params.get("pipe_id")
expected_lds_status, expected_lds_status_reason_1, expected_lds_status_reason_2 = test_data.expected_result
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_poll_subscribed_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', [])
# Получает ДУ
diagnostic_area = t_utils.find_diagnostic_area_by_pipe_id(flow_areas, pipe_id)
StepCheck("Проверка наличия данных диагностического участка", "diagnosticAreas").actual(
diagnostic_area
).is_not_none()
diagnostic_area_lds_status = LdsStatus(diagnostic_area.ldsStatus) if diagnostic_area.ldsStatus else None
StepCheck(f"Проверка режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatus").actual(
diagnostic_area_lds_status
).expected(expected_lds_status).equal_to()
if diagnostic_area_lds_status != LdsStatus.SERVICEABLE:
with SoftAssertions() as soft_failures:
lds_status_reasons = t_utils.parse_lds_status_reasons(
diagnostic_area.ldsStatus, diagnostic_area.ldsStatusReasons, soft_failures
)
StepCheck(
f"Проверка причины режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatusReasons", soft_failures
).contains(lds_status_reasons, expected_lds_status_reason_1)
StepCheck(
f"Проверка причины режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatusReasons", soft_failures
).contains(lds_status_reasons, expected_lds_status_reason_2)
async def lds_status_check_on_multiple_diagnostic_areas(
ws_client, cfg: SmokeSuiteConfig | LDSStatusConfig, test_data: CaseData
):
"""
Проверка режима работы и причины режима СОУ на заданных ДУ
"""
# Распаковка данных для теста
pipe_ids = test_data.params.get("pipe_ids")
expected_result = test_data.expected_result
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_poll_subscribed_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', [])
diagnostic_areas = t_utils.find_diagnostic_areas_by_pipe_ids(flow_areas, pipe_ids)
StepCheck("Проверка наличия данных диагностических участков", "diagnosticAreas").actual(
diagnostic_areas
).is_not_empty()
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(expected_result).equal_to()
async def lds_and_stationary_status_check_with_reasons(
ws_client, cfg: SmokeSuiteConfig | LDSStatusConfig, test_data: CaseData
):
"""
Проверка режима работы и причины режима СОУ и МТ на заданном ДУ
"""
# Распаковка данных для теста
pipe_id = test_data.params.get("pipe_id")
(
expected_lds_status,
expected_lds_status_reasons,
expected_stationary_status,
expected_stationary_status_reasons,
) = test_data.expected_result
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_poll_subscribed_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', [])
# Получает ДУ
diagnostic_area = t_utils.find_diagnostic_area_by_pipe_id(flow_areas, pipe_id)
StepCheck("Проверка наличия данных диагностического участка", "diagnosticAreas").actual(
diagnostic_area
).is_not_none()
diagnostic_area_lds_status = LdsStatus(diagnostic_area.ldsStatus) if diagnostic_area.ldsStatus else None
diagnostic_area_stationary_status = (
StationaryStatus(diagnostic_area.stationaryStatus) if diagnostic_area.stationaryStatus else None
)
StepCheck(f"Проверка режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatus").actual(
diagnostic_area_lds_status
).expected(expected_lds_status).equal_to()
StepCheck(f"Проверка режима работы МТ на ДУ с id:{diagnostic_area.id}", "stationaryStatus").actual(
diagnostic_area_stationary_status
).expected(expected_stationary_status).equal_to()
with SoftAssertions() as soft_failures:
lds_status_reasons = t_utils.parse_lds_status_reasons(
diagnostic_area.ldsStatus, diagnostic_area.ldsStatusReasons, soft_failures
)
stationary_status_reasons = t_utils.parse_stationary_status_reasons(
diagnostic_area.stationaryStatus, diagnostic_area.stationaryStatusReasons, soft_failures
)
StepCheck(
f"Проверка причины режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatusReasons", soft_failures
).contains(lds_status_reasons, expected_lds_status_reasons)
StepCheck(
f"Проверка причины режима работы МТ на ДУ с id:{diagnostic_area.id}",
"stationaryStatusReasons",
soft_failures,
).contains(stationary_status_reasons, expected_stationary_status_reasons)
async def lds_status_check_degradation_pig_sensor_passage(
ws_client, http_client, cfg: SmokeSuiteConfig | LDSStatusConfig, test_data: CaseData
):
"""
Проверка режима работы и причины режима СОУ на заданном ДУ, с командой на включение СОД
"""
# Распаковка данных для теста
pipe_id = test_data.params.get("pipe_id")
pig_trap_id = test_data.params.get("pig_trap_id")
expected_lds_status, expected_lds_status_reasons = test_data.expected_result
with allure.step("Http запрос, получение и обработка ответа о запуске СОД: LaunchPigRequest"):
request_body = {'pigTrapId': pig_trap_id, 'timeToLaunch': 0, 'tuId': cfg.tu_id}
response = http_client.post_request(HttpConst.LAUNCH_PIG_URL_PATH, request_body)
launch_pig_reply_status = response.status_code
StepCheck("Проверка кода ответа на запрос о запуске СОД", "replyStatus").actual(
launch_pig_reply_status
).expected(ReplyStatus.OK.value).equal_to()
time.sleep(cfg.basic_message_timeout)
with allure.step("Подключение по ws, получение и обработка сообщения типа: CommonSchemeContent"):
payload = await t_utils.connect_and_poll_subscribed_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', [])
# Получает ДУ
diagnostic_area = t_utils.find_diagnostic_area_by_pipe_id(flow_areas, pipe_id)
StepCheck("Проверка наличия данных диагностического участка", "diagnosticAreas").actual(
diagnostic_area
).is_not_none()
diagnostic_area_lds_status = LdsStatus(diagnostic_area.ldsStatus) if diagnostic_area.ldsStatus else None
StepCheck(f"Проверка режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatus").actual(
diagnostic_area_lds_status
).expected(expected_lds_status).equal_to()
lds_status_reasons = t_utils.parse_lds_status_reasons(diagnostic_area.ldsStatus, diagnostic_area.ldsStatusReasons)
StepCheck(f"Проверка причины режима работы СОУ на ДУ с id:{diagnostic_area.id}", "ldsStatusReasons").contains(
lds_status_reasons, expected_lds_status_reasons
)
def lds_status_in_journal(http_client, cfg: LDSStatusConfig, test_data: CaseData):
"""
Проверка наличия записи в журнале о выходе СОУ из режима Инициализация.
"""
# Распаковка данных для теста
control_points = test_data.params.get("control_points")
if isinstance(test_data.expected_result, tuple):
expected_lds_status, expected_lds_status_reasons = test_data.expected_result
else:
expected_lds_status, expected_lds_status_reasons = test_data.expected_result, None
with allure.step("Http запрос сообщений журнала с фильтром messageTypes=MASKING_LDS"):
end_time = datetime.now()
start_time = t_utils.datetime_minus_seconds(end_time, TestConst.JOURNAL_STATUS_TOTAL_WAIT)
request_body = t_utils.create_journal_req_body(
pagination=Pagination(limit=TestConst.JOURNAL_PAGINATION_STATUS_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("Фильтрация сообщений по времени и controlPoint"):
filter_start_msk = t_utils.localize_as_moscow(start_time)
filter_end_msk = t_utils.localize_as_moscow(end_time)
lds_msg_by_control_points = []
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)
# Поиск нужных сообщений по КП
for control_point in control_points:
lds_msg = next(
(msg for msg in time_filtered if msg.controlPoint == control_point),
None,
)
if lds_msg:
lds_msg_by_control_points.append(lds_msg)
StepCheck(
f"Проверка наличия сообщений c controlPoint из списка {control_points} в журнале", "messagesInfo"
).actual(lds_msg_by_control_points).is_not_empty()
with SoftAssertions() as soft_failures:
for msg in lds_msg_by_control_points:
msg_event = getattr(msg, 'event', None)
cp_lds_status, cp_lds_status_reasons = t_utils.parse_event(msg_event)
StepCheck(f"Проверка режима работы СОУ на КП:{msg.controlPoint}", "event", soft_failures).actual(
cp_lds_status
).expected(expected_lds_status).equal_to()
if cp_lds_status != LdsStatus.SERVICEABLE.report_text:
StepCheck(
f"Проверка причины режима работы СОУ на КП:{msg.controlPoint}", "event", soft_failures
).contains(cp_lds_status_reasons, expected_lds_status_reasons)
