https://pastein.ru/t/WJB
Загрузка данных
async def export_rejection_report(ws_client, cfg: IsRejectedConfig, imitator_start_time: datetime):
"""
Сценарий формирования общего xlsx-отчёта об отбракованных входных данных.
Этапы:
1. Подписка SubscribeReportsDataExportedRequest на пуш-нотификации.
2. ExportReportsCommandRequest с периодом от старта имитатора до offset теста.
3. Ожидание ReportDataExportedNotification.
4. Лонг-поллинг GetExportedDataListRequest до появления отчёта в списке.
5. DownloadExportedDataRequest и получение fileChunk.
6. Проверка формата, имени файла, двойной шапки и строк по RejectionTestCase.
"""
report_state = ExportRejectedReportState()
with allure.step("Подготовка параметров сценария формирования отчёта об отбракованных входных данных"):
# offset теста задаёт конец периода отчёта - после всех отдельных проверок отбраковок
report_state.expected_report_test = cfg.rejection_report_test
StepCheck("В конфигурации задан rejection_report_test", "rejection_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_file_name = report_utils.build_export_report_file_name(
cfg.technological_unit.description,
report_state.expected_period_start,
report_state.expected_period_end,
RejectedReportConst.REJECTED_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
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}",
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.REJECTED_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.REJECTED_REPORT,
name_substring=RejectedReportConst.REJECTED_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.REJECTED_REPORT.to_download_name(),
"additionalProperties": None,
"timeOffset": report_state.actual_time_offset_hours,
}
download_purpose = (
f"скачивание xlsx-отчёта об отбракованных входных данных "
f"(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("Подготовка данных для проверки имени файла отчёта"):
# имя файла берём из ответа бэка (список сформированных отчётов), не из шапки xlsx
actual_file_name = report_state.actual_report_item.name if report_state.actual_report_item else ""
actual_file_name_lower = actual_file_name.lower()
file_name_period_start, file_name_period_end = report_utils.parse_period_from_export_file_name(
actual_file_name,
RejectedReportConst.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(actual_file_name)
rejected_report_file_name_part_lower = RejectedReportConst.REJECTED_REPORT_NAME_PART.lower()
rejected_report_file_name_part_alt_lower = RejectedReportConst.REJECTED_REPORT_NAME_PART_ALT.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="rejected_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_title_info = report_utils.parse_report_title(
report_utils.get_report_title_cell(report_state.actual_worksheet),
RejectedReportConst.REPORT_HEADER_PERIOD_PATTERN,
)
allure.attach(
f"Шапка (raw): {report_state.actual_title_info.raw_title}\n"
f"period_start: {report_state.actual_title_info.period_start}\n"
f"period_end: {report_state.actual_title_info.period_end}",
name="Первая строка шапки xlsx-отчёта",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Этап 8. Извлечение строк данных из отчёта"):
report_state.actual_data_rows = rejection_report_utils.iter_rejection_report_rows(
report_state.actual_worksheet
)
report_state.actual_monitored_tag_rows = rejection_report_utils.filter_rows_by_monitored_tags(
report_state.actual_data_rows,
RejectionSensorTag,
)
allure.attach(
rejection_report_utils.format_rejection_rows_for_allure(report_state.actual_monitored_tag_rows),
name="Строки отчёта по тегам RejectionSensorTag",
attachment_type=allure.attachment_type.TEXT,
)
with allure.step("Подготовка данных шапки xlsx для проверки"):
title_info = report_state.actual_title_info
column_headers = report_utils.get_report_column_headers(
report_state.actual_worksheet,
headers_row=RejectedReportConst.REPORT_COLUMN_HEADERS_ROW,
)
header_period_start = title_info.period_start
header_period_end = title_info.period_end
header_contains_expected_title = rejection_report_utils.report_header_contains_expected_title(
title_info.raw_title
)
with allure.step("Проверка первой строки шапки xlsx-отчёта"):
StepCheck("Лист xlsx открыт", "worksheet").actual(report_state.actual_worksheet).is_not_none()
with SoftAssertions() as soft_failures:
StepCheck(
"Первая строка шапки содержит заголовок отчёта об отбракованных входных данных",
"report_title",
soft_failures,
).actual(header_contains_expected_title).expected(True).equal_to()
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(RejectedReportConst.EXPECTED_COLUMN_HEADERS).equal_to()
with allure.step("Проверка строк отчёта по каждому RejectionTestCase из конфигурации набора"):
# Проверяем только установление отбраковок: каждый кейс из набора должен быть в таблице
with SoftAssertions() as soft_failures:
for rejection_case in cfg.rejection_cases:
report_event = rejection_report_utils.expected_event_to_report_event(rejection_case.expected_event)
window_start, window_end = rejection_report_utils.get_case_time_window(
imitator_start_time,
rejection_case,
)
raw_case_rows = rejection_report_utils.filter_rows_for_rejection_case(
report_state.actual_monitored_tag_rows,
rejection_case,
imitator_start_time,
)
merged_case_rows = rejection_report_utils.merge_rejection_rows(raw_case_rows)
primary_row = rejection_report_utils.select_primary_merged_row(merged_case_rows)
expected_signal_suffix = rejection_report_utils.report_signal_suffix_by_expected_name(
rejection_case.expected_signal_name,
)
case_label = f"события '{report_event}' - {rejection_case.sensor.description}"
StepCheck(
f"В отчёте найдена отбраковка для {case_label} в интервале времени "
f"{window_start} - {window_end}",
RejectedReportConst.COL_TAG,
soft_failures,
).actual(primary_row).is_not_none()
if primary_row is None:
continue
merge_key = rejection_report_utils.build_merge_key(primary_row)
expected_duration_seconds = rejection_report_utils.sum_duration_for_merge_key(
raw_case_rows,
merge_key,
)
pipe_section, actual_signal_suffix = rejection_report_utils.split_object_column(
primary_row.object_value
)
StepCheck(
f"Для {case_label} время получения отбраковки в допустимом диапазоне",
RejectedReportConst.COL_DATETIME,
soft_failures,
).actual(primary_row.datetime_value).is_between(window_start, window_end)
StepCheck(
f"Для {case_label} суммарная продолжительность отбраковки "
f"({format_duration_seconds(expected_duration_seconds)}) совпадает",
RejectedReportConst.COL_DURATION,
soft_failures,
).actual(primary_row.duration_seconds).expected(expected_duration_seconds).equal_to()
StepCheck(
f"Для {case_label} участок трубопровода в колонке '{RejectedReportConst.COL_OBJECT}' не пустой",
RejectedReportConst.COL_OBJECT,
soft_failures,
).actual(pipe_section).is_not_empty()
StepCheck(
f"Для {case_label} после последней точки в колонке '{RejectedReportConst.COL_OBJECT}' "
f"указан сигнал '{expected_signal_suffix}'",
RejectedReportConst.COL_OBJECT,
soft_failures,
).actual(actual_signal_suffix).expected(expected_signal_suffix).equal_to()
except Exception:
with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
attachment_name = (
report_state.actual_report_item.name
if report_state.actual_report_item and report_state.actual_report_item.name
else report_state.expected_file_name
)
if report_state.actual_temp_file_path and attachment_name:
report_utils.attach_report_file_to_allure(report_state.actual_temp_file_path, attachment_name)
raise
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(
"Имя файла из ответа бэка содержит название отчёта об отбракованных входных данных",
"file_name",
soft_failures,
).actual(
rejected_report_file_name_part_lower in actual_file_name_lower
or rejected_report_file_name_part_alt_lower in actual_file_name_lower
).expected(
True
).equal_to()
StepCheck(
f"Имя файла содержит описание ТУ '{cfg.technological_unit.description}'",
"file_name",
soft_failures,
).contains(actual_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)
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(
report_state.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()
class ExportedDataType(IntEnum):
"""
Тип экспортируемых данных
"""
STATIONARY_STATUS_REPORT = 6
LDS_STATUS_REPORT = 5
LEAKS_REPORT = 4
REJECTED_REPORT = 7
def to_download_name(self) -> str:
"""Строковый тип для DownloadExportedDataRequest.exportedDataType"""
return _EXPORTED_DATA_TYPE_DOWNLOAD_NAMES[self]
_EXPORTED_DATA_TYPE_DOWNLOAD_NAMES = {
ExportedDataType.STATIONARY_STATUS_REPORT: "StationaryStatusReport",
ExportedDataType.LDS_STATUS_REPORT: "LdsStateReport",
ExportedDataType.LEAKS_REPORT: "LeaksReport",
ExportedDataType.REJECTED_REPORT: "RejectedReport",
}
Этап 5. Streaming-вызов DownloadExportedDataRequest по id=301 1 вложенный шаг0s
Streaming-вызов DownloadExportedDataRequest c параметрами {'exportedDataId': 301, 'exportedDataType': 'RejectedReport', 'additionalProperties': None, 'timeOffset': 3}0s
Этап 6. Получение fileChunk - скачивание отчёта об отбракованных входных данных 2 вложения503ms
WS parse failure
294 B
received ws message
248 B
[3,
{},
'4',
1,
"Failed to invoke 'DownloadExportedDataRequest' due to an error on the server. InvalidDataException: Error binding "
'arguments. Make sure that the types of the provided values match the types of the hub method being invoked.']
Failed: При скачивание xlsx-отчёта об отбракованных входных данных (exportedDataId=301) после формирования отчёта и выбора файла в списке GetExportedDataListRequest бэк вернул Completion с ошибкой (DownloadExportedDataRequest, invocation_id=4): Failed to invoke 'DownloadExportedDataRequest' due to an error on the server. InvalidDataException: Error binding arguments. Make sure that the types of the provided values match the types of the hub method being invoked.
