---

модели тестов 
    title_info: Optional[ReportTitleInfo] = None








константсы

class ExportReportConstants:
    """Константы для теста формирования отчёта об утечках"""

    # ===== Имена WS-сообщений (для шагов allure и connect) =====
    SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST: str = "SubscribeReportsDataExportedRequest"
    EXPORT_REPORTS_COMMAND_REQUEST: str = "ExportReportsCommandRequest"
    REPORT_DATA_EXPORTED_NOTIFICATION: str = "ReportDataExportedNotification"
    GET_EXPORTED_FILES_LIST_REQUEST: str = "getExportedFilesListRequest"
    DOWNLOAD_EXPORTED_DATA_REQUEST: str = "DownloadExportedDataRequest"










export reports model

@dataclass
class SubscribeReportsDataExportedRequest:
    """Подписка на пуш ReportDataExportedNotification. Параметров нет (invoke с args=[[]])."""























сценарий

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. Лонг-поллинг getExportedFilesListRequest до появления нашего отчёта в списке.
    5. Отправка DownloadExportedDataRequest по id отчёта.
    6. Получение fileChunk по ответу на скачивание.
    7-10. Проверки: формат файла, имя, шапка xlsx, строка утечки.

    Скачанный файл удаляется по завершению, прикладывается к Allure только при падении теста.
    """
    report_state = ExportLeaksReportState()

    with allure.step("Подготовка параметров сценария формирования отчёта об утечках"):
        report_state.report_test = leak.export_leaks_report_test
        StepCheck("В конфигурации задан export_leaks_report_test", "export_leaks_report_test").actual(
            report_state.report_test
        ).is_not_none()

        report_state.period_start = t_utils.localize_as_moscow(imitator_start_time)
        report_state.period_end = t_utils.localize_as_moscow(
            imitator_start_time + timedelta(minutes=report_state.report_test.offset)
        )
        report_state.period_start_naive = report_state.period_start.replace(tzinfo=None)
        report_state.period_end_naive = report_state.period_end.replace(tzinfo=None)
        report_state.expected_mt_mode = ReportConst.STATIONARY_STATUS_TO_REPORT_TEXT.get(
            leak.expected_stationary_status
        )
        report_state.tu_description_lower = cfg.technological_unit.description.lower()

        StepCheck(
            "Задан ожидаемый текст режима МТ для отчёта",
            "expected_mt_mode",
        ).actual(report_state.expected_mt_mode).is_not_none()

        allure.attach(
            f"period.start={report_state.period_start}\n"
            f"period.end={report_state.period_end}\n"
            f"offset_minutes={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": ReportConst.MOSCOW_TIME_OFFSET_HOURS,
            "period": {
                "start": t_utils.datetime_to_msgpack_timestamp(report_state.period_start),
                "end": t_utils.datetime_to_msgpack_timestamp(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} о готовности отчёта"
    ):
        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("Проверка пуш-нотификации о готовности отчёта"):
        StepCheck("Получена пуш-нотификация о готовности отчёта", "notification").actual(
            report_state.notification
        ).is_not_none()
        StepCheck("Проверка статуса пуш-нотификации", "replyStatus").actual(
            report_state.notification.replyStatus if report_state.notification else None
        ).expected(ReplyStatus.OK.value).equal_to()
        StepCheck("Проверка наличия контента нотификации", "replyContent").actual(
            report_state.notification.replyContent if report_state.notification else None
        ).is_not_none()
        StepCheck("Проверка exportStatus в нотификации", "exportStatus").actual(
            report_state.notification.replyContent.exportStatus
            if report_state.notification and report_state.notification.replyContent
            else None
        ).expected(ExportStatus.DONE).equal_to()
        StepCheck("Проверка отсутствия ошибки в нотификации", "errorMessage").actual(
            (report_state.notification.replyContent.errorMessage or "")
            if report_state.notification and report_state.notification.replyContent
            else None
        ).expected("").equal_to()

    with allure.step(
        f"Этап 4. Лонг-поллинг {ReportConst.GET_EXPORTED_FILES_LIST_REQUEST} до появления отчёта в списке"
    ):
        report_state.report_item = await t_utils.poll_for_exported_file(
            ws_client=ws_client,
            parser=parser,
            tu_id=cfg.tu_id,
            expected_data_type=ExportedDataType.LEAKS_REPORT,
            name_substring=ReportConst.LEAKS_REPORT_NAME_PART,
            period_start=report_state.period_start,
            period_end=report_state.period_end,
            total_wait_seconds=ReportConst.LIST_POLL_TOTAL_WAIT_SECONDS,
            poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
        )

    with allure.step("Проверка: отчёт найден в списке сформированных файлов"):
        StepCheck("Отчёт найден в списке сформированных файлов", "report_item").actual(
            report_state.report_item
        ).is_not_none()
        report_item = report_state.report_item
        allure.attach(
            f"id={report_item.id}, name={report_item.name}, "
            f"exportedDataType={report_item.exportedDataType}, "
            f"start={report_item.start}, end={report_item.end}",
            name="Найденный отчёт в списке",
            attachment_type=allure.attachment_type.TEXT,
        )
        report_state.report_file_name = report_item.name + ReportConst.XLSX_EXTENSION

    with allure.step(
        f"Этап 5. Отправка {ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST} по id={report_state.report_item.id}"
    ):
        download_request = {
            "exportedDataId": report_state.report_item.id,
            "exportedDataType": ExportedDataType.LEAKS_REPORT.to_download_name(),
            "additionalProperties": None,
            "timeOffset": ReportConst.MOSCOW_TIME_OFFSET_HOURS,
        }
        await t_utils.connect(ws_client, ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST, download_request)
        report_state.download_invocation_id = ws_client.invocation_id

    with allure.step("Этап 6. Получение fileChunk - скачивание отчета по утечкам"):
        try:
            report_state.download_payload = await ws_client.receive_by_invocation_id(
                report_state.download_invocation_id, timeout=ReportConst.DOWNLOAD_TIMEOUT_SECONDS
            )
        except (TimeoutError, OSError) as error:
            pytest.fail(
                f"Не получили ответ на {ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST} "
                f"за {ReportConst.DOWNLOAD_TIMEOUT_SECONDS} секунд. Ошибка: {error}"
            )
        report_state.download_reply = parser.parse_download_exported_data_msg(report_state.download_payload)

    with allure.step("Проверка ответа на скачивание и формата xlsx"):
        download_reply = report_state.download_reply
        StepCheck("Проверка статуса ответа на скачивание", "replyStatus").actual(download_reply.replyStatus).expected(
            ReplyStatus.OK.value
        ).equal_to()
        StepCheck("Проверка наличия контента ответа на скачивание", "replyContent").actual(
            download_reply.replyContent
        ).is_not_none()
        report_state.file_bytes = download_reply.replyContent.fileChunk
        StepCheck("Проверка наличия байт файла", "fileChunk").actual(report_state.file_bytes).is_not_empty()
        StepCheck("Проверка xlsx (zip) сигнатуры файла", "file_signature").actual(
            report_utils.is_xlsx_file_bytes(report_state.file_bytes)
        ).expected(True).equal_to()

    with allure.step("Проверка имени файла отчёта"):
        report_file_name_lower = report_state.report_file_name.lower()
        StepCheck(f"Имя файла оканчивается на {ReportConst.XLSX_EXTENSION}", "file_name").actual(
            report_utils.is_xlsx_extension(report_state.report_file_name)
        ).expected(True).equal_to()
        StepCheck(f"Имя файла содержит '{ReportConst.LEAKS_REPORT_NAME_PART}'", "file_name").actual(
            ReportConst.LEAKS_REPORT_NAME_PART.lower() in report_file_name_lower
        ).expected(True).equal_to()
        StepCheck(
            f"Имя файла содержит описание ТУ '{cfg.technological_unit.description}'",
            "file_name",
        ).actual(
            report_state.tu_description_lower in report_file_name_lower
        ).expected(True).equal_to()

    with allure.step("Этап 8. Сохранение, обработка и проверка отчета по утечкам"):
        report_state.temp_file_path = report_utils.save_report_bytes_to_temp_file(report_state.file_bytes)

    try:
        with allure.step("Проверка: временный xlsx файл создан"):
            StepCheck("Временный xlsx файл создан", "temp_file_path").actual(report_state.temp_file_path).is_not_none()

        with allure.step("Этап 9. Открытие xlsx и чтение шапки"):
            report_state.worksheet = report_utils.load_report_worksheet(report_state.temp_file_path)
            report_state.title_info = report_utils.parse_report_title(
                report_utils.get_report_title_cell(report_state.worksheet)
            )
            allure.attach(
                f"Шапка отчёта (raw): {report_state.title_info.raw_title}\n"
                f"period_start: {report_state.title_info.period_start}\n"
                f"period_end: {report_state.title_info.period_end}",
                name="Шапка отчёта (1-я строка)",
                attachment_type=allure.attachment_type.TEXT,
            )

        with allure.step("Проверка двойной шапки отчёта"):
            title_info = report_state.title_info
            StepCheck("Лист xlsx открыт", "worksheet").actual(report_state.worksheet).is_not_none()
            with SoftAssertions() as soft_failures:
                StepCheck(
                    f"В шапке отчёта присутствует '{ReportConst.LEAKS_REPORT_NAME_PART}'",
                    "report_title",
                    soft_failures,
                ).actual(ReportConst.LEAKS_REPORT_NAME_PART.lower() in title_info.raw_title.lower()).expected(
                    True
                ).equal_to()

                StepCheck(
                    "Время начала периода в шапке совпадает с фильтром запроса",
                    "period_start",
                    soft_failures,
                ).actual(title_info.period_start).expected(report_state.period_start_naive).equal_to()
                StepCheck(
                    "Время конца периода в шапке совпадает с фильтром запроса",
                    "period_end",
                    soft_failures,
                ).actual(title_info.period_end).expected(report_state.period_end_naive).equal_to()
                StepCheck(
                    "Названия колонок в шапке отчёта",
                    "column_headers",
                    soft_failures,
                ).actual(
                    report_utils.get_report_column_headers(report_state.worksheet)
                ).expected(ReportConst.EXPECTED_COLUMN_HEADERS).equal_to()

        with allure.step("Этап 10. Извлечение строк данных из отчёта"):
            report_state.data_rows = report_utils.iter_report_data_rows(report_state.worksheet)
            report_state.target_row = report_utils.find_row_with_object(
                report_state.data_rows, cfg.technological_unit.description
            )
            allure.attach(
                "\n".join(f"row#{row.row_index}: {row.cells}" for row in report_state.data_rows),
                name="Все строки данных отчёта",
                attachment_type=allure.attachment_type.TEXT,
            )

        with allure.step("Проверка содержимого строки утечки"):
            StepCheck("В отчёте есть хотя бы одна строка с данными", "data_rows").actual(
                report_state.data_rows
            ).is_not_empty()
            StepCheck(
                f"Строка с объектом, содержащим '{cfg.technological_unit.description}'",
                ReportConst.COL_OBJECT,
            ).actual(report_state.target_row).is_not_none()

            target_row = report_state.target_row
            with SoftAssertions() as soft_failures:
                StepCheck(
                    "Время утечки в диапазоне [старт имитатора, старт + offset теста]",
                    ReportConst.COL_DATETIME,
                    soft_failures,
                ).actual(target_row.datetime_value).is_between(
                    report_state.period_start_naive, report_state.period_end_naive
                )

                StepCheck(
                    f"Колонка '{ReportConst.COL_OBJECT}' содержит " f"'{cfg.technological_unit.description}'",
                    ReportConst.COL_OBJECT,
                    soft_failures,
                ).actual(report_state.tu_description_lower in target_row.object_value.lower()).expected(True).equal_to()

                StepCheck(
                    f"Колонка '{ReportConst.COL_LDS_STATUS}'",
                    ReportConst.COL_LDS_STATUS,
                    soft_failures,
                ).actual(
                    target_row.lds_status.strip()
                ).expected(ReportConst.LDS_STATUS_OK_TEXT).equal_to()

                StepCheck(
                    f"Колонка '{ReportConst.COL_MASK_INFO}' содержит " f"'{ReportConst.MASKING_NOT_MASKED_TEXT}'",
                    ReportConst.COL_MASK_INFO,
                    soft_failures,
                ).actual(ReportConst.MASKING_NOT_MASKED_TEXT.lower() in target_row.masking_info.lower()).expected(
                    True
                ).equal_to()

                StepCheck(
                    f"Колонка '{ReportConst.COL_COORDINATE}' (с погрешностью " f"{cfg.allowed_distance_diff_meters} м)",
                    ReportConst.COL_COORDINATE,
                    soft_failures,
                ).actual(target_row.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(target_row.leak_volume).is_not_none()

                StepCheck(
                    f"Колонка '{ReportConst.COL_MT_MODE}' содержит '{report_state.expected_mt_mode}'",
                    ReportConst.COL_MT_MODE,
                    soft_failures,
                ).actual(report_state.expected_mt_mode.lower() in target_row.mt_mode.lower()).expected(True).equal_to()
    except Exception:
        with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
            if report_state.temp_file_path and report_state.report_file_name:
                report_utils.attach_report_file_to_allure(report_state.temp_file_path, report_state.report_file_name)
        raise
    finally:
        with allure.step("Удаление временного xlsx файла"):
            temp_path = report_state.temp_file_path
            if temp_path is not None:
                try:
                    temp_path.unlink(missing_ok=True)
                except OSError:
                    pass