https://pastein.ru/t/SQ8
Загрузка данных
сцен
async def export_leaks_report(ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time: datetime):
"""
Сценарий формирования отчёта об утечках.
Этапы:
1. Подписка SubscribeReportsDataExportedRequest на пуш-нотификации.
2. Отправка ExportReportsCommandRequest с фильтром по времени
(start = старт имитатора, end = старт имитатора + offset теста).
3. Ожидание пуш-нотификации ReportDataExportedNotification о готовности отчёта.
4. Лонг-поллинг GetExportedDataListRequest до появления нашего отчёта в списке.
5. Отправка DownloadExportedDataRequest по id отчёта.
6. Получение fileChunk по ответу на скачивание.
7-10. Проверки: формат файла, имя, шапка xlsx, строка утечки.
Скачанный файл удаляется по завершению, прикладывается к Allure только при падении теста.
"""
actual_report_state = ExportLeaksReportState()
with allure.step("Подготовка параметров сценария формирования отчёта об утечках"):
actual_report_state.report_test = leak.export_leaks_report_test
StepCheck("В конфигурации задан export_leaks_report_test", "export_leaks_report_test").actual(
actual_report_state.report_test
).is_not_none()
actual_report_state.period_start = t_utils.localize_as_moscow(imitator_start_time)
actual_report_state.period_end = t_utils.localize_as_moscow(
imitator_start_time + timedelta(minutes=actual_report_state.report_test.offset)
)
actual_report_state.period_start_naive = report_utils.normalize_report_period_naive(
actual_report_state.period_start
)
actual_report_state.period_end_naive = report_utils.normalize_report_period_naive(
actual_report_state.period_end
)
actual_report_state.expected_mt_mode = ReportConst.STATIONARY_STATUS_TO_REPORT_TEXT.get(
leak.expected_stationary_status
)
actual_report_state.expected_lds_status_text = LdsStatus.report_text_by_value(
leak.expected_lds_status_in_leaks_report
)
actual_report_state.tu_description_lower = cfg.technological_unit.description.lower()
time_offset_hours = t_utils.report_time_offset_hours()
StepCheck(
f"Смещение timeOffset для запросов отчёта (часовой пояс {TestConst.ZONE_INFO})",
"time_offset_hours",
).actual(time_offset_hours).is_not_none()
actual_report_state.time_offset_hours = time_offset_hours
StepCheck(
"Задан ожидаемый текст режима МТ для отчёта",
"expected_mt_mode",
).actual(actual_report_state.expected_mt_mode).is_not_none()
allure.attach(
f"period.start={actual_report_state.period_start}\n"
f"period.end={actual_report_state.period_end}\n"
f"offset_minutes={actual_report_state.report_test.offset}",
name="Фильтр периода отчёта",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step(f"Этап 1. Подписка на пуш-нотификации ({ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST})"):
await t_utils.connect(ws_client, ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST, [])
with allure.step(f"Этап 2. Запрос формирования отчёта ({ReportConst.EXPORT_REPORTS_COMMAND_REQUEST})"):
request_payload = {
"tuId": cfg.tu_id,
"exportedDataTypes": [ExportedDataType.LEAKS_REPORT.value],
"timeOffset": actual_report_state.time_offset_hours,
"period": {
"start": t_utils.datetime_to_msgpack_timestamp(actual_report_state.period_start),
"end": t_utils.datetime_to_msgpack_timestamp(actual_report_state.period_end),
"additionalProperties": {},
},
}
await t_utils.connect(ws_client, ReportConst.EXPORT_REPORTS_COMMAND_REQUEST, request_payload)
with allure.step(
f"Этап 3. Ожидание пуш-нотификации {ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION} о готовности отчёта"
):
actual_report_state.notification = await t_utils.poll_for_report_export_notification(
ws_client=ws_client,
parser=parser,
total_wait_seconds=ReportConst.NOTIFICATION_TIMEOUT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step("Извлечение полей пуш-нотификации"):
notification = actual_report_state.notification
notification_reply_status = notification.replyStatus if notification else None
notification_reply_content = notification.replyContent if notification else None
notification_export_status = notification_reply_content.exportStatus if notification_reply_content else None
notification_error_message = (
(notification_reply_content.errorMessage or "") if notification_reply_content else ""
)
with allure.step(f"Этап 4. Лонг-поллинг {ReportConst.GET_EXPORTED_DATA_LIST_REQUEST} до появления отчёта в списке"):
actual_report_state.report_item = await t_utils.poll_for_exported_file(
ws_client=ws_client,
parser=parser,
list_limit=ReportConst.EXPORTED_DATA_LIST_LIMIT,
expected_data_type=ExportedDataType.LEAKS_REPORT,
name_substring=ReportConst.LEAKS_REPORT_NAME_PART,
tu_name_substring=cfg.technological_unit.description,
period_start=actual_report_state.period_start,
period_end=actual_report_state.period_end,
total_wait_seconds=ReportConst.LIST_POLL_TOTAL_WAIT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step("Подготовка данных найденного отчёта в списке"):
report_item = actual_report_state.report_item
if report_item is not None:
allure.attach(
f"id={report_item.id}, name={report_item.name}, "
f"exportedDataType={report_item.exportedDataType}, "
f"start={t_utils.format_datetime_moscow(report_item.start)}, "
f"end={t_utils.format_datetime_moscow(report_item.end)}",
name="Найденный отчёт в списке",
attachment_type=allure.attachment_type.TEXT,
)
actual_report_state.report_file_name = report_utils.build_export_report_file_name(
cfg.technological_unit.description,
actual_report_state.period_start,
actual_report_state.period_end,
)
with allure.step("Проверка: отчёт найден в списке сформированных файлов"):
StepCheck("Отчёт найден в списке сформированных файлов", "report_item").actual(
actual_report_state.report_item
).is_not_none()
with allure.step(
f"Этап 5. Streaming-вызов {ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST} по "
f"id={actual_report_state.report_item.id}"
):
download_request = {
"exportedDataId": actual_report_state.report_item.id,
"exportedDataType": ExportedDataType.LEAKS_REPORT.to_download_name(),
"additionalProperties": None,
"timeOffset": actual_report_state.time_offset_hours,
}
download_purpose = (
f"скачивание xlsx-отчёта об утечках (exportedDataId={actual_report_state.report_item.id}) "
f"после формирования отчёта и выбора файла в списке GetExportedDataListRequest - "
f"выпадашка уведомлений на UI"
)
await t_utils.connect_stream(
ws_client,
ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
download_request,
purpose=download_purpose,
)
actual_report_state.download_invocation_id = ws_client.invocation_id
with allure.step("Этап 6. Получение fileChunk - скачивание отчёта по утечкам"):
actual_report_state.download_reply = await t_utils.receive_download_exported_data_reply(
ws_client=ws_client,
parser=parser,
invocation_id=actual_report_state.download_invocation_id,
request_name=ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
total_wait_seconds=ReportConst.DOWNLOAD_TIMEOUT_SECONDS,
purpose=download_purpose,
)
with allure.step("Извлечение данных ответа на скачивание"):
download_reply = actual_report_state.download_reply
download_reply_status = download_reply.replyStatus
has_download_reply_content = download_reply.replyContent is not None
actual_report_state.file_bytes = download_reply.replyContent.fileChunk if has_download_reply_content else None
is_xlsx_signature = (
report_utils.is_xlsx_file_bytes(actual_report_state.file_bytes) if actual_report_state.file_bytes else False
)
with allure.step("Проверка ответа на скачивание и формата xlsx"):
StepCheck("Проверка статуса ответа на скачивание", "replyStatus").actual(download_reply_status).expected(
ReplyStatus.OK.value
).equal_to()
StepCheck("Проверка наличия контента ответа на скачивание", "replyContent").actual(
has_download_reply_content
).expected(True).equal_to()
StepCheck("Проверка наличия байт файла", "fileChunk").actual(actual_report_state.file_bytes).is_not_empty()
StepCheck("Проверка xlsx (zip) сигнатуры файла", "file_signature").actual(is_xlsx_signature).expected(
True
).equal_to()
with allure.step("Подготовка данных для проверки имени файла отчёта"):
report_file_name = actual_report_state.report_file_name
report_file_name_lower = report_file_name.lower()
file_name_period_start, file_name_period_end = report_utils.parse_period_from_export_file_name(report_file_name)
period_start_lo, period_start_hi, period_end_lo, period_end_hi = report_utils.report_period_comparison_bounds(
actual_report_state.period_start_naive,
actual_report_state.period_end_naive,
)
has_xlsx_extension = report_utils.is_xlsx_extension(report_file_name)
leaks_report_name_part_lower = ReportConst.LEAKS_REPORT_NAME_PART.lower()
with allure.step("Этап 8. Сохранение, обработка и проверка отчета по утечкам"):
actual_report_state.temp_file_path = report_utils.save_report_bytes_to_temp_file(actual_report_state.file_bytes)
try:
with allure.step("Проверка: временный xlsx файл создан"):
StepCheck("Временный xlsx файл создан", "temp_file_path").actual(
actual_report_state.temp_file_path
).is_not_none()
with allure.step("Этап 9. Открытие xlsx и чтение шапки"):
actual_report_state.worksheet = report_utils.load_report_worksheet(actual_report_state.temp_file_path)
actual_report_state.title_info = report_utils.parse_report_title(
report_utils.get_report_title_cell(actual_report_state.worksheet)
)
allure.attach(
f"Шапка отчёта (raw): {actual_report_state.title_info.raw_title}\n"
f"period_start: {actual_report_state.title_info.period_start}\n"
f"period_end: {actual_report_state.title_info.period_end}",
name="Шапка отчёта (1-я строка)",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных шапки отчёта для проверки"):
title_info = actual_report_state.title_info
report_title_lower = title_info.raw_title.lower()
leaks_report_name_part_lower = ReportConst.LEAKS_REPORT_NAME_PART.lower()
column_headers = report_utils.get_report_column_headers(actual_report_state.worksheet)
period_start_lo, period_start_hi, period_end_lo, period_end_hi = (
report_utils.report_period_comparison_bounds(
actual_report_state.period_start_naive,
actual_report_state.period_end_naive,
)
)
header_period_start = title_info.period_start
header_period_end = title_info.period_end
with allure.step("Этап 10. Извлечение строк данных из отчёта"):
actual_report_state.data_rows = report_utils.iter_report_data_rows(actual_report_state.worksheet)
actual_report_state.target_row = report_utils.find_row_with_object(
actual_report_state.data_rows, cfg.technological_unit.description
)
allure.attach(
"\n".join(f"row#{row.row_index}: {row.cells}" for row in actual_report_state.data_rows),
name="Все строки данных отчёта",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных строки утечки для проверки"):
target_row = actual_report_state.target_row
leak_datetime_value = target_row.datetime_value if target_row else None
object_value_lower = target_row.object_value.lower() if target_row else ""
lds_status_value = target_row.lds_status.strip() if target_row else ""
masking_info_lower = target_row.masking_info.lower() if target_row else ""
leak_coordinate_meters = target_row.coordinate_meters if target_row else None
leak_volume_value = target_row.leak_volume if target_row else None
mt_mode_lower = target_row.mt_mode.lower() if target_row else ""
expected_mt_mode_lower = actual_report_state.expected_mt_mode.lower()
expected_lds_status_lower = actual_report_state.expected_lds_status_text.lower()
masking_not_masked_lower = ReportConst.MASKING_NOT_MASKED_TEXT.lower()
period_start_lo, period_start_hi, period_end_lo, period_end_hi = (
report_utils.report_period_comparison_bounds(
actual_report_state.period_start_naive,
actual_report_state.period_end_naive,
)
)
with allure.step("Проверка содержимого строки утечки"):
StepCheck("В отчёте есть хотя бы одна строка с данными", "data_rows").actual(
actual_report_state.data_rows
).is_not_empty()
StepCheck(
f"Строка с объектом, содержащим '{cfg.technological_unit.description}'",
ReportConst.COL_OBJECT,
).actual(actual_report_state.target_row).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
"Время утечки в диапазоне [старт имитатора, старт + offset теста] (+-1 мин)",
ReportConst.COL_DATETIME,
soft_failures,
).actual(leak_datetime_value).is_between(period_start_lo, period_end_hi)
StepCheck(
f"Колонка '{ReportConst.COL_OBJECT}' содержит '{cfg.technological_unit.description}'",
ReportConst.COL_OBJECT,
soft_failures,
).contains(object_value_lower, actual_report_state.tu_description_lower)
StepCheck(
f"Колонка '{ReportConst.COL_LDS_STATUS}' содержит "
f"'{actual_report_state.expected_lds_status_text}'",
ReportConst.COL_LDS_STATUS,
soft_failures,
).contains(lds_status_value.lower(), expected_lds_status_lower)
StepCheck(
f"Колонка '{ReportConst.COL_MASK_INFO}' содержит '{ReportConst.MASKING_NOT_MASKED_TEXT}'",
ReportConst.COL_MASK_INFO,
soft_failures,
).contains(masking_info_lower, masking_not_masked_lower)
StepCheck(
f"Колонка '{ReportConst.COL_COORDINATE}' (с погрешностью {cfg.allowed_distance_diff_meters} м)",
ReportConst.COL_COORDINATE,
soft_failures,
).actual(leak_coordinate_meters).is_close_to(
leak.coordinate_meters,
cfg.allowed_distance_diff_meters,
f"значение допустимой погрешности координаты {cfg.allowed_distance_diff_meters}",
)
StepCheck(
f"Колонка '{ReportConst.COL_LEAK_VOLUME}' не пустая",
ReportConst.COL_LEAK_VOLUME,
soft_failures,
).actual(leak_volume_value).is_not_none()
StepCheck(
f"Колонка '{ReportConst.COL_MT_MODE}' содержит '{actual_report_state.expected_mt_mode}'",
ReportConst.COL_MT_MODE,
soft_failures,
).contains(mt_mode_lower, expected_mt_mode_lower)
except Exception:
with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
if actual_report_state.temp_file_path and actual_report_state.report_file_name:
report_utils.attach_report_file_to_allure(
actual_report_state.temp_file_path, actual_report_state.report_file_name
)
raise
finally:
with allure.step("Удаление временного xlsx файла"):
temp_path = actual_report_state.temp_file_path
if temp_path is not None:
try:
temp_path.unlink(missing_ok=True)
except OSError:
pass
with allure.step("Проверка имени файла отчёта"):
with SoftAssertions() as soft_failures:
StepCheck(f"Имя файла оканчивается на {ReportConst.XLSX_EXTENSION}", "file_name", soft_failures).actual(
has_xlsx_extension
).expected(True).equal_to()
StepCheck(
f"Имя файла содержит '{ReportConst.LEAKS_REPORT_NAME_PART}'", "file_name", soft_failures
).contains(report_file_name_lower, leaks_report_name_part_lower)
StepCheck(
f"Имя файла содержит описание ТУ '{cfg.technological_unit.description}'", "file_name", soft_failures
).contains(report_file_name_lower, actual_report_state.tu_description_lower)
StepCheck(
"Дата начала периода в имени файла совпадает с фильтром запроса (+-1 мин)",
"period_start_in_file_name",
soft_failures,
).actual(file_name_period_start).is_between(period_start_lo, period_start_hi)
StepCheck(
"Дата конца периода в имени файла совпадает с фильтром запроса (+-1 мин)",
"period_end_in_file_name",
soft_failures,
).actual(file_name_period_end).is_between(period_end_lo, period_end_hi)
with allure.step("Проверка двойной шапки отчёта"):
StepCheck("Лист xlsx открыт", "worksheet").actual(actual_report_state.worksheet).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
f"В шапке отчёта присутствует '{ReportConst.LEAKS_REPORT_NAME_PART}'",
"report_title",
soft_failures,
).contains(report_title_lower, leaks_report_name_part_lower)
StepCheck(
"Время начала периода в шапке совпадает с фильтром запроса (+-1 мин)",
"period_start",
soft_failures,
).actual(header_period_start).is_between(period_start_lo, period_start_hi)
StepCheck(
"Время конца периода в шапке совпадает с фильтром запроса (+-1 мин)",
"period_end",
soft_failures,
).actual(header_period_end).is_between(period_end_lo, period_end_hi)
StepCheck(
"Названия колонок в шапке отчёта",
"column_headers",
soft_failures,
).actual(
column_headers
).expected(ReportConst.EXPECTED_COLUMN_HEADERS).equal_to()
with allure.step("Проверка пуш-нотификации о готовности отчёта"):
with SoftAssertions() as soft_failures:
StepCheck("Получена пуш-нотификация о готовности отчёта", "notification", soft_failures).actual(
actual_report_state.notification
).is_not_none()
StepCheck("Проверка статуса пуш-нотификации", "replyStatus", soft_failures).actual(
notification_reply_status
).expected(ReplyStatus.OK.value).equal_to()
StepCheck("Проверка наличия контента нотификации", "replyContent", soft_failures).actual(
notification_reply_content
).is_not_none()
StepCheck("Проверка exportStatus в нотификации", "exportStatus", soft_failures).actual(
notification_export_status
).expected(ExportStatus.DONE).equal_to()
StepCheck("В нотификации нет текста ошибки", "errorMessage", soft_failures).actual(
notification_error_message
).is_empty()
enum
class LdsStatus(Enum):
"""
Режим работы СОУ.
report_text - значение колонки 'Режим работы СОУ' в xlsx-отчёте об утечках.
"""
FAULTY = (1, "СОУ неисправна")
INITIALIZATION = (2, "СОУ в инициализации")
DEGRADATION = (3, "СОУ в ухудшенных характеристиках")
SERVICEABLE = (4, "СОУ исправна")
def __new__(cls, value: int, report_text: str) -> "LdsStatus":
member = object.__new__(cls)
member._value_ = value
member.report_text = report_text
return member
@classmethod
def report_text_by_value(cls, status_value: int) -> str | None:
"""Текст режима СОУ для отчёта по числовому значению статуса"""
try:
return cls(status_value).report_text
except ValueError:
return None
select 6
"""
Конфигурация тестового набора Select_6_tn3_56km_113
Особенности набора:
- Режим стационара (StationaryStatus.STATIONARY)
- Одна утечка на координате 56 км
- Объём утечки 113.6 м³
"""
from constants.enums import TU, ConfirmationStatus, LdsStatus, LdsStatusInitialization, ReservedType, StationaryStatus
from test_config.models_for_tests import (
CaseData,
CaseMarkers,
DiagnosticAreaStatusConfig,
LeakTestConfig,
SmokeSuiteConfig,
)
# ===== Константы набора =====
SUITE_NAME = "Select_6_tn3_56km_113"
SUITE_DATA_ID = 4
ARCHIVE_NAME = f"{SUITE_NAME}.tar.gz"
# Технологический участок
TECHNOLOGICAL_UNIT = TU.TIKHORETSK_NOVOROSSIYSK_3
# Название МН
MAIN_PIPELINE = "МН Тихорецк-Новороссийск-3"
# Параметры утечки
LEAK_COORDINATE_METERS = 56000.0
LEAK_VOLUME_M3 = 113.6
ALLOWED_TIME_DIFF_SECONDS = 1440 # 24 минуты
LEAK_START_INTERVAL_SECONDS = 2100 # 35 минут
LEAK_TECHNOLOGICAL_OBJECT = "НПС-5 Тихорецкая - НПС-3 Нововеличковская"
FLOW_RATE_SETTINGS_THRESHOLD = 17
# ID диагностических участков
LEAK_DIAGNOSTIC_AREA_ID = 2
LEAK_DIAGNOSTIC_AREA_NAME = "Т-Н-3.НПС-5 «Тихорецкая».УЗР вых - Т-Н-3.УЗР НПС-3 «Нововеличковская»."
# ID труб для определения ДУ
DIAGNOSTIC_AREA_2_PIPE_ID = 1463 # Труба на ДУ с утечкой
DIAGNOSTIC_AREA_3_PIPE_ID = 1444 # OUT_NEIGHBOR_DIAGNOSTIC_AREA_PIPE_ID
# ID линейного участка
LINEAR_PART_ID = 407
# ===== Конфигурация набора =====
SELECT_6_CONFIG = SmokeSuiteConfig(
# ===== Метаданные =====
suite_name=SUITE_NAME,
suite_data_id=SUITE_DATA_ID,
archive_name=ARCHIVE_NAME,
technological_unit=TECHNOLOGICAL_UNIT,
main_pipeline=MAIN_PIPELINE,
# ===== Ожидаемый статус стационара =====
expected_stationary_status=StationaryStatus.STATIONARY.value,
# ----- Ожидаемый статус СОУ -----
lds_status_after_confirming_leak_data=CaseData(
params={"pipe_id": DIAGNOSTIC_AREA_2_PIPE_ID},
expected_result=(
LdsStatus.INITIALIZATION.value,
LdsStatusInitialization.ACCUMULATION_DATA.value,
),
),
# ----- Ожидаемые статусы для проверки режимов на ЭФ Диагностика сигналов -----
exp_tixoreczkaya_novovelichkovskaya_reg_lu=StationaryStatus.STATIONARY.value,
exp_tixoreczkaya_novovelichkovskaya_reg_sou=LdsStatus.SERVICEABLE.value,
exp_novovelichkovskaya_krymskaya_reg_lu=StationaryStatus.STATIONARY.value,
exp_novovelichkovskaya_krymskaya_reg_sou=LdsStatus.SERVICEABLE.value,
exp_krymskaya_grushovaya_reg_lu=StationaryStatus.STATIONARY.value,
exp_krymskaya_grushovaya_reg_sou=LdsStatus.DEGRADATION.value,
exp_backup_route_bejsug_reg_lu=StationaryStatus.STOPPED.value,
exp_backup_route_bejsug_reg_sou=LdsStatus.FAULTY.value,
exp_backup_route_ponura_reg_lu=StationaryStatus.STOPPED.value,
exp_backup_route_ponura_reg_sou=LdsStatus.FAULTY.value,
exp_backup_route_kuban_reg_lu=StationaryStatus.STOPPED.value,
exp_backup_route_kuban_reg_sou=LdsStatus.FAULTY.value,
exp_npz_afipskij_reg_lu=StationaryStatus.STOPPED.value,
exp_npz_afipskij_reg_sou=LdsStatus.FAULTY.value,
exp_npz_ilinskij_reg_lu=StationaryStatus.STOPPED.value,
exp_npz_ilinskij_reg_sou=LdsStatus.FAULTY.value,
# ===== БАЗОВЫЕ ТЕСТЫ =====
basic_info_test=CaseMarkers(test_case_id="1", offset=5),
journal_info_test=CaseMarkers(test_case_id="2", offset=5),
lds_status_initialization_test=CaseMarkers(test_case_id="29", offset=5),
lds_status_init_in_journal_test=CaseMarkers(test_case_id="", offset=5),
main_page_info_test=CaseMarkers(test_case_id="3", offset=6),
mask_signal_test=CaseMarkers(test_case_id="45", offset=8),
mask_info_in_journal_test=CaseMarkers(test_case_id="213", offset=9),
lds_status_initialization_out_test=CaseMarkers(test_case_id="30", offset=30),
lds_status_init_out_in_journal_test=CaseMarkers(test_case_id="214", offset=31),
lds_status_after_confirming_leak_test=CaseMarkers(test_case_id="201", offset=60),
diagnostics_of_signals_after_initialization_test=CaseMarkers(test_case_id="210", offset=25),
# ===== КОНФИГУРАЦИЯ УТЕЧКИ =====
leak=LeakTestConfig(
# ----- Конфигурация статусов СОУ во время утечки -----
lds_status_during_leak_config=DiagnosticAreaStatusConfig(
leak_diagnostic_area_id=LEAK_DIAGNOSTIC_AREA_ID,
leak_diagnostic_area_pipe_id=DIAGNOSTIC_AREA_2_PIPE_ID,
leak_du_expected_lds_status=LdsStatus.INITIALIZATION.value,
out_neighbors={
DIAGNOSTIC_AREA_3_PIPE_ID: LdsStatus.DEGRADATION.value,
},
),
# ----- Параметры утечки -----
coordinate_meters=LEAK_COORDINATE_METERS,
volume_m3=LEAK_VOLUME_M3,
linear_part_id=LINEAR_PART_ID,
technological_object=LEAK_TECHNOLOGICAL_OBJECT,
flow_rate_settings_threshold=FLOW_RATE_SETTINGS_THRESHOLD,
diagnostic_area_name=LEAK_DIAGNOSTIC_AREA_NAME,
# ----- Временные интервалы -----
leak_start_interval_seconds=LEAK_START_INTERVAL_SECONDS,
allowed_time_diff_seconds=ALLOWED_TIME_DIFF_SECONDS,
# ----- Ожидаемые статусы -----
expected_algorithm_type=ReservedType.STATIONARY_FLOW.value,
expected_leak_status=ConfirmationStatus.CONFIRMED.value,
expected_lds_status=LdsStatus.SERVICEABLE.value,
expected_lds_status_in_leaks_report=LdsStatus.SERVICEABLE.value,
expected_stationary_status=StationaryStatus.STATIONARY.value,
# ----- Тест BalanceAlgorithmResultsContent -----
balance_algorithm_leak_waiting_test=CaseMarkers(test_case_id="175", offset=42), # Длительность теста 5 минут
balance_algorithm_leak_detected_test=CaseMarkers(test_case_id="177", offset=59),
# ----- Тест MainPageInfoContent -----
leak_is_confirm_on_main_page_test=CaseMarkers(test_case_id="182", offset=60),
# ----- Тест AllLeaksInfo -----
all_leaks_info_test=CaseMarkers(test_case_id="4", offset=59),
# ----- Тест LeaksContent -----
leaks_content_test=CaseMarkers(test_case_id="97", offset=59),
# ----- Тест MessageInfo -----
leak_info_in_journal=CaseMarkers(test_case_id="119", offset=59),
possible_leak_in_journal_test=CaseMarkers(test_case_id="211", offset=47),
acknowledge_leak_in_journal_test=CaseMarkers(test_case_id="212", offset=60.5),
# ----- Тест TuLeaksInfo -----
tu_leaks_info_test=CaseMarkers(test_case_id="5", offset=59),
# ----- Тест CommonSchemeContent -----
lds_status_during_leak_test=CaseMarkers(test_case_id="31", offset=59.5),
# ----- Тест AcknowledgeLeak -----
acknowledge_leak_test=CaseMarkers(test_case_id="6", offset=60),
# ----- Тест OutputSignals -----
output_signals_test=CaseMarkers(test_case_id="33", offset=61),
# ----- Тест ExportReports -----
export_leaks_report_test=CaseMarkers(test_case_id="234", offset=62),
),
)
в мод тест
expected_mt_mode: Optional[str] = None
expected_lds_status_text: Optional[str] = None
expected_lds_status: int = LdsStatus.SERVICEABLE.value
# Режим СОУ в xlsx export_leaks_report (колонка 'Режим работы СОУ')
expected_lds_status_in_leaks_report: Optional[int] = None
