Загрузка данных
сцен
async def export_mt_mode_report(
ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time: datetime
):
"""
Сценарий формирования xlsx-отчёта о режиме работы МТ.
Этапы:
1. Подписка SubscribeReportsDataExportedRequest на пуш-нотификации.
2. Отправка ExportReportsCommandRequest (тип StationaryStatusReport, фильтр по периоду).
3. Ожидание ReportDataExportedNotification о готовности отчёта.
4. Лонг-поллинг GetExportedDataListRequest до появления отчёта в списке.
5. DownloadExportedDataRequest и получение fileChunk.
6. Проверка xlsx: участки, длительности режимов МТ, основной режим.
7. Проверка двойной шапки и имени файла.
Скачанный файл удаляется по завершению, прикладывается к Allure только при падении теста.
"""
report_state = ExportMtModeReportState()
with allure.step("Подготовка параметров сценария формирования отчёта о режиме работы МТ"):
report_state.expected_report_test = leak.export_mt_mode_report_test
StepCheck("В конфигурации задан export_mt_mode_report_test", "export_mt_mode_report_test").actual(
report_state.expected_report_test
).is_not_none()
report_state.expected_period_start = t_utils.localize_as_moscow(imitator_start_time)
report_state.expected_period_end = t_utils.localize_as_moscow(
imitator_start_time + timedelta(minutes=report_state.expected_report_test.offset)
)
report_state.expected_period_start_naive = report_utils.normalize_report_period_naive(
report_state.expected_period_start
)
report_state.expected_period_end_naive = report_utils.normalize_report_period_naive(
report_state.expected_period_end
)
report_state.expected_tu_description_lower = cfg.technological_unit.description.lower()
report_state.expected_section_names = list(MtReportConst.SECTION_NAMES)
report_state.expected_dominant_mode_column = MtReportConst.STATIONARY_STATUS_TO_COLUMN.get(
leak.expected_report_stationary_status
)
report_state.expected_file_name = report_utils.build_export_report_file_name(
cfg.technological_unit.description,
report_state.expected_period_start,
report_state.expected_period_end,
MtReportConst.MT_MODE_REPORT_NAME_PART,
". ",
)
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()
report_state.actual_time_offset_hours = time_offset_hours
StepCheck(
"Для набора задан ожидаемый доминирующий режим МТ",
"expected_dominant_mode_column",
).actual(report_state.expected_dominant_mode_column).is_not_none()
allure.attach(
f"period.start={report_state.expected_period_start}\n"
f"period.end={report_state.expected_period_end}\n"
f"offset_minutes={report_state.expected_report_test.offset}\n"
f"sections={report_state.expected_section_names}\n"
f"expected_dominant_mode_column={report_state.expected_dominant_mode_column}",
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.STATIONARY_STATUS_REPORT.value],
"timeOffset": report_state.actual_time_offset_hours,
"period": {
"start": t_utils.datetime_to_msgpack_timestamp(report_state.expected_period_start),
"end": t_utils.datetime_to_msgpack_timestamp(report_state.expected_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.actual_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(f"Этап 4. Лонг-поллинг {ReportConst.GET_EXPORTED_DATA_LIST_REQUEST} до появления отчёта в списке"):
report_state.actual_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.STATIONARY_STATUS_REPORT,
name_substring=MtReportConst.MT_MODE_REPORT_NAME_PART,
tu_name_substring=cfg.technological_unit.description,
period_start=report_state.expected_period_start,
period_end=report_state.expected_period_end,
total_wait_seconds=ReportConst.LIST_POLL_TOTAL_WAIT_SECONDS,
poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
)
with allure.step("Подготовка данных найденного отчёта в списке"):
report_item = report_state.actual_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,
)
with allure.step("Проверка: отчёт найден в списке сформированных файлов"):
StepCheck("Отчёт найден в списке сформированных файлов", "report_item").actual(
report_state.actual_report_item
).is_not_none()
with allure.step(
f"Этап 5. Streaming-вызов {ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST} "
f"по id={report_state.actual_report_item.id}"
):
download_request = {
"exportedDataId": report_state.actual_report_item.id,
"exportedDataType": ExportedDataType.STATIONARY_STATUS_REPORT.to_download_name(),
"additionalProperties": None,
"timeOffset": report_state.actual_time_offset_hours,
}
download_purpose = (
f"скачивание xlsx-отчёта о режиме МТ (exportedDataId={report_state.actual_report_item.id}) "
f"после формирования отчёта и выбора файла в списке GetExportedDataListRequest"
)
await t_utils.connect_stream(
ws_client,
ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
download_request,
purpose=download_purpose,
)
report_state.actual_download_invocation_id = ws_client.invocation_id
with allure.step("Этап 6. Получение fileChunk - скачивание отчёта о режиме МТ"):
report_state.actual_download_reply = await t_utils.receive_download_exported_data_reply(
ws_client=ws_client,
parser=parser,
invocation_id=report_state.actual_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 = report_state.actual_download_reply
download_reply_status = download_reply.replyStatus
has_download_reply_content = download_reply.replyContent is not None
report_state.actual_file_bytes = download_reply.replyContent.fileChunk if has_download_reply_content else None
is_xlsx_signature = (
report_utils.is_xlsx_file_bytes(report_state.actual_file_bytes)
if report_state.actual_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(report_state.actual_file_bytes).is_not_empty()
StepCheck("Проверка xlsx (zip) сигнатуры файла", "file_signature").actual(is_xlsx_signature).expected(
True
).equal_to()
with allure.step("Подготовка данных для проверки имени файла отчёта"):
report_file_name = report_state.expected_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,
MtReportConst.REPORT_FILE_NAME_PERIOD_PATTERN,
)
period_start_lo, period_start_hi, period_end_lo, period_end_hi = report_utils.report_period_comparison_bounds(
report_state.expected_period_start_naive,
report_state.expected_period_end_naive,
)
has_xlsx_extension = report_utils.is_xlsx_extension(report_file_name)
mt_report_name_part_lower = MtReportConst.MT_MODE_REPORT_NAME_PART.lower()
try:
with allure.step("Этап 7. Сохранение и разбор xlsx-отчёта о режиме МТ"):
report_state.actual_temp_file_path = report_utils.save_report_bytes_to_temp_file(
report_state.actual_file_bytes,
prefix="mt_mode_report_",
)
StepCheck("Временный xlsx файл создан", "temp_file_path").actual(
report_state.actual_temp_file_path
).is_not_none()
report_state.actual_worksheet = report_utils.load_report_worksheet(report_state.actual_temp_file_path)
report_state.actual_parsed_report = mt_report_utils.parse_mt_mode_report_worksheet(
report_state.actual_worksheet,
report_state.expected_section_names,
)
parsed_report = report_state.actual_parsed_report
allure.attach(
f"Шапка (raw): {parsed_report.title_info.raw_title}\n"
f"period_start: {parsed_report.title_info.period_start}\n"
f"period_end: {parsed_report.title_info.period_end}\n"
f"total_duration: {parsed_report.total_duration_raw}",
name="Шапка отчёта о режиме МТ",
attachment_type=allure.attachment_type.TEXT,
)
allure.attach(
mt_report_utils.format_mt_mode_section_rows_for_allure(parsed_report.section_rows),
name="Строки участков отчёта",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных таблицы отчёта для проверки"):
parsed_report = report_state.actual_parsed_report
section_rows = parsed_report.section_rows
total_duration_seconds = parsed_report.total_duration_seconds
duration_tolerance = MtReportConst.TOTAL_DURATION_TOLERANCE_SECONDS
mode_totals = mt_report_utils.sum_duration_columns_across_rows(
section_rows,
MtReportConst.MODE_DURATION_COLUMNS,
)
with allure.step("Проверка содержимого таблицы отчёта о режиме МТ"):
StepCheck("Лист xlsx открыт", "worksheet").actual(report_state.actual_worksheet).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
"Количество строк участков в отчёте",
"section_rows_count",
soft_failures,
).actual(len(section_rows)).expected(len(report_state.expected_section_names)).equal_to()
for section_index, expected_section_name in enumerate(report_state.expected_section_names):
actual_section_name = (
section_rows[section_index].section_name if section_index < len(section_rows) else None
)
StepCheck(
f"Наименование участка #{section_index + 1}",
MtReportConst.COL_SECTION,
soft_failures,
).actual(actual_section_name).expected(expected_section_name).equal_to()
for section_row in section_rows:
for column_name in MtReportConst.MODE_DURATION_COLUMNS:
cell_value = section_row.cells.get(column_name)
duration_text = (section_row.cells.get(column_name) or "").strip() or MtReportConst.ZERO_DURATION_TEXT
StepCheck(
f"Участок '{section_row.section_name}': колонка '{column_name}'",
"длительность",
soft_failures,
).actual(mt_report_utils.is_duration_cell_filled(cell_value)).is_true_with_details(
expected_text="указано время в формате H:MM:SS (допускается 0:00:00)",
actual_text=duration_text,
)
total_duration_text = (
parsed_report.total_duration_raw
or lds_report_utils.format_duration_seconds(total_duration_seconds or 0)
)
total_label_row = parsed_report.total_label_row_index
StepCheck(
"В отчёте присутствует строка 'Суммарное время работы:'",
"структура отчёта",
soft_failures,
).actual(total_label_row is not None).is_true_with_details(
expected_text=f"найдена строка с текстом '{MtReportConst.TOTAL_WORK_DURATION_LABEL}'",
actual_text=(
f"строка {total_label_row} содержит '{MtReportConst.TOTAL_WORK_DURATION_LABEL}'"
if total_label_row is not None
else "строка не найдена"
),
)
StepCheck(
"Суммарное время работы в отчёте больше нуля",
"суммарное время",
soft_failures,
).actual((total_duration_seconds or 0) > 0).is_true_with_details(
expected_text="суммарное время больше 0:00:00",
actual_text=total_duration_text,
)
for section_row in section_rows:
duration_diff = abs(section_row.modes_sum_seconds - (total_duration_seconds or 0))
section_sum_text = lds_report_utils.format_duration_seconds(section_row.modes_sum_seconds)
diff_text = lds_report_utils.format_duration_seconds(duration_diff)
sums_match = duration_diff < duration_tolerance + 1
StepCheck(
f"Участок '{section_row.section_name}': сумма режимов МТ совпадает с общим временем "
f"(±{duration_tolerance} с)",
"согласованность длительностей",
soft_failures,
).actual(sums_match).is_true_with_details(
expected_text=(
f"сумма режимов ({section_sum_text}) равна суммарному времени ({total_duration_text}), "
f"погрешность не более {duration_tolerance} с"
),
actual_text=(
f"сумма режимов: {section_sum_text}, суммарное время: {total_duration_text}, "
f"разница: {diff_text} ({duration_diff} с)"
),
)
dominant_column = report_state.expected_dominant_mode_column
dominant_total = mode_totals.get(dominant_column, 0)
max_column = max(mode_totals, key=mode_totals.get)
max_total = mode_totals[max_column]
all_modes_text = ", ".join(
f"'{column_name}': {lds_report_utils.format_duration_seconds(total_seconds)}"
for column_name, total_seconds in mode_totals.items()
)
StepCheck(
f"Режим '{dominant_column}' имеет максимальное суммарное время по всем участкам",
"доминирующий режим МТ",
soft_failures,
).actual(
mt_report_utils.is_expected_dominant_mode_column(mode_totals, dominant_column)
).is_true_with_details(
expected_text=(
f"'{dominant_column}': {lds_report_utils.format_duration_seconds(dominant_total)} - "
f"наибольшее время; по всем участкам: {all_modes_text}"
),
actual_text=(
f"наибольшее время у режима '{max_column}': "
f"{lds_report_utils.format_duration_seconds(max_total)}"
),
)
allure.attach(
"\n".join(
f"{column_name}: {lds_report_utils.format_duration_seconds(total_seconds)}"
for column_name, total_seconds in mode_totals.items()
),
name="Суммарные длительности режимов МТ по всем участкам",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных шапки отчёта для проверки"):
title_info = parsed_report.title_info
report_title_lower = title_info.raw_title.lower()
column_headers = parsed_report.column_headers
header_period_start = title_info.period_start
header_period_end = title_info.period_end
with allure.step("Проверка двойной шапки отчёта о режиме МТ"):
with SoftAssertions() as soft_failures:
StepCheck(
f"В шапке отчёта присутствует '{MtReportConst.MT_MODE_REPORT_NAME_PART}'",
"report_title",
soft_failures,
).contains(report_title_lower, mt_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(MtReportConst.EXPECTED_COLUMN_HEADERS).equal_to()
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"Имя файла содержит '{MtReportConst.MT_MODE_REPORT_NAME_PART}'",
"file_name",
soft_failures,
).contains(report_file_name_lower, mt_report_name_part_lower)
StepCheck(
f"Имя файла содержит описание ТУ '{cfg.technological_unit.description}'",
"file_name",
soft_failures,
).contains(report_file_name_lower, report_state.expected_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)
except Exception:
with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
if report_state.actual_temp_file_path and report_state.expected_file_name:
report_utils.attach_report_file_to_allure(
report_state.actual_temp_file_path,
report_state.expected_file_name,
)
raise
with allure.step("Проверка пуш-нотификации о готовности отчёта"):
notification = report_state.actual_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 SoftAssertions() as soft_failures:
StepCheck("Получена пуш-нотификация о готовности отчёта", "notification", soft_failures).actual(
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()
"""
Утилиты для разбора xlsx-отчёта о режиме работы МТ.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from openpyxl.worksheet.worksheet import Worksheet
from constants.test_constants import ExportMtModeReportConstants as MtReportConst
from utils.helpers.lds_status_report_xlsx_utils import (
find_total_work_duration,
format_duration_seconds,
is_duration_cell_filled,
parse_duration_seconds,
)
from utils.helpers.report_xlsx_utils import (
ReportTitleInfo,
build_column_cells,
get_report_column_headers,
parse_report_title,
sum_duration_columns_across_rows,
)
@dataclass
class MtModeReportSectionRow:
"""Строка участка с длительностями режимов МТ."""
row_index: int
section_name: str
cells: Dict[str, str] = field(default_factory=dict)
@property
def mode_durations_seconds(self) -> Dict[str, int]:
"""Длительности режимов МТ в секундах по колонкам отчёта."""
return {
column_name: parse_duration_seconds(self.cells.get(column_name)) or 0
for column_name in MtReportConst.MODE_DURATION_COLUMNS
}
@property
def modes_sum_seconds(self) -> int:
"""Сумма длительностей всех режимов МТ для участка."""
return sum(self.mode_durations_seconds.values())
@dataclass
class MtModeReportParsed:
"""Разобранный отчёт о режиме работы МТ."""
title_info: ReportTitleInfo
column_headers: List[str]
section_rows: List[MtModeReportSectionRow]
total_duration_seconds: Optional[int] = None
total_duration_raw: str = ""
total_label_row_index: Optional[int] = None
def is_expected_dominant_mode_column(mode_totals: Dict[str, int], expected_column: str) -> bool:
"""
Проверяет, что суммарное время ожидаемого режима строго максимально и больше нуля.
Не допускает «мягкую» проверку <=, чтобы не пропустить ничью с другим режимом.
"""
expected_total = mode_totals.get(expected_column, 0)
if expected_total <= 0:
return False
return expected_total == max(mode_totals.values())
def parse_mt_mode_report_worksheet(
worksheet: Worksheet,
expected_section_names: List[str],
) -> MtModeReportParsed:
"""
Разбирает лист xlsx-отчёта о режиме МТ: шапка, колонки, строки участков и суммарное время.
В section_rows попадают только участки из expected_section_names (без учёта регистра).
"""
headers = get_report_column_headers(worksheet, MtReportConst.REPORT_COLUMN_HEADERS_ROW)
title_info = parse_report_title(
worksheet.cell(row=MtReportConst.REPORT_TITLE_ROW, column=1).value,
MtReportConst.REPORT_HEADER_PERIOD_PATTERN,
)
total_duration_seconds, total_duration_raw, total_label_row_index = find_total_work_duration(
worksheet,
data_first_row=MtReportConst.REPORT_DATA_FIRST_ROW,
total_work_duration_label=MtReportConst.TOTAL_WORK_DURATION_LABEL,
)
section_rows: List[MtModeReportSectionRow] = []
expected_names_lower = {name.lower() for name in expected_section_names}
for row_index, row_values in enumerate(
worksheet.iter_rows(
min_row=MtReportConst.REPORT_DATA_FIRST_ROW,
max_col=len(headers) if headers else 5,
values_only=True,
),
start=MtReportConst.REPORT_DATA_FIRST_ROW,
):
if total_label_row_index is not None and row_index >= total_label_row_index:
break
cells = build_column_cells(row_values, headers)
section_name = cells.get(MtReportConst.COL_SECTION, "").strip()
if not section_name:
continue
if section_name.lower() not in expected_names_lower:
continue
section_rows.append(
MtModeReportSectionRow(
row_index=row_index,
section_name=section_name,
cells=cells,
)
)
return MtModeReportParsed(
title_info=title_info,
column_headers=headers,
section_rows=section_rows,
total_duration_seconds=total_duration_seconds,
total_duration_raw=total_duration_raw,
total_label_row_index=total_label_row_index,
)
def format_mt_mode_section_rows_for_allure(section_rows: List[MtModeReportSectionRow]) -> str:
"""Форматирует строки участков отчёта о режиме МТ для вложения в Allure."""
lines = []
for row in section_rows:
durations_text = ", ".join(
f"{column}={format_duration_seconds(seconds)}"
for column, seconds in row.mode_durations_seconds.items()
)
lines.append(
f"row#{row.row_index}: {row.section_name} | sum={format_duration_seconds(row.modes_sum_seconds)} | "
f"{durations_text}"
)
return "\n".join(lines)
__all__ = [
"MtModeReportParsed",
"MtModeReportSectionRow",
"format_mt_mode_section_rows_for_allure",
"is_duration_cell_filled",
"is_expected_dominant_mode_column",
"parse_mt_mode_report_worksheet",
"sum_duration_columns_across_rows",
]