https://pastein.ru/t/N20
Загрузка данных
бики
"""
Конфигурация тестового набора 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
DIAGNOSTIC_AREA_2_CONTROL_POINT = "ЗА 105-3 - ЗА 129-3"
DIAGNOSTIC_AREA_3_CONTROL_POINT = "ЗА 180-3 - ЗА 207-3"
# Технологический участок
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,
),
),
degradation_temperature_du_2_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.REJECTION_TEMPERATURE_SENSOR.report_text,
),
),
degradation_density_du_2_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.REJECTION_DENSITY_SENSOR.report_text,
),
),
degradation_viscosity_du_2_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.REJECTION_VISCOSITY_SENSOR.report_text,
),
),
degradation_temperature_sensor_du_3_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_3_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.REJECTION_TEMPERATURE_SENSOR.report_text,
),
),
degradation_density_du_3_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_3_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.REJECTION_DENSITY_SENSOR.report_text,
),
),
degradation_viscosity_du_3_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_3_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.REJECTION_VISCOSITY_SENSOR.report_text,
),
),
# ===== Базовые ТЕСТЫ ====
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=54),
deg_rejection_density_and_viscosity_on_du_3_test=CaseMarkers(test_case_id="181", offset=80),
# ===== ТЕСТЫ в журнале для ДУ2 =====
degradation_temperature_du_2_in_journal_test=CaseMarkers(test_case_id="179", offset=23),
degradation_density_du_2_in_journal_test=CaseMarkers(test_case_id="181", offset=32),
degradation_viscosity_du_2_in_journal_test=CaseMarkers(test_case_id="181", offset=32),
# ===== ТЕСТЫ в журнале для ДУ3 =====
degradation_temperature_du_3_in_journal_test=CaseMarkers(test_case_id="179", offset=46),
degradation_density_du_3_in_journal_test=CaseMarkers(test_case_id="181", offset=53),
degradation_viscosity_du_3_in_journal_test=CaseMarkers(test_case_id="181", offset=53),
)
флоу
"""
Конфигурация тестового набора 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
DIAGNOSTIC_AREA_2_CONTROL_POINT = "ЗА 105-3 - ЗА 129-3"
DIAGNOSTIC_AREA_3_CONTROL_POINT = "ЗА 180-3 - ЗА 207-3"
DIAGNOSTIC_AREA_5_CONTROL_POINT = "ЗА 235-3-1 - ЗА 246-3-2"
# Технологический участок
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,
),
),
# ===== Ожидания для сообщений о состонии СОУ. ЭФ Журнал =====
between_si_pressure_more_50_km_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_3_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.EXCEEDING_DISTANCE_BETWEEN_SERVICEABLE_PRESSURE_SENSORS.report_text,
),
),
deg_gravity_section_pumping_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_5_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.GRAVITY_SECTION_IN_PUMPING_MODE.report_text,
),
),
deg_absence_min_pressure_sensors_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_5_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.ABSENCE_MIN_PRESSURE_SENSORS_REQUIRED_NUMBER.report_text,
),
),
serviceable_after_deg_absence_min_pressure_sensors_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_5_CONTROL_POINT]},
expected_result=(LdsStatus.SERVICEABLE.report_text,),
),
deg_starting_pumping_out_pumps_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.STARTING_PUMPING_OUT_PUMPS.report_text,
),
),
serviceable_after_deg_starting_pumping_out_pumps_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(LdsStatus.SERVICEABLE.report_text,),
),
faulty_absence_min_flow_meters_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_5_CONTROL_POINT]},
expected_result=(
LdsStatus.FAULTY.report_text,
LdsStatusFaulty.ABSENCE_MIN_FLOW_METERS_REQUIRED_NUMBER.report_text,
),
),
serviceable_after_faulty_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(LdsStatus.SERVICEABLE.report_text,),
),
deg_exceeding_distance_between_flow_meters_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_3_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.EXCEEDING_DISTANCE_BETWEEN_FLOW_METERS.report_text,
),
),
deg_faulty_pressure_sensors_at_pump_station_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.FAULTY_PRESSURE_SENSORS_AT_PUMP_STATION_NODES.report_text,
),
),
serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(LdsStatus.SERVICEABLE.report_text,),
),
deg_additive_injectors_operation_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_3_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.ADDITIVE_INJECTORS_OPERATION.report_text,
),
),
deg_pig_sensor_passage_in_journal_test_data=CaseData(
params={"control_points": [DIAGNOSTIC_AREA_2_CONTROL_POINT]},
expected_result=(
LdsStatus.DEGRADATION.report_text,
LdsStatusDegradation.PIG_SENSOR_PASSAGE.report_text,
),
),
# ===== ТЕСТЫ состония СОУ на схеме =====
lds_status_basic_info_test=CaseMarkers(test_case_id="1", offset=5),
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),
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),
# ===== ТЕСТЫ состония СОУ на схеме =====
init_cold_start_test=CaseMarkers(test_case_id="160", offset=7),
between_si_pressure_more_50_km_in_journal_test=CaseMarkers(test_case_id="228", offset=35),
deg_gravity_section_pumping_in_journal_test=CaseMarkers(test_case_id="167", offset=34),
deg_absence_min_pressure_sensors_in_journal_test=CaseMarkers(test_case_id="166", offset=36),
serviceable_after_deg_absence_min_pressure_sensors_in_journal_test=CaseMarkers(test_case_id="186", offset=43),
deg_starting_pumping_out_pumps_in_journal_test=CaseMarkers(test_case_id="174", offset=46),
serviceable_after_deg_starting_pumping_out_pumps_in_journal_test=CaseMarkers(test_case_id="186", offset=53),
faulty_absence_min_flow_meters_in_journal_test=CaseMarkers(test_case_id="170", offset=57),
serviceable_after_faulty_in_journal_test=CaseMarkers(test_case_id="186", offset=76),
deg_exceeding_distance_between_flow_meters_in_journal_test=CaseMarkers(test_case_id="169", offset=107), # 106
deg_faulty_pressure_sensors_at_pump_station_in_journal_test=CaseMarkers(test_case_id="184", offset=120),
deg_additive_injectors_operation_in_journal_test=CaseMarkers(test_case_id="188", offset=132),
deg_pig_sensor_passage_in_journal_test=CaseMarkers(test_case_id="165", offset=140),
)
модели
"""
Датаклассы для конфигурации тестовых наборов.
Архитектура:
- 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. Тесты с маркерами
"""
# ===== Данные для тестов =====
main_pipeline: Optional[str] = None
# ===== Данные для тестов на схеме=====
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
# ===== Данные для тестов в журнале бики ДУ2=====
degradation_temperature_du_2_in_journal_test_data: Optional[CaseData] = None
degradation_density_du_2_in_journal_test_data: Optional[CaseData] = None
degradation_viscosity_du_2_in_journal_test_data: Optional[CaseData] = None
# ===== Данные для тестов в журнале бики ДУ3=====
degradation_temperature_sensor_du_3_in_journal_test_data: Optional[CaseData] = None
degradation_density_du_3_in_journal_test_data: Optional[CaseData] = None
degradation_viscosity_du_3_in_journal_test_data: Optional[CaseData] = None
# ===== Данные для тестов в журнале для набора данных "В течении"=====
between_si_pressure_more_50_km_in_journal_test_data: Optional[CaseData] = None
deg_gravity_section_pumping_in_journal_test_data: Optional[CaseData] = None
deg_absence_min_pressure_sensors_in_journal_test_data: Optional[CaseData] = None
serviceable_after_deg_absence_min_pressure_sensors_in_journal_test_data: Optional[CaseData] = None
deg_starting_pumping_out_pumps_in_journal_test_data: Optional[CaseData] = None
serviceable_after_faulty_in_journal_test_data: Optional[CaseData] = None
deg_additive_injectors_operation_in_journal_test_data: Optional[CaseData] = None
deg_pig_sensor_passage_in_journal_test_data: Optional[CaseData] = None
deg_exceeding_distance_between_flow_meters_in_journal_test_data: Optional[CaseData] = None
serviceable_after_deg_starting_pumping_out_pumps_in_journal_test_data: Optional[CaseData] = None
faulty_absence_min_flow_meters_in_journal_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
# ===== ТЕСТЫ в журнале бики ДУ2=====
degradation_temperature_du_2_in_journal_test: Optional[CaseMarkers] = None
degradation_density_du_2_in_journal_test: Optional[CaseMarkers] = None
degradation_viscosity_du_2_in_journal_test: Optional[CaseMarkers] = None
# ===== ТЕСТЫ в журнале бики ДУ3=====
degradation_temperature_du_3_in_journal_test: Optional[CaseMarkers] = None
degradation_density_du_3_in_journal_test: Optional[CaseMarkers] = None
degradation_viscosity_du_3_in_journal_test: Optional[CaseMarkers] = None
# ===== ТЕСТЫ в журнале для набора данных "В течении"=====
between_si_pressure_more_50_km_in_journal_test: Optional[CaseMarkers] = None
deg_gravity_section_pumping_in_journal_test: Optional[CaseMarkers] = None
deg_absence_min_pressure_sensors_in_journal_test: Optional[CaseMarkers] = None
serviceable_after_deg_absence_min_pressure_sensors_in_journal_test: Optional[CaseMarkers] = None
deg_starting_pumping_out_pumps_in_journal_test: Optional[CaseMarkers] = None
serviceable_after_deg_starting_pumping_out_pumps_in_journal_test: Optional[CaseMarkers] = None
faulty_absence_min_flow_meters_in_journal_test: Optional[CaseMarkers] = None
serviceable_after_faulty_in_journal_test: Optional[CaseMarkers] = None
deg_exceeding_distance_between_flow_meters_in_journal_test: Optional[CaseMarkers] = None
deg_additive_injectors_operation_in_journal_test: Optional[CaseMarkers] = None
deg_pig_sensor_passage_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
лдс статус сценариос
"""
Сценарии тестов по режимам СОУ - функции-обёртки без 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=LDS_STATUS"):
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"):
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
]
# Поиск нужных сообщений по КП
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)
лдс статус регрес
"""
Архитектура параметризации:
- SUITE_PARAMS: тесты уровня набора (один раз на набор данных LDSStatusConfig)
Для добавления нового набора:
1. Создать файл в test_config/datasets/ с LDSStatusConfig
2. Добавить маппинг новых тестов в conftest.py -> LDS_STATUS_SUITE_LEVEL_MAPPING
3. Тесты подхватятся автоматически
Запуск:
- Все тесты: pytest tests/test_lds_status_regress.py
- Один набор: pytest tests/test_lds_status_regress.py --suites=Lds_status_regress_in_flow
- Несколько наборов: pytest tests/test_lds_status_regress.py --suites=Lds_status_regress_in_flow,Lds_status_regress_2
"""
import datetime
from typing import Any, List, Optional
import allure
import pytest
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from test_config.datasets import ALL_LDS_STATUS_CONFIGS
from test_config.models_for_tests import CaseMarkers, LDSStatusConfig
from test_scenarios import lds_status_scenarios as scenarios
from test_scenarios import smoke_scenarios
# ===== ГЕНЕРАЦИЯ ПАРАМЕТРОВ =====
def _get_suite_markers(config: LDSStatusConfig) -> List[pytest.MarkDecorator]:
"""Маркеры для тестового набора."""
return [
pytest.mark.test_suite_name(config.suite_name),
pytest.mark.test_suite_data_id(config.suite_data_id),
pytest.mark.test_data_name(config.archive_name),
pytest.mark.tu_id(config.technological_unit.id),
pytest.mark.ost_name(config.ost_name),
]
def _generate_suite_params() -> List[Any]:
"""
Генерирует параметры для тестов уровня набора данных.
Один параметр на каждый config.
"""
return [
pytest.param(config, id=config.suite_name, marks=_get_suite_markers(config))
for config in ALL_LDS_STATUS_CONFIGS
]
# ===== ПАРАМЕТРЫ ДЛЯ ТЕСТОВ =====
SUITE_PARAMS: List[Any] = _generate_suite_params()
# ===== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =====
def _apply_allure_markers(test_config: CaseMarkers, tag: str, title: str, description: Optional[str] = None) -> None:
"""Применяет allure-маркеры из конфига теста."""
if not test_config:
pytest.skip("Не заполнена конфигурация теста: тест пропущен")
allure.dynamic.tag(tag)
allure.dynamic.tag("REGRESS")
allure.dynamic.title(title)
if description:
allure.dynamic.description(description)
# ===== ТЕСТЫ УРОВНЯ НАБОРА =====
# Запускаются один раз для каждого config
@pytest.mark.parametrize("config", SUITE_PARAMS)
class TestSuiteScenarios:
"""
Тесты уровня набора данных.
Запускаются один раз для каждого конфига.
"""
@pytest.mark.critical_stop
def test_lds_status_basic_info(self, http_client: StandHttpClient, config: LDSStatusConfig) -> None:
"""[BasicInfo] Проверка базовой информации СОУ: список ТУ"""
tag = "BasicInfo"
title = f"[{tag}] Проверка списка ТУ. ЭФ: Главная страница"
_apply_allure_markers(config.lds_status_basic_info_test, tag, title)
smoke_scenarios.basic_info(http_client, config)
@pytest.mark.asyncio
async def test_lds_status_init_accumulation_data(self, ws_client: WebSocketClient, config: LDSStatusConfig) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация,
Причина: Накопление данных
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Инициализация', по причине: 'Накопление данных'. ЭФ: Схема"
_apply_allure_markers(
config.init_accumulation_data_test,
tag,
title,
(
f"Проверка режима работы СОУ на базовых ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_accumulation_data_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Накопление данных"
),
)
test_data = config.init_accumulation_data_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_init_cold_start(self, ws_client: WebSocketClient, config: LDSStatusConfig) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация,
Причина: Холодный пуск
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Инициализация', по причине: 'Холодный пуск'. ЭФ: Схема"
_apply_allure_markers(
config.init_cold_start_test,
tag,
title,
(
f"Проверка режима работы СОУ на базовых ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_cold_start_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n Ожидаемая причина режима работы СОУ: Холодный пуск"
),
)
test_data = config.init_cold_start_test_data
await scenarios.lds_status_check_on_longest_flow_area(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_init_switching_shut_off(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация,
Причина: Переключение запорной арматуры
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Инициализация', "
"по причине: 'Переключение запорной арматуры'. ЭФ: Схема"
)
_apply_allure_markers(
config.init_switching_shut_off_test,
tag,
title,
(
f"Проверка режима работы СОУ на базовых ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_switching_shut_off_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Переключение запорной арматуры"
),
)
test_data = config.init_switching_shut_off_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_cold_start(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после Инициализации, по причине Холодный пуск.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после Иницализации. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_cold_start_test,
tag,
title,
(
"Проверка режима работы СОУ после Инициализации, по причине: Холодный пуск, на показательных ДУ"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.serviceable_after_cold_start_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_additive_injectors_operation(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_additive_injectors_operation_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.technological_unit.description}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_additive_injectors_operation_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_absence_min_pressure_sensors(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_absence_min_pressure_sensors_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_absence_min_pressure_sensors_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_exceeding_distance_between_flow_meters(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_exceeding_distance_between_flow_meters_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.technological_unit.description}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_exceeding_distance_between_flow_meters_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_starting_pumping_out_pumps(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_starting_pumping_out_pumps_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_starting_pumping_out_pumps_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_deg_faulty_pressure_sensors_at_pump(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна' после режима Ухудшение характеристик. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_deg_faulty_pressure_sensors_at_pump_test,
tag,
title,
(
"Проверка режима работы СОУ, после режима Ухудшение характеристик, "
"на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки: "
f"{config.serviceable_after_deg_faulty_pressure_sensors_at_pump_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_faulty(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после Инициализации, по причине выхода из Неисправности.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна', после Неисправности. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_faulty_test,
tag,
title,
(
"Проверка режима работы СОУ, после Неисправности, на показательных ДУ,"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки: {config.serviceable_after_faulty_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_serviceable_after_switching_shut_off(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Исправна.
Установка режима Исправна после Инициализации, по причине Переключение запорной арматуры.
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Исправна', после Инициализации. ЭФ: Схема"
_apply_allure_markers(
config.serviceable_after_switching_shut_off_test,
tag,
title,
(
"Проверка режима работы СОУ, после Инициализации по причине Переключение запорной арматуры"
f" на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки: "
f"{config.serviceable_after_switching_shut_off_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Исправна"
),
)
test_data = config.serviceable_all_test_data
await scenarios.lds_status_check_on_multiple_diagnostic_areas(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_additive_injectors_operation(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Наличие ПТП
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', по причине: 'Наличие ПТП'. ЭФ: Схема"
_apply_allure_markers(
config.deg_additive_injectors_operation_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_additive_injectors_operation_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Наличие ПТП"
),
)
test_data = config.deg_additive_injectors_operation_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_exceeding_distance_between_pressure_sensors(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: расстояние между СИ давления более 50 км
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Расстояние между СИ давления более 50 км'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_exceeding_distance_between_pressure_sensors_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_exceeding_distance_between_pressure_sensors_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Расстояние между СИ давления более 50 км"
),
)
test_data = config.deg_exceeding_distance_between_pressure_sensors_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_absence_min_pressure_sensors(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: менее 4 исправных СИ давления
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'менее 4 исправных СИ давления'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_absence_min_pressure_sensors_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_absence_min_pressure_sensors_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Ухудшение При наличии менее четырех (3, 6) исправных СИ давления "
"на разных КП ЛЧ МТ/НПС на диагностическом участке (кроме случая нахождения трубопровода "
"в режиме остановленной перекачки)"
),
)
test_data = config.deg_absence_min_pressure_sensors_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_faulty_pressure_sensors_at_pump_station(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Отказ СИ давления на входе/выходе НПС
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ давления на входе/выходе НПС'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_faulty_pressure_sensors_at_pump_station_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_faulty_pressure_sensors_at_pump_station_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ давления на входе/выходе НПС"
),
)
test_data = config.deg_faulty_pressure_sensors_at_pump_station_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_pig_sensor_passage(
self, ws_client: WebSocketClient, http_client: StandHttpClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: прохождение СОД
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', по причине: 'прохождение СОД'"
_apply_allure_markers(
config.deg_pig_sensor_passage_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.deg_pig_sensor_passage_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: прохождение СОД "
),
)
test_data = config.deg_pig_sensor_passage_test_data
await scenarios.lds_status_check_degradation_pig_sensor_passage(ws_client, http_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_gravity_section_pumping(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Наличие самотечного участка/участка с неполным сечением
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Наличие самотечного участка/участка с неполным сечением'"
)
_apply_allure_markers(
config.deg_gravity_section_pumping_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.deg_gravity_section_pumping_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Наличие самотечного участка/участка с неполным сечением "
),
)
test_data = config.deg_gravity_section_pumping_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_gravity_section_pumping_in_stopping(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Наличие самотечного участка в режиме остановленной перекачки
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Наличие самотечного участка в режиме остановленной перекачки'"
)
_apply_allure_markers(
config.deg_gravity_section_pumping_in_stopping_test,
tag,
title,
(
f"Проверка режима работы СОУ на всей ЛЧ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_gravity_section_pumping_in_stopping_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Наличие самотечного участка в режиме остановленной перекачки "
),
)
test_data = config.deg_gravity_section_pumping_in_stopping_test_data
await scenarios.lds_status_check_on_longest_flow_area(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_starting_pumping_out_pumps(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ и МТ
Причина: Работа насосов откачки
"""
tag = "CommonScheme"
title = f"[{tag}] Проверка режимов работы СОУ и МТ, по причине: 'Работа насосов откачки'. ЭФ: Схема"
_apply_allure_markers(
config.deg_starting_pumping_out_pumps_test,
tag,
title,
(
f"Проверка режимов работы СОУ и МТ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_starting_pumping_out_pumps_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемый режим работы МТ: Нестационарный\n "
"Ожидаемая причина режима работы СОУ: Работа насосов откачки"
"Ожидаемая причина режима работы МТ: Работа насосов откачки"
),
)
test_data = config.deg_starting_pumping_out_pumps_test_data
await scenarios.lds_and_stationary_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_exceeding_distance_between_flow_meters(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Расстояние между СИ расхода на пути перекачки более 200 км
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Расстояние между СИ расхода на пути перекачки более 200 км'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_exceeding_distance_between_flow_meters_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_exceeding_distance_between_flow_meters_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Расстояние между СИ расхода на пути перекачки более 200 км"
),
)
test_data = config.deg_exceeding_distance_between_flow_meters_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_init_exiting_faulty(self, ws_client: WebSocketClient, config: LDSStatusConfig) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Инициализация,
Причина: Выход СОУ из режима «Неисправна»
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Инициализация', "
"по причине: 'Выход СОУ из режима «Неисправна»'. ЭФ: Схема"
)
_apply_allure_markers(
config.init_exiting_faulty_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_exiting_faulty_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Выход СОУ из режима «Неисправна»"
),
)
test_data = config.init_exiting_faulty_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_temperature_sensor_on_du_2(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Отказ СИ температуры
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 2: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ температуры'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_temperature_sensor_on_du_2_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 2, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_temperature_sensor_on_du_2_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ температуры"
),
)
test_data = config.deg_rejection_temperature_sensor_on_du_2_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_temperature_sensor_on_du_3(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Отказ СИ температуры
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 3: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ температуры'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_temperature_sensor_on_du_3_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 3, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_temperature_sensor_on_du_3_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ температуры"
),
)
test_data = config.deg_rejection_temperature_sensor_on_du_3_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_temperature_sensor_on_du_5(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причина: Отказ СИ температуры
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 5: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ температуры'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_temperature_sensor_on_du_5_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 5, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_temperature_sensor_on_du_5_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ температуры"
),
)
test_data = config.deg_rejection_temperature_sensor_on_du_5_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_density_and_viscosity_on_du_2(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причины: Отказ СИ плотности и СИ вязкости
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 2: 'Ухудшение характеристик', "
"по причинам: 'Отказ СИ плотности' и 'Отказ СИ вязкости'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_density_and_viscosity_on_du_2_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 2, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_density_and_viscosity_on_du_2_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ плотности"
"Ожидаемая причина режима работы СОУ: Отказ СИ вязкости"
),
)
test_data = config.deg_rejection_density_and_viscosity_on_du_2_test_data
await scenarios.lds_status_check_with_2_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_density_and_viscosity_on_du_3(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причины: Отказ СИ плотности и СИ вязкости
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 3: 'Ухудшение характеристик', "
"по причинам: 'Отказ СИ плотности' и 'Отказ СИ вязкости'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_density_and_viscosity_on_du_3_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 3, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_density_and_viscosity_on_du_3_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ плотности"
"Ожидаемая причина режима работы СОУ: Отказ СИ вязкости"
),
)
test_data = config.deg_rejection_density_and_viscosity_on_du_3_test_data
await scenarios.lds_status_check_with_2_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_degradation_rejection_density_and_viscosity_on_du_5(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Ухудшение характеристик
Причины: Отказ СИ плотности и СИ вязкости
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ на ДУ 5: 'Ухудшение характеристик', "
"по причинам: 'Отказ СИ плотности' и 'Отказ СИ вязкости'. ЭФ: Схема"
)
_apply_allure_markers(
config.deg_rejection_density_and_viscosity_on_du_5_test,
tag,
title,
(
f"Проверка режима работы СОУ на ДУ 5, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_rejection_density_and_viscosity_on_du_5_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ плотности"
"Ожидаемая причина режима работы СОУ: Отказ СИ вязкости"
),
)
test_data = config.deg_rejection_density_and_viscosity_on_du_5_test_data
await scenarios.lds_status_check_with_2_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_faulty_absence_min_flow_meters(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Неисправна
Причина: Недостоверность граничного СИ расхода
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Неисправна', "
"по причине: 'Недостоверность граничного СИ расхода'. ЭФ: Схема"
)
_apply_allure_markers(
config.faulty_absence_min_flow_meters_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_flow_meters_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Неисправна\n "
"Ожидаемая причина режима работы СОУ: Недостоверность граничного СИ расхода"
),
)
test_data = config.faulty_absence_min_flow_meters_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_faulty_absence_min_flow_meters_continuous(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Неисправна
Причина:Недостоверность граничного СИ расхода
После устранения причины неисправности
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Неисправна', "
"по причине: 'Недостоверность граничного СИ расхода'. "
"После устранения причины неисправности. ЭФ: Схема"
)
_apply_allure_markers(
config.faulty_absence_min_flow_meters_continuous_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_flow_meters_continuous_test.offset} мин.\n"
"После устранения причины неисправности (Заменяет Инициализацию при непродолжительной неисправности)\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Неисправна\n "
"Ожидаемая причина режима работы СОУ: Недостоверность граничного СИ расхода"
),
)
test_data = config.faulty_absence_min_flow_meters_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
@pytest.mark.asyncio
async def test_lds_status_faulty_absence_min_pressure_sensors(
self, ws_client: WebSocketClient, config: LDSStatusConfig
) -> None:
"""
[CommonScheme] Проверка режима работы СОУ: Неисправна
Причина: Менее 4 КП с достоверными СИ давления
"""
tag = "CommonScheme"
title = (
f"[{tag}] Проверка режима работы СОУ: 'Неисправна', "
"по причине: 'Менее 4 КП с достоверными СИ давления'. ЭФ: Схема"
)
_apply_allure_markers(
config.faulty_absence_min_pressure_sensors_test,
tag,
title,
(
f"Проверка режима работы СОУ на выбранном ДУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_pressure_sensors_test.offset} мин.\n"
"Подписка на сообщения типа: CommonScheme\n"
"Ожидаемый режим работы СОУ: Неисправна\n "
"Ожидаемая причина режима работы СОУ: Менее 4 КП с достоверными СИ давления"
),
)
test_data = config.faulty_absence_min_pressure_sensors_test_data
await scenarios.lds_status_check_with_reasons(ws_client, config, test_data)
def test_lds_status_init_accumulation_data_in_journal(
self, http_client: StandHttpClient, config: LDSStatusConfig
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Инициализация,
Причина: Накопление данных
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Инициализация', по причине: 'Накопление данных'. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.init_accumulation_data_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_accumulation_data_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Инициализация\n Ожидаемая причина режима работы СОУ: Накопление данных"
),
)
test_data = config.init_accumulation_data_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
def test_lds_status_init_cold_start_in_journal(
self, http_client: StandHttpClient, config: LDSStatusConfig, imitator_start_time: datetime
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Инициализация,
Причина: Холодный пуск
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Инициализация', по причине: 'Холодный пуск'. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.init_cold_start_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
f"Время проведения проверки : {config.init_cold_start_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Инициализация\n Ожидаемая причина режима работы СОУ: Холодный пуск"
),
)
smoke_scenarios.lds_status_init_in_journal(http_client, config, imitator_start_time)
def test_lds_status_init_switching_shut_off_in_journal(
self, http_client: StandHttpClient, config: LDSStatusConfig
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Инициализация,
Причина: Переключение запорной арматуры
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Инициализация', по причине: 'Переключение запорной арматуры'. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.init_switching_shut_off_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.init_switching_shut_off_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Инициализация\n "
"Ожидаемая причина режима работы СОУ: Переключение запорной арматуры"
),
)
test_data = config.init_switching_shut_off_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_serviceable_after_cold_start_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Исправна.
Установка режима Исправна после Инициализации, по причине Холодный пуск.
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Исправна', после Инициализации. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.serviceable_after_cold_start_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.serviceable_after_cold_start_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Исправна\n"
),
)
test_data = config.serviceable_all_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_serviceable_after_switching_shut_off_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Исправна.
Установка режима Исправна после Инициализации, по причине Переключение запорной арматуры.
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Исправна', после Инициализации. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.serviceable_after_switching_shut_off_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.serviceable_after_switching_shut_off_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Исправна\n"
),
)
test_data = config.serviceable_all_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Исправна.
Установка режима Исправна после режима Ухудшение характеристик.
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: "
"'Исправна', после режима Ухудшение характеристик. ЭФ: Журнал. Реальное время"
)
_apply_allure_markers(
config.serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Исправна\n"
),
)
test_data = config.serviceable_all_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_degradation_gravity_section_pumping_in_stopping_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение характеристик
Причина: Наличие самотечного участка в режиме остановленной перекачки
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Наличие самотечного участка в режиме остановленной перекачки'"
)
_apply_allure_markers(
config.deg_gravity_section_pumping_in_stopping_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_gravity_section_pumping_in_stopping_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Наличие самотечного участка в режиме остановленной перекачки "
),
)
test_data = config.deg_gravity_section_pumping_in_stopping_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_degradation_exceeding_distance_between_pressure_sensors_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение характеристик
Причина: расстояние между СИ давления более 50 км
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Расстояние между СИ давления более 50 км'"
)
_apply_allure_markers(
config.deg_exceeding_distance_between_pressure_sensors_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_exceeding_distance_between_pressure_sensors_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Расстояние между СИ давления более 50 км "
),
)
test_data = config.deg_exceeding_distance_between_pressure_sensors_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_degradation_faulty_pressure_sensors_at_pump_station_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение характеристик
Причина: Отказ СИ давления на входе/выходе НПС
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: 'Ухудшение характеристик', "
"по причине: 'Отказ СИ давления на входе/выходе НПС'"
)
_apply_allure_markers(
config.deg_faulty_pressure_sensors_at_pump_station_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_faulty_pressure_sensors_at_pump_station_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ давления на входе/выходе НПС "
),
)
test_data = config.deg_faulty_pressure_sensors_at_pump_station_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
def test_lds_status_faulty_absence_min_pressure_sensors_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Неисправна
Причина: Менее 4 КП с достоверными СИ давления
"""
tag = "MessagesInfo"
title = (
f"[{tag}] Проверка записи в журнале о режиме работы СОУ: 'Неисправна', "
"по причине: 'Менее 4 КП с достоверными СИ давления'"
)
_apply_allure_markers(
config.faulty_absence_min_pressure_sensors_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_pressure_sensors_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Неисправна\n "
"Ожидаемая причина режима работы СОУ: Менее 4 КП с достоверными СИ давления "
),
)
test_data = config.faulty_absence_min_pressure_sensors_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_degradation_temperature_du_2_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина: Отказ СИ температуры
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" "Причина: 'Отказ СИ температуры'.\n" "ДУ 2."
_apply_allure_markers(
config.degradation_temperature_du_2_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.degradation_temperature_du_2_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ температуры.\n"
"ДУ Тихорецкая-Нововеличковская."
),
)
test_data = config.degradation_temperature_du_2_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_degradation_density_du_2_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина: Отказ СИ плотности
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" "Причина: Отказ СИ плотности.\n" "ДУ 2."
_apply_allure_markers(
config.degradation_density_du_2_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.degradation_density_du_2_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: СИ плотности.\n"
"ДУ 2."
),
)
test_data = config.degradation_density_du_2_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_degradation_viscosity_du_2_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина: Отказ СИ вязкости
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" "Причина: Отказ СИ вязкости.\n" "ДУ 2."
_apply_allure_markers(
config.degradation_viscosity_du_2_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.degradation_viscosity_du_2_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ вязкости.\n"
"ДУ Тихорецкая-Нововеличковская."
),
)
test_data = config.degradation_viscosity_du_2_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_degradation_temperature_du_3_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина: Отказ СИ температуры
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" "Причина: Отказ СИ температуры.\n" "ДУ 3."
_apply_allure_markers(
config.degradation_temperature_du_3_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.degradation_temperature_du_3_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ температуры.\n"
"ДУ Нововеличковская-Крымская."
),
)
test_data = config.degradation_temperature_sensor_du_3_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_degradation_density_du_3_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина: Отказ СИ плотности
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" "Причина: Отказ СИ плотности.\n" "ДУ 3."
_apply_allure_markers(
config.degradation_density_du_3_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.degradation_density_du_3_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ плотности.\n"
"ДУ Нововеличковская-Крымская."
),
)
test_data = config.degradation_density_du_3_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_degradation_viscosity_du_3_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина: Отказ СИ вязкости
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" "Причина: Отказ СИ вязкости.\n" "ДУ 3."
_apply_allure_markers(
config.degradation_viscosity_du_3_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.degradation_viscosity_du_3_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: Отказ СИ вязкости.\n"
"ДУ Нововеличковская-Крымская."
),
)
test_data = config.degradation_viscosity_du_3_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_between_si_pressure_more_50_km_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина:'Расстояние между ближайшими исправными СИ давления на пути перекачки более 50 км',
"""
tag = "MessagesInfo"
title = (
f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n"
f"Причина: Расстояние между ближайшими исправными СИ давления на пути перекачки более 50 км"
)
_apply_allure_markers(
config.between_si_pressure_more_50_km_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.between_si_pressure_more_50_km_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: "
"Расстояние между ближайшими исправными СИ давления на пути перекачки более 50 км"
),
)
test_data = config.between_si_pressure_more_50_km_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_deg_gravity_section_pumping_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина:'Наличие самотечного участка/участка с неполным сечением',
"""
tag = "MessagesInfo"
title = (
f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n"
f"Причина: Наличие самотечного участка/участка с неполным сечением"
)
_apply_allure_markers(
config.deg_gravity_section_pumping_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_gravity_section_pumping_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: Наличие самотечного участка/участка с неполным сечением"
),
)
test_data = config.deg_gravity_section_pumping_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_deg_absence_min_pressure_sensors_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
Причина: Менее 4-х исправных СИ давления,
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" f"Причина: Менее 4-х исправных СИ давления"
_apply_allure_markers(
config.deg_absence_min_pressure_sensors_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_absence_min_pressure_sensors_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: Менее 4-х исправных СИ давления"
),
)
test_data = config.deg_absence_min_pressure_sensors_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_deg_starting_pumping_out_pumps_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" f"Причина: Работа насосов откачки"
_apply_allure_markers(
config.deg_starting_pumping_out_pumps_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_starting_pumping_out_pumps_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение\n "
"Ожидаемая причина режима работы СОУ: Работа насосов откачки"
),
)
test_data = config.deg_starting_pumping_out_pumps_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_faulty_absence_min_flow_meters_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: неисправна.\n" f"Причина: Недостоверность граничного СИ расхода "
_apply_allure_markers(
config.faulty_absence_min_flow_meters_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.faulty_absence_min_flow_meters_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: неисправна\n "
"Ожидаемая причина режима работы СОУ: Недостоверность граничного СИ расхода "
),
)
test_data = config.faulty_absence_min_flow_meters_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_deg_exceeding_distance_between_flow_meters_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" f"Причина: Расстояние между СИ расхода более 200 км"
_apply_allure_markers(
config.deg_exceeding_distance_between_flow_meters_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_exceeding_distance_between_flow_meters_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: Расстояние между СИ расхода более 200 км"
),
)
test_data = config.deg_exceeding_distance_between_flow_meters_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_deg_additive_injectors_operation_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" f"Причина: Наличие ПТП"
_apply_allure_markers(
config.deg_additive_injectors_operation_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_additive_injectors_operation_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик.\n "
"Ожидаемая причина режима работы СОУ: Наличие ПТП"
),
)
test_data = config.deg_additive_injectors_operation_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
@pytest.mark.asyncio
async def test_deg_pig_sensor_passage_in_journal(
self,
http_client: StandHttpClient,
config: LDSStatusConfig,
) -> None:
"""
[MessagesInfo] Проверка режима работы СОУ в журнале: Ухудшение
"""
tag = "MessagesInfo"
title = f"[{tag}] ЭФ Журнал. Режим СОУ: Ухудшение.\n" f"Причина: прохождение СОД"
_apply_allure_markers(
config.deg_pig_sensor_passage_in_journal_test,
tag,
title,
(
f"Проверка записи в журнале о режиме работы СОУ, на наборе данных {config.suite_name}, \n"
f"на технологическом участке {config.tu_name}\n"
"Время проведения проверки : "
f"{config.deg_pig_sensor_passage_in_journal_test.offset} мин.\n"
"Синхронный запрос типа: MessagesInfo с фильтром messageTypes=LDS_STATUS\n"
"Ожидаемый режим работы СОУ: Ухудшение характеристик\n "
"Ожидаемая причина режима работы СОУ: прохождение СОД "
),
)
test_data = config.deg_pig_sensor_passage_in_journal_test_data
scenarios.lds_status_in_journal(http_client, config, test_data)
конфтест
import asyncio
import glob
import os
import shutil
import threading
import time
import allure
import pytest
import pytest_asyncio
from clients.testops_client import AllureResultsUploader, logger
from constants.architecture_constants import ImitatorConstants as ImConst
from constants.enums import RejectionSensorTag
from constants.test_constants import BaseTN3Constants
from infra.stand_setup_manager import StandSetupManager
from test_config.datasets import get_config_by_name
from test_scenarios import lds_configurator_scenarios
from utils.helpers import lds_configurator_utils as lds_cfg_utils
from utils.helpers.pytest_auth import (
clear_suite_auth,
ensure_auth_for_fixture,
ensure_suite_auth,
init_http_stand_client,
init_ws_stand_client,
)
from utils.helpers.ws_message_parser import ws_message_parser as lds_ws_parser
def pytest_addoption(parser):
"""
Добавляет кастомные опции командной строки pytest.
"""
parser.addoption(
"--suites",
action="store",
default=None,
help="Запустить только указанные наборы данных. Пример: --suites=select_4,select_19_20",
)
def _find_config_by_suite_name(suite_name: str):
"""Находит конфиг по имени набора данных."""
try:
return get_config_by_name(suite_name)
except ValueError:
return None
@pytest.fixture(autouse=True)
def allure_suite_hierarchy(request):
"""
Автоматически устанавливает иерархию Allure для группировки тестов по наборам данных.
В Allure отчёте тесты группируются:
- Parent Suite: SingleLeakSuite / MultiLeakSuite (тип набора)
- Suite: select_4 / select_6 / ... (имя набора данных)
Работает как с параметризованными тестами (config в параметрах),
так и с обычными тестами (через маркер test_suite_name).
"""
config = None
suite_name = None
# Пробуем получить конфиг из параметризации
if hasattr(request, 'fixturenames') and 'config' in request.fixturenames:
try:
config = request.getfixturevalue('config')
suite_name = config.suite_name
except Exception:
pass
# Если не нашли, пробуем найти конфиг по маркеру test_suite_name
if not config:
marker = request.node.get_closest_marker('test_suite_name')
if marker:
suite_name = marker.args[0]
config = _find_config_by_suite_name(suite_name)
if config and suite_name:
parent_suite = "MultiLeakSuite" if config.has_multiple_leaks else "SingleLeakSuite"
allure.dynamic.parent_suite(parent_suite)
allure.dynamic.suite(suite_name)
def pytest_configure(config):
"""
Храним состояние сессии
"""
config.addinivalue_line("markers", "critical_stop: если тест упал, останавливаем дальнейшее выполнение сессии")
config.group_state = {
"current_suite": None,
"suite_start_time": None,
"stand_manager": None,
"imitator_start_time": None, # datetime объект времени старта имитатора для расчёта интервалов утечек
"is_target_tu_active": False,
"use_lds_configurator": True,
"resolved_tu_id": None,
"admin_tu_name": None,
"pre_run_running_tus": None,
"suite_infra_ready": False,
"suite_setup_failure": None,
"stand_host": None,
"auth_token": None,
"x_user_id": None,
"auth_suite": None,
}
def _update_sensor_ids(stand_manager: StandSetupManager) -> None:
"""
Для тестов датчиков обновляет sensor_id по address из конфигурации стенда.
"""
sensor_ids_by_address = stand_manager.get_sensor_ids_by_address()
BaseTN3Constants.SENSOR_IDS_BY_ADDRESS.update(sensor_ids_by_address)
RejectionSensorTag.update_ids_from_config(sensor_ids_by_address)
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
"""
Делает падение критического теста с маркировкой critical_stop однозначным:
- рисуем fail для теста
- после него прекращаем запуск остальных тестов
"""
outcome = yield
report = outcome.get_result()
if report.when == "call" and report.failed and item.get_closest_marker("critical_stop"):
item.session.shouldstop = f"Критическая проверка упала: {item.nodeid}"
# ===== Маппинг имён тестов на атрибуты конфига для получения маркеров =====
# Используется для добавления offset и test_case_id маркеров во время сбора тестов
# Smoke-тесты уровня набора (маркеры из SmokeSuiteConfig)
SMOKE_SUITE_LEVEL_MAPPING = {
'test_basic_info': 'basic_info_test',
'test_journal_info': 'journal_info_test',
'test_imitate_pressure_sensor_signal': 'imitate_pressure_sensor_signal_test',
'test_imitate_flowmeter_signal': 'imitate_flowmeter_signal_test',
'test_lds_status_initialization': 'lds_status_initialization_test',
'test_lds_status_init_in_journal': 'lds_status_init_in_journal_test',
'test_main_page_info': 'main_page_info_test',
'test_main_page_info_signals': 'main_page_info_signals_test',
'test_mask_signal': 'mask_signal_test',
'test_mask_info_in_journal': 'mask_info_in_journal_test',
'test_lds_status_initialization_out': 'lds_status_initialization_out_test',
'test_lds_status_init_out_in_journal': 'lds_status_init_out_in_journal_test',
'test_main_page_info_unstationary': 'main_page_info_unstationary_test',
'test_mask_du_on_mini_scheme': 'mask_du_on_mini_scheme_test',
'test_unmask_du_on_mini_scheme': 'unmask_du_on_mini_scheme_test',
'test_lds_status_after_confirming_leak': 'lds_status_after_confirming_leak_test',
'test_lds_status_completed_leak': 'lds_status_completed_leak_test',
'test_diagnostics_of_signals_after_initialization': 'diagnostics_of_signals_after_initialization_test',
'test_mode_mt_in_journal': 'mode_mt_in_journal_test',
'test_export_lds_status_report': 'export_lds_status_report_test',
'test_export_mt_mode_report': 'export_mt_mode_report_test',
}
# Regress-тесты режимов СОУ (маркеры из LDSStatusConfig)
LDS_STATUS_SUITE_LEVEL_MAPPING = {
'test_lds_status_basic_info': 'lds_status_basic_info_test',
'test_lds_status_init_accumulation_data': 'init_accumulation_data_test',
'test_lds_status_init_accumulation_data_in_journal': 'init_accumulation_data_in_journal_test',
'test_lds_status_init_cold_start': 'init_cold_start_test',
'test_lds_status_init_cold_start_in_journal': 'init_cold_start_in_journal_test',
'test_lds_status_init_exiting_faulty': 'init_exiting_faulty_test',
'test_lds_status_init_switching_shut_off': 'init_switching_shut_off_test',
'test_lds_status_init_switching_shut_off_in_journal': 'init_switching_shut_off_in_journal_test',
'test_lds_status_serviceable_after_cold_start': 'serviceable_after_cold_start_test',
'test_lds_status_serviceable_after_cold_start_in_journal': 'serviceable_after_cold_start_in_journal_test',
'test_lds_status_serviceable_after_switching_shut_off': 'serviceable_after_switching_shut_off_test',
'test_lds_status_serviceable_after_switching_shut_off_in_journal': 'serviceable_after_switching_shut_off_in_journal_test', # noqa: E501
'test_lds_status_serviceable_after_deg_absence_min_pressure_sensors': 'serviceable_after_deg_absence_min_pressure_sensors_test', # noqa: E501
'test_lds_status_serviceable_after_deg_additive_injectors_operation': 'serviceable_after_deg_additive_injectors_operation_test', # noqa: E501
'test_lds_status_serviceable_after_deg_exceeding_distance_between_flow_meters': 'serviceable_after_deg_exceeding_distance_between_flow_meters_test', # noqa: E501
'test_lds_status_serviceable_after_deg_starting_pumping_out_pumps': 'serviceable_after_deg_starting_pumping_out_pumps_test', # noqa: E501
'test_lds_status_serviceable_after_deg_faulty_pressure_sensors_at_pump': 'serviceable_after_deg_faulty_pressure_sensors_at_pump_test', # noqa: E501
'test_lds_status_serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal': 'serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test', # noqa: E501
'test_lds_status_serviceable_after_faulty': 'serviceable_after_faulty_test',
'test_lds_status_degradation_additive_injectors_operation': 'deg_additive_injectors_operation_test',
'test_lds_status_degradation_exceeding_distance_between_pressure_sensors': 'deg_exceeding_distance_between_pressure_sensors_test', # noqa: E501
'test_lds_status_degradation_exceeding_distance_between_pressure_sensors_in_journal': 'deg_exceeding_distance_between_pressure_sensors_in_journal_test', # noqa: E501
'test_lds_status_degradation_absence_min_pressure_sensors': 'deg_absence_min_pressure_sensors_test',
'test_lds_status_degradation_faulty_pressure_sensors_at_pump_station': 'deg_faulty_pressure_sensors_at_pump_station_test', # noqa: E501
'test_lds_status_degradation_faulty_pressure_sensors_at_pump_station_in_journal': 'deg_faulty_pressure_sensors_at_pump_station_in_journal_test', # noqa: E501
'test_lds_status_degradation_gravity_section_pumping': 'deg_gravity_section_pumping_test',
'test_lds_status_degradation_gravity_section_pumping_in_stopping': 'deg_gravity_section_pumping_in_stopping_test',
'test_lds_status_degradation_pig_sensor_passage': 'deg_pig_sensor_passage_test',
'test_lds_status_degradation_starting_pumping_out_pumps': 'deg_starting_pumping_out_pumps_test',
'test_lds_status_degradation_exceeding_distance_between_flow_meters': 'deg_exceeding_distance_between_flow_meters_test', # noqa: E501
# ===== ТЕСТЫ на схеме=====
'test_lds_status_degradation_rejection_temperature_sensor_on_du_2': 'deg_rejection_temperature_sensor_on_du_2_test',
'test_lds_status_degradation_rejection_temperature_sensor_on_du_3': 'deg_rejection_temperature_sensor_on_du_3_test',
'test_lds_status_degradation_rejection_temperature_sensor_on_du_5': 'deg_rejection_temperature_sensor_on_du_5_test',
'test_lds_status_degradation_rejection_density_and_viscosity_on_du_2': 'deg_rejection_density_and_viscosity_on_du_2_test', # noqa: E501
'test_lds_status_degradation_rejection_density_and_viscosity_on_du_3': 'deg_rejection_density_and_viscosity_on_du_3_test', # noqa: E501
'test_lds_status_degradation_rejection_density_and_viscosity_on_du_5': 'deg_rejection_density_and_viscosity_on_du_5_test', # noqa: E501
# ===== ТЕСТЫ в журнале бики ДУ2=====
'test_degradation_temperature_du_2_in_journal': 'degradation_temperature_du_2_in_journal_test',
'test_degradation_density_du_2_in_journal': 'degradation_density_du_2_in_journal_test',
'test_degradation_viscosity_du_2_in_journal': 'degradation_viscosity_du_2_in_journal_test',
# ===== ТЕСТЫ в журнале бики ДУ3=====
'test_degradation_temperature_du_3_in_journal': 'degradation_temperature_du_3_in_journal_test',
'test_degradation_density_du_3_in_journal': 'degradation_density_du_3_in_journal_test',
'test_degradation_viscosity_du_3_in_journal': 'degradation_viscosity_du_3_in_journal_test',
# ===== ТЕСТЫ на схеме течение=====
'test_lds_status_degradation_gravity_section_pumping_in_stopping_in_journal': 'deg_gravity_section_pumping_in_stopping_in_journal_test', # noqa: E501
'test_between_si_pressure_more_50_km_in_journal': 'between_si_pressure_more_50_km_in_journal_test',
'test_deg_gravity_section_pumping_in_journal': 'deg_gravity_section_pumping_in_journal_test',
'test_deg_absence_min_pressure_sensors_in_journal': 'deg_absence_min_pressure_sensors_in_journal_test',
'test_deg_starting_pumping_out_pumps_in_journal': 'deg_starting_pumping_out_pumps_in_journal_test',
'test_faulty_absence_min_flow_meters_in_journal': 'faulty_absence_min_flow_meters_in_journal_test',
'test_faulty_absence_min_flow_meters_continuous_in_journal': 'faulty_absence_min_flow_meters_continuous_in_journal_test', # noqa: E501
'test_deg_exceeding_distance_between_flow_meters_in_journal': 'deg_exceeding_distance_between_flow_meters_in_journal_test', # noqa: E501
'test_deg_pig_sensor_passage_in_journal': 'deg_pig_sensor_passage_in_journal_test',
'test_deg_additive_injectors_operation_in_journal': 'deg_additive_injectors_operation_in_journal_test',
'test_lds_status_faulty_absence_min_pressure_sensors_in_journal': 'faulty_absence_min_pressure_sensors_in_journal_test', # noqa: E501
# ===== ТЕСТЫ на схеме течение=====
'test_lds_status_faulty_absence_min_flow_meters_continuous': 'faulty_absence_min_flow_meters_continuous_test',
'test_lds_status_faulty_absence_min_flow_meters': 'faulty_absence_min_flow_meters_test',
'test_lds_status_faulty_absence_min_pressure_sensors': 'faulty_absence_min_pressure_sensors_test',
}
# Тесты уровня утечки (маркеры из LeakTestConfig - параметр leak)
LEAK_LEVEL_TEST_MAPPING = {
'test_leaks_content': 'leaks_content_test',
'test_all_leaks_info': 'all_leaks_info_test',
'test_tu_leaks_info': 'tu_leaks_info_test',
'test_leak_info_in_journal': 'leak_info_in_journal',
'test_possible_leak_in_journal': 'possible_leak_in_journal_test',
'test_acknowledge_leak_info': 'acknowledge_leak_test',
'test_acknowledge_leak_in_journal': 'acknowledge_leak_in_journal_test',
'test_output_signals': 'output_signals_test',
'test_lds_status_during_leak': 'lds_status_during_leak_test',
'test_balance_algorithm_leak_waiting': 'balance_algorithm_leak_waiting_test',
'test_balance_algorithm_leak_detected': 'balance_algorithm_leak_detected_test',
'test_the_leak_is_complete_on_kg': 'the_leak_is_complete_on_kg_test',
'test_leak_is_complete_in_output_signals': 'leak_is_complete_in_output_signals_test',
'test_complete_tu_leaks_info_content': 'complete_tu_leaks_info_content_test',
'test_all_leaks_is_empty': 'all_leaks_is_empty_test',
'test_leak_is_confirm_on_main_page': 'leak_is_confirm_on_main_page_test',
'test_leak_is_complete_on_main_page': 'leak_is_complete_on_main_page_test',
'test_balance_algorithm_leak_completed': 'balance_algorithm_leak_completed_test',
'test_completed_leak_info_in_journal': 'completed_leak_info_in_journal_test',
'test_export_leaks_report': 'export_leaks_report_test',
}
STATIONARY_STATUS_SUITE_LEVEL_MAPPING = {
'test_stationary_status_basic_info': 'stationary_status_basic_info_test',
'test_stationary_status_common_scheme': 'stationary_status_common_scheme_test',
'test_stationary_status_journal': 'stationary_status_journal_test',
'test_stationary_status_main_page_info': 'stationary_status_main_page_info_test',
}
# Тесты уровня отбраковки (маркеры из RejectionTestCase - параметр rejection_case)
IS_REJECTED_LEVEL_TEST_MAPPING = {
'test_rejection_input_signals': 'rejection_input_signals_test',
'test_rejection_journal': 'rejection_journal_test',
'test_rejection_main_page': 'rejection_main_page_test',
'test_rejection_scheme_signals_state': 'rejection_scheme_signals_state_test',
}
# Suite-level тесты отбраковки (маркеры из IsRejectedConfig - параметр config)
IS_REJECTED_SUITE_LEVEL_MAPPING = {
'test_rejection_report': 'rejection_report_test',
}
# Мержим все вместе чтобы не переписывать логику коллектора айтемов (тестов)
SUITE_LEVEL_TEST_MAPPING = {
**SMOKE_SUITE_LEVEL_MAPPING,
**LDS_STATUS_SUITE_LEVEL_MAPPING,
**STATIONARY_STATUS_SUITE_LEVEL_MAPPING,
**IS_REJECTED_SUITE_LEVEL_MAPPING,
}
def _get_test_markers_config(item, test_name):
"""
Получает конфигурацию с маркерами для теста.
Для leak-level тестов: маркеры берутся из параметра leak
Для suite-level тестов: маркеры берутся из config
:return: CaseMarkers объект или None
"""
if not hasattr(item, 'callspec'):
return None
params = item.callspec.params
# Проверяем, есть ли параметр leak (для leak-level тестов)
if 'leak' in params and test_name in LEAK_LEVEL_TEST_MAPPING:
leak = params['leak']
attr_name = LEAK_LEVEL_TEST_MAPPING[test_name]
return getattr(leak, attr_name, None)
# Проверяем, есть ли параметр rejection_case для тестов отбраковки
if 'rejection_case' in params and test_name in IS_REJECTED_LEVEL_TEST_MAPPING:
rejection_case = params['rejection_case']
attr_name = IS_REJECTED_LEVEL_TEST_MAPPING[test_name]
return getattr(rejection_case, attr_name, None)
# Suite-level тесты отбраковки (без rejection_case)
if 'config' in params and 'rejection_case' not in params and test_name in IS_REJECTED_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = IS_REJECTED_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
# Для suite-level тестов берём из config
if 'config' in params:
if test_name in SMOKE_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = SMOKE_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
if test_name in LDS_STATUS_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = LDS_STATUS_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
if test_name in STATIONARY_STATUS_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = STATIONARY_STATUS_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
return None
def pytest_collection_modifyitems(session, config, items):
"""
1. Фильтрует тесты по --suites (если указано)
2. Исключает тесты, у которых конфиг = None (тест отключён для этого набора данных)
3. Добавляет маркеры offset и test_case_id из конфига к каждому параметризованному тесту
4. Сортирует тесты по test_suite_name для группировки по наборам данных
"""
# Получаем список выбранных наборов из --suites
suites_option = config.getoption("--suites")
selected_suites = None
if suites_option:
# Парсим список наборов: "select_4,select_19_20" -> ["select_4", "select_19_20"]
selected_suites = [s.strip().lower() for s in suites_option.split(",")]
selected_items = []
deselected_items = []
for item in items:
# Фильтрация по --suites
if selected_suites:
suite_marker = item.get_closest_marker("test_suite_name")
if suite_marker:
suite_name = suite_marker.args[0].lower()
# Проверяем, содержит ли имя набора одну из выбранных подстрок
if not any(selected in suite_name for selected in selected_suites):
deselected_items.append(item)
continue
# Получаем имя функции теста (без параметров)
test_name = item.originalname or item.name.split('[')[0]
# Получаем конфиг с маркерами для теста
test_config = _get_test_markers_config(item, test_name)
if test_config is not None:
# Добавляем маркер offset
if hasattr(test_config, 'offset') and test_config.offset is not None:
item.add_marker(pytest.mark.offset(test_config.offset))
# Добавляем маркер test_case_id
if hasattr(test_config, 'test_case_id') and test_config.test_case_id is not None:
item.add_marker(pytest.mark.test_case_id(test_config.test_case_id))
elif (
test_name in SUITE_LEVEL_TEST_MAPPING
or test_name in LEAK_LEVEL_TEST_MAPPING
or test_name in IS_REJECTED_LEVEL_TEST_MAPPING
):
# Конфиг теста = None - исключаем тест из прогона
deselected_items.append(item)
continue
selected_items.append(item)
# Уведомляем pytest об исключённых тестах
if deselected_items:
config.hook.pytest_deselected(items=deselected_items)
# Заменяем список тестов на отфильтрованный
items[:] = selected_items
# Сортировка тестов по test_suite_name и offset
# Цель: обеспечить запуск тестов строго по offset строго внутри набора данных
# При равных offset сохраняем исходный порядок коллекции, чтобы порядок параметризации не перескакивал
def suite_offset_key(item):
"""
Сортировка тестов по test_suite_name и offset (без падения на None).
"""
test_suite_name_marker = item.get_closest_marker("test_suite_name")
test_suite_name = test_suite_name_marker.args[0] if test_suite_name_marker else ""
offset_marker = item.get_closest_marker("offset")
if offset_marker:
try:
offset_value = float(offset_marker.args[0])
except Exception:
offset_value = float("inf")
else:
offset_value = float("inf")
original_index = getattr(item, "_collection_index", 0)
# Возвращаем тройку ключей сортировки
# 1) test_suite_name - группировка по набору
# 2) offset_value - порядок внутри набора по времени
# 3) original_index - стабильность при равных offset
return test_suite_name, offset_value, original_index
# Сохраняем исходный порядок коллекции для стабильной сортировки
for index, item in enumerate(items):
item._collection_index = index
# по кортежу питон сортирует слева направо, благодаря этому сортировка по offset идет строго внутри test_suite_name
items.sort(key=suite_offset_key)
for item in items:
if hasattr(item, "_collection_index"):
delattr(item, "_collection_index")
@pytest.fixture(autouse=True)
def allure_tms_link(request):
"""
Allure TMS‑линки по test_case_id
"""
if test_case_id_marker := request.node.get_closest_marker("test_case_id"):
test_case_id = test_case_id_marker.args[0]
allure.dynamic.link(
f"https://{os.environ['TESTOPS_BASE_URL']}/testcases?selected_id={test_case_id}",
name=f"TestCase-{test_case_id}",
link_type="tms",
)
@pytest.fixture(autouse=True)
def require_suite_infra(request):
"""
Пропускает тесты набора, если infra-setup не завершился успешно.
"""
if not request.node.get_closest_marker("test_suite_name"):
return
cfg = request.config.group_state
if cfg.get("current_suite") and not cfg.get("suite_infra_ready"):
pytest.skip("[SETUP] [ERROR] Набор пропущен: инфраструктура не готова")
@pytest.fixture(autouse=True)
def offset_wait(request):
"""
Offset‑ожидание перед каждым тестом относительно фактического старта core
"""
cfg = request.config.group_state
if cfg.get("current_suite") and not cfg.get("suite_infra_ready"):
return
if offset_marker := request.node.get_closest_marker("offset"):
offset_sec = float(offset_marker.args[0]) * BaseTN3Constants.SEC_PER_MIN
start = request.config.group_state["suite_start_time"] or 0
elapsed = time.monotonic() - start
to_wait = max(0, offset_sec - elapsed)
if to_wait:
time.sleep(to_wait)
def compute_imitator_duration(item, current_test_suite: str) -> float:
"""
Вычисляет длительность для имитатора (в минутах).
Правило:
- Собирает все тесты (item.session.items) с меткой test_suite_name == current_test_suite
- Извлекает все значения @pytest.mark.offset(...) (в минутах)
- Если offsets найдены: возвращает max(offsets) + IMITATOR_FINISH_DELAY задержка остановки имитатора
- Иначе: если у текущего item есть @pytest.mark.imitator_duration — используется как fallback и логируется
- Если ничего не найдено — pytest.fail с понятным текстом
"""
suite_items = [
suite_item
for suite_item in item.session.items
if (marker := suite_item.get_closest_marker("test_suite_name")) and marker.args[0] == current_test_suite
]
offsets = []
for suite_item in suite_items:
offset_marker = suite_item.get_closest_marker("offset")
if offset_marker:
try:
offsets.append(float(offset_marker.args[0]))
except Exception:
continue
if offsets:
max_offset = max(offsets)
imitator_duration = float(max_offset) + ImConst.IMITATOR_FINISH_DELAY_MINUTE
return imitator_duration
else:
# fallback- если все еще задан старый маркер imitator_duration, то используем его
if imitator_mark := item.get_closest_marker("imitator_duration"):
imitator_duration = float(imitator_mark.args[0])
logger.warning(
"[DEPRECATED] использован pytest.mark.imitator_duration()"
f"рекомендуется убрать и полагаться на max_offset + {ImConst.IMITATOR_FINISH_DELAY_MINUTE}"
)
return imitator_duration
pytest.fail(
"Не удалось вычислить imitator_duration: в тестовом модуле одновременно отсутствуют "
"и @pytest.mark.offset(), и pytest.mark.imitator_duration()"
)
def _skip_current_suite_after_setup_failure(cfg: dict, message: str) -> None:
"""
Пропускает текущий набор после ошибки setup: cleanup частичной инфраструктуры и pytest.skip.
"""
logger.error(message)
try:
allure.attach(message, name="Ошибка setup набора", attachment_type=allure.attachment_type.TEXT)
except Exception:
logger.debug("Не удалось прикрепить ошибку setup к Allure", exc_info=True)
logger.info("[TEARDOWN] LDS Configurator очистка после ошибки setup набора")
_run_lds_configurator_teardown_if_needed(cfg)
if stand_manager := cfg.get("stand_manager"):
try:
stand_manager.stop_imitator_wrapper()
except Exception:
logger.exception("[SETUP] Ошибка остановки имитатора после неудачного setup набора")
stand_manager.server_test_data_remover()
cfg["suite_infra_ready"] = False
cfg["suite_setup_failure"] = message
cfg["suite_start_time"] = None
pytest.skip("Набор пропущен: ошибка подготовки инфраструктуры")
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_setup(item):
"""
Перезапуск имитатора при смене test_suite_name
"""
cfg = item.config.group_state
test_suite_marker = item.get_closest_marker("test_suite_name")
if not test_suite_marker:
pytest.fail("Тест без @pytest.mark.test_suite_name")
current_test_suite = test_suite_marker.args[0]
if current_test_suite != cfg["current_suite"]:
# stop old
if stand_manager := cfg["stand_manager"]:
try:
stand_manager.stop_imitator_wrapper()
try:
stand_manager.restore_signal_unit_conversion_rules()
except Exception:
logger.exception(
"[ERROR] [SETUP] Ошибка при восстановлении signal_unit_conversion_rules.json "
"перед запуском нового набора"
)
stand_manager.server_test_data_remover()
finally:
cfg["stand_manager"] = None
# start new
cfg["current_suite"] = current_test_suite
cfg["suite_start_time"] = None
cfg["suite_infra_ready"] = False
cfg["suite_setup_failure"] = None
clear_suite_auth(cfg)
data_id = item.get_closest_marker("test_suite_data_id").args[0]
test_data_name = item.get_closest_marker("test_data_name").args[0]
# legacy: id из enum TU для имитатора (tn{id}_tags.txt), не resolved_tu_id из Администрирования
tu_id = item.get_closest_marker("tu_id").args[0]
ost_name = item.get_closest_marker("ost_name").args[0]
imitator_duration = compute_imitator_duration(item, current_test_suite)
suite_config = _find_config_by_suite_name(current_test_suite)
measure_conversion_rules = suite_config.measure_conversion_rules if suite_config is not None else None
if suite_config is not None and suite_config.use_lds_configurator:
if not suite_config.admin_tu_name.strip():
pytest.exit(
f"[SETUP] [ERROR] Набор '{suite_config.suite_name}': admin_tu_name обязателен "
"при use_lds_configurator=True"
)
stand_manager = StandSetupManager(
duration_m=imitator_duration,
test_data_id=data_id,
test_data_name=test_data_name,
tu_id=tu_id,
ost_name=ost_name,
measure_conversion_rules=measure_conversion_rules,
)
cfg["stand_manager"] = stand_manager
try:
stand_manager.check_opc_server_status()
except RuntimeError as error:
msg = (
"[SETUP] [ERROR] OPC сервер недоступен. Имитатор и автотесты не запущены. "
f"Ошибка при проверке статуса OPC: {error}"
)
allure.attach(msg, name="OPC сервер недоступен", attachment_type=allure.attachment_type.TEXT)
_skip_current_suite_after_setup_failure(cfg, msg)
try:
stand_manager.setup_stand_for_imitator_run()
except Exception as error:
_skip_current_suite_after_setup_failure(cfg, f"[SETUP] [ERROR] ошибка при подготовке стенда: {error}")
try:
ensure_suite_auth(cfg, current_test_suite)
except BaseException as error:
_skip_current_suite_after_setup_failure(cfg, f"[SETUP] [ERROR] не удалось инициализировать auth: {error}")
try:
_update_sensor_ids(stand_manager)
except Exception as error:
_skip_current_suite_after_setup_failure(
cfg,
f"[SETUP] [ERROR] ошибка обновления id датчиков отбраковки из конфигурации: {error}",
)
if suite_config is not None and suite_config.use_lds_configurator:
try:
_run_lds_admin_setup(suite_config, cfg)
except BaseException as error:
_skip_current_suite_after_setup_failure(cfg, f"[SETUP] [ERROR] LDS Configurator admin setup: {error}")
imitator_thread = threading.Thread(
target=stand_manager.start_imitator, name=f"imitator->{current_test_suite}", daemon=True
)
core_thread = threading.Thread(target=stand_manager.start_core)
try:
imitator_thread.start()
except Exception as error:
_skip_current_suite_after_setup_failure(cfg, f"[SETUP] [ERROR] ошибка запуска имитатора: {error}")
time.sleep(ImConst.CORE_START_DELAY_S)
try:
cfg["suite_start_time"] = time.monotonic()
core_thread.start()
core_thread.join(timeout=5)
except Exception as error:
_skip_current_suite_after_setup_failure(cfg, f"[SETUP] [ERROR] ошибка запуска СORE контейнеров: {error}")
# Сохраняем время старта имитатора для расчёта интервалов утечек в тестах
cfg["imitator_start_time"] = stand_manager.start_time
if suite_config is not None and suite_config.use_lds_configurator:
if suite_config.admin_tu is None:
_skip_current_suite_after_setup_failure(
cfg,
f"[SETUP] [ERROR] Набор '{suite_config.suite_name}': admin_tu обязателен "
"при use_lds_configurator=True",
)
try:
_run_lds_verify_after_core(suite_config, cfg)
except BaseException as error:
_skip_current_suite_after_setup_failure(
cfg, f"[SETUP] [ERROR] LDS Configurator проверка после запуска ядра: {error}"
)
cfg["suite_infra_ready"] = True
yield # pytest продолжит выполнение теста
def _run_lds_configurator_ws(coro_factory) -> None:
"""Запускает async WS-сценарий lds-configurator"""
lds_cfg_utils.set_configurator_flow_active(True)
lds_ws_parser.suppress_recv_logging = True
try:
asyncio.run(coro_factory())
finally:
lds_ws_parser.suppress_recv_logging = False
lds_cfg_utils.set_configurator_flow_active(False)
def _run_lds_admin_setup(suite_config, group_state: dict) -> None:
"""
Setup СОУ через Администрирование до старта имитатора.
"""
async def _admin_setup() -> None:
http_client = init_http_stand_client(group_state)
http_client.suppress_recv_logging = True
await lds_configurator_scenarios.lds_configurator_admin_setup(http_client, suite_config, group_state)
_run_lds_configurator_ws(_admin_setup)
def _run_lds_verify_after_core(suite_config, group_state: dict) -> None:
"""
WS-проверка готовности стенда после запуска lds-core.
"""
async def _verify() -> None:
websocket_client = init_ws_stand_client(group_state)
http_client = init_http_stand_client(group_state)
http_client.suppress_recv_logging = True
async with websocket_client as ws_client:
ws_client.suppress_recv_logging = True
await lds_configurator_scenarios.lds_configurator_verify_after_core(ws_client, http_client, suite_config)
_run_lds_configurator_ws(_verify)
def _run_lds_configurator_teardown_if_needed(cfg: dict) -> None:
"""
Выполняет WS-teardown СОУ при смене набора или завершении сессии.
Ошибки не прерывают pytest: логируются и оформляются как ALERT внутри сценария.
Состояние group_state сбрасывается в finally.
"""
if not cfg.get("use_lds_configurator") or not cfg.get("resolved_tu_id"):
return
tu_id = cfg["resolved_tu_id"]
admin_tu_name = cfg.get("admin_tu_name") or ""
pre_run_running_tus = cfg.get("pre_run_running_tus") or []
async def _teardown() -> None:
http_client = init_http_stand_client(cfg)
http_client.suppress_recv_logging = True
await lds_configurator_scenarios.lds_configurator_teardown(
http_client, tu_id, admin_tu_name, cfg, pre_run_running_tus
)
try:
_run_lds_configurator_ws(_teardown)
except BaseException as error:
logger.warning(
"[TEARDOWN] [ALERT] LDS Configurator teardown завершился с ошибкой для tuId=%s: %s",
tu_id,
error,
)
finally:
cfg["resolved_tu_id"] = None
cfg["use_lds_configurator"] = False
cfg["is_target_tu_active"] = False
cfg["admin_tu_name"] = None
cfg["pre_run_running_tus"] = None
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_teardown(item, nextitem):
"""
Teardown имитатора при выходе из группы
"""
yield
cfg = item.config.group_state
next_marker = nextitem.get_closest_marker("test_suite_name") if nextitem else None
next_suite = next_marker.args[0] if next_marker else None
if next_suite != cfg["current_suite"]:
if stand_manager := cfg["stand_manager"]:
try:
if cfg.get("suite_infra_ready"):
stand_manager.stop_imitator_wrapper()
try:
stand_manager.restore_signal_unit_conversion_rules()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при восстановлении signal_unit_conversion_rules.json")
stand_manager.server_test_data_remover()
finally:
cfg["stand_manager"] = None
cfg["current_suite"] = None
cfg["suite_start_time"] = None
cfg["imitator_start_time"] = None
cfg["suite_infra_ready"] = False
# опционально дождаться завершения потока (если не daemon) — безопасный join
imitator_thread = cfg.get("imitator_thread")
if imitator_thread and not getattr(imitator_thread, "daemon", False):
try:
imitator_thread.join(timeout=5)
except RuntimeError:
logger.exception("Ошибка при join() фона имитатора")
@pytest_asyncio.fixture
async def ws_client(request):
"""
Фикстура для работы с websocket клиентом.
request - даёт доступ к request.config.group_state (общий кэш auth на dataset: stand_host, auth_token, x_user_id).
:return: Объект wss соединения
"""
cfg = request.config.group_state
ensure_auth_for_fixture(cfg)
ws_client = init_ws_stand_client(cfg)
async with ws_client as client:
yield client
@pytest.fixture
def http_client(request):
"""
Фикстура для работы с http клиентом.
request - даёт доступ к request.config.group_state (общий кэш auth на dataset: stand_host, auth_token).
:return: экземпляр класса для выполнения http запросов к стенду
"""
cfg = request.config.group_state
ensure_auth_for_fixture(cfg)
return init_http_stand_client(cfg)
@pytest.fixture
def imitator_start_time(request):
"""
Фикстура для получения времени старта имитатора (datetime объект).
Используется для точного расчёта времени обнаружения утечек:
- leak_start_time = imitator_start_time + timedelta(seconds=LEAK_START_INTERVAL)
- leak_end_time = imitator_start_time + timedelta(seconds=LEAK_START_INTERVAL + ALLOWED_TIME_DIFF_SECONDS)
"""
start_time = request.config.group_state.get("imitator_start_time")
if start_time is None:
pytest.fail("imitator_start_time не установлен. Убедитесь что тест запущен после инициализации имитатора.")
return start_time
def pytest_sessionfinish(session, exitstatus):
"""
В завершении сессии — отправляем единый Allure‑отчёт в TestOps.
"""
# 1) teardown стенда: LDS Configurator + остановка имитатора
try:
group_state = getattr(session.config, "group_state", {})
_run_lds_configurator_teardown_if_needed(group_state)
stand_manager = group_state.get("stand_manager")
if stand_manager:
try:
stand_manager.stop_imitator_wrapper()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при остановке имитатора")
try:
stand_manager.restore_signal_unit_conversion_rules()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при восстановлении signal_unit_conversion_rules.json")
try:
stand_manager.server_test_data_remover()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при удалении тестового набора данных со стенда")
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при получении stand_manager из group_state")
# 2) Выгрузка allure-results в TestOps
try:
uploader = AllureResultsUploader()
logger.info("[INFO] [TEARDOWN] Uploading Allure results to TestOps")
uploader.upload_allure_results()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при выгрузке allure-results в TestOps")
# 3) Удаление локальных архивов с данными
shutil.rmtree("allure-results")
project_root = os.path.dirname(os.path.abspath(__file__))
files_for_drop = glob.glob(os.path.join(project_root, "*.tar.gz"))
if not files_for_drop:
logger.warning("[WARNING] [TEARDOWN] Не нашлось архивов .tar.gz с данным для удаления")
else:
for file in files_for_drop:
os.remove(file)
