https://pastein.ru/t/iJ0
Загрузка данных
configuration_utils.py
from typing import Any, Optional, Tuple
def extract_data_from_configuration(configuration_json: Any) -> Tuple[dict, dict]:
"""
1. Собирает соответствие address -> id для всех объектов конфигурации.
Учитываются только словари, где address и id находятся на одном уровне.
2. Получает список controlledSite и преобразует его в словарь controlled_sites_segments
"""
sensor_ids_by_address: dict[str, int] = {}
controlled_sites_segments: dict[str, Tuple[Optional[int], Optional[int]]] = {}
stack = [configuration_json]
while stack:
current_element = stack.pop()
if isinstance(current_element, dict):
address = current_element.get("address")
sensor_id = current_element.get("id")
controlled_sites = current_element.get("controlledSites")
if isinstance(address, str) and isinstance(sensor_id, int) and sensor_id != 0:
sensor_ids_by_address[address] = sensor_id
if isinstance(controlled_sites, list):
if any(
isinstance(item, dict) and "realObjectId" in item and "segments" in item
for item in controlled_sites
):
controlled_sites_segments = build_segments_dict(controlled_sites)
stack.extend(reversed(current_element.values()))
elif isinstance(current_element, list):
stack.extend(reversed(current_element))
return sensor_ids_by_address, controlled_sites_segments
def build_segments_dict(controlled_sites: list[dict]) -> dict[str, Tuple[Optional[int], Optional[int]]]:
"""
Преобразует список controlledSites в словарь segments вида:
{"ЗА 0-3-3 - ЗА 0-1-3": (6006, 6007)}
"""
segments_dict: dict[str, Tuple[Optional[int], Optional[int]]] = {}
for controlled_site in controlled_sites:
controlled_site_id = controlled_site.get("id")
segments = controlled_site.get("segments", [])
if not isinstance(segments, list):
continue
for segment in segments:
if not isinstance(segment, dict):
continue
segment_name = segment.get("name")
segment_id = segment.get("id")
if segment_name is None or segment_id is None:
continue
segments_dict[segment_name] = (controlled_site_id, segment_id)
return segments_dict
lds_configurator_utils.py
"""
Вспомогательные функции setup/teardown СОУ через раздел Администрирование.
"""
from __future__ import annotations
import asyncio
import logging
import time
from contextlib import contextmanager
from typing import Any, Optional
import allure
from _pytest.outcomes import Failed
from pytest import fail
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.enums import ReplyStatus, SouAdminStatus
from constants.test_constants import LdsConfiguratorConstants as LdsCfgConst
from models.basic_info_model import BasicInfoReply, BasicTUInfo
from models.get_basic_info_admin_model import AdminTuInfo, GetBasicInfoAdminReply
from models.get_tus_information_model import GetTusInformationReply
from utils.helpers import ws_test_utils as t_utils
from utils.helpers.ws_message_parser import WsMessageParser
from utils.msgpack_utils.message_filters import is_desired_type
logger = logging.getLogger(__name__)
_configurator_flow_active: bool = False
def set_configurator_flow_active(enabled: bool) -> None:
"""Включает логирование setup/teardown при запуске СОУ через Администрирование"""
global _configurator_flow_active
_configurator_flow_active = enabled
def is_configurator_flow_active() -> bool:
"""True во время setup/teardown/verify СОУ через Администрирование"""
return _configurator_flow_active
def _fail_or_raise(message: str) -> None:
"""В configurator flow - RuntimeError для retry/teardown; иначе pytest.fail."""
if _configurator_flow_active:
raise RuntimeError(message)
fail(message, pytrace=False)
@contextmanager
def _step(name: str):
if _configurator_flow_active:
logger.info("[LDS_CONFIGURATOR] %s", name)
yield
else:
with allure.step(name):
yield
def attach_allure_alert(message: str) -> None:
"""
Публикует предупреждение в Allure и лог без падения прогона.
Используется в teardown при некритичных отклонениях.
"""
logger.warning("[LDS_CONFIGURATOR] %s", message)
if not _configurator_flow_active:
allure.attach(message, name="ALERT", attachment_type=allure.attachment_type.TEXT)
def get_basic_info(http_client: StandHttpClient, parser: WsMessageParser) -> BasicInfoReply:
"""
Выполняет getBasicInfoRequest и парсит ответ BasicInfoContent.
"""
response = http_client.post_request(HttpConst.GET_BASIC_INFO_URL_PATH, {})
payload = t_utils.get_json_from_http_response(response)
return parser.parse_basic_info_msg(payload)
def is_tu_in_basic_info(tus: Optional[list[BasicTUInfo]], tu_id: int, tu_name: str) -> bool:
"""True, если в basicInfo.tus есть запись с указанными tuId и tuName."""
return any(tu.tuId == tu_id and tu.tuName == tu_name for tu in (tus or []))
def get_basic_info_admin(
http_client: StandHttpClient,
parser: WsMessageParser,
) -> GetBasicInfoAdminReply:
"""
Выполняет GetBasicInfoAdminRequest и парсит ответ.
"""
response = http_client.post_request(HttpConst.GET_BASIC_INFO_ADMIN_URL_PATH, {})
payload = t_utils.get_json_from_http_response(response)
return parser.parse_get_basic_info_admin_msg(payload)
def get_basic_info_admin_with_retry(
http_client: StandHttpClient,
parser: WsMessageParser,
retries: int = LdsCfgConst.GET_BASIC_INFO_ADMIN_RETRIES,
) -> GetBasicInfoAdminReply:
"""
Запрашивает GetBasicInfoAdminResponse с повторными попытками.
"""
last_error: Optional[BaseException] = None
for attempt in range(1, retries + 1):
with _step(f"Запрос списка ТУ в Администрировании - попытка {attempt} из {retries}"):
try:
return get_basic_info_admin(http_client, parser)
except (
asyncio.TimeoutError,
ConnectionError,
ConnectionResetError,
OSError,
RuntimeError,
KeyError,
Failed,
) as error:
last_error = error
logger.warning(
"[LDS_CONFIGURATOR] GetBasicInfoAdmin попытка %s/%s: %s: %r",
attempt,
retries,
type(error).__name__,
error,
)
if attempt < retries:
time.sleep(LdsCfgConst.GET_BASIC_INFO_ADMIN_TIMEOUT_SECONDS)
with _step("Проверка: GetBasicInfoAdminResponse получен"):
_fail_or_raise(
f"Не удалось получить GetBasicInfoAdminResponse за {retries} попыток: {last_error}",
)
def find_tu_by_name(admin_reply: GetBasicInfoAdminReply, tu_name: str) -> AdminTuInfo:
"""
Ищет ТУ по точному совпадению tuName в ответе Администрирования.
"""
with _step(f"Поиск ТУ '{tu_name}' по точному совпадению tuName в GetBasicInfoAdminResponse"):
tus = admin_reply.replyContent.basicInfo.tus if admin_reply.replyContent else None
with _step("Проверка: в ответе есть список ТУ"):
if not tus:
fail(
f"GetBasicInfoAdminResponse не содержит списка ТУ (ожидался tuName='{tu_name}')",
pytrace=False,
)
for tu in tus:
if tu.tuName == tu_name:
return tu
available = [tu.tuName for tu in tus]
with _step("Проверка: ТУ из набора данных найден в Администрировании"):
fail(
f"ТУ '{tu_name}' не найден в GetBasicInfoAdminResponse. Доступные tu_name: {available}",
pytrace=False,
)
def validate_admin_tu(tu: AdminTuInfo) -> None:
"""
Проверяет обязательные поля AdminTuInfo и допустимость статуса СОУ.
"""
with _step(f"Валидация параметров ТУ '{tu.tuName}' (tuId={tu.tuId})"):
with _step("Проверка: tuId и mnId заполнены"):
if not tu.tuId:
fail(f"Некорректный tuId для ТУ '{tu.tuName}': {tu.tuId}", pytrace=False)
if not tu.mnId:
fail(f"Некорректный mnId для ТУ '{tu.tuName}': {tu.mnId}", pytrace=False)
with _step("Проверка: статус СОУ известен Администрированию"):
try:
SouAdminStatus(tu.status)
except ValueError:
fail(f"Неизвестный статус СОУ для ТУ '{tu.tuName}': {tu.status}", pytrace=False)
def extract_running_tus(admin_reply: GetBasicInfoAdminReply) -> list[AdminTuInfo]:
"""Возвращает все ТУ со статусом RUNNING из GetBasicInfoAdminResponse."""
tus = admin_reply.replyContent.basicInfo.tus if admin_reply.replyContent else []
return [tu for tu in tus if tu.status == SouAdminStatus.RUNNING.value]
def running_tus_to_snapshot(tus: list[AdminTuInfo]) -> list[dict[str, Any]]:
"""Сериализует список ТУ для хранения в group_state."""
return [{"tuId": tu.tuId, "tuName": tu.tuName} for tu in tus]
def _tu_id_in_main_page_message(msg: Any, tu_id: int) -> bool:
"""True, если WS-сообщение MainPageInfoContent содержит указанный tuId."""
if not isinstance(msg, list) or not is_desired_type(msg, LdsCfgConst.MAIN_PAGE_INFO_CONTENT):
return False
for item in msg:
if isinstance(item, dict) and item.get("replyContent", {}).get("tuId") == tu_id:
return True
if isinstance(item, list):
for elem in item:
if isinstance(elem, dict) and elem.get("replyContent", {}).get("tuId") == tu_id:
return True
return False
def _drain_recv_queue(ws_client: WebSocketClient) -> list[Any]:
"""Забирает все сообщения из очереди WS без блокирующего ожидания."""
messages: list[Any] = []
while not ws_client.recv_queue.empty():
try:
messages.append(ws_client.recv_queue.get_nowait())
except asyncio.QueueEmpty:
break
return messages
async def is_tu_present_on_main_page(
ws_client: WebSocketClient,
parser: WsMessageParser,
tu_id: int,
timeout: float = LdsCfgConst.MAIN_PAGE_SYNC_TIMEOUT_SECONDS,
) -> bool:
"""
Подписывается на MainPageInfoContent и определяет, отображается ли ТУ в Состоянии МТ.
"""
with _step(f"Подписка на Состояние МТ (MainPageInfoContent) для tuId={tu_id}"):
ws_client.clear_queue()
await t_utils.connect(
ws_client,
LdsCfgConst.SUBSCRIBE_MAIN_PAGE_INFO_REQUEST,
{"tuIds": [tu_id], "additionalProperties": None},
)
deadline = asyncio.get_running_loop().time() + timeout
while asyncio.get_running_loop().time() < deadline:
remaining = deadline - asyncio.get_running_loop().time()
if remaining <= 0:
break
try:
payload = await ws_client.receive_by_type(LdsCfgConst.MAIN_PAGE_INFO_CONTENT, timeout=remaining)
except asyncio.TimeoutError:
break
parsed = parser.parse_main_page_msg(payload)
if parsed.replyContent and parsed.replyContent.tuId == tu_id:
return True
return False
def check_sou_status_sync(
sou_status: SouAdminStatus,
is_in_basic_info: bool,
is_on_main_page: bool,
tu_id: int,
tu_name: str,
) -> None:
"""
Сверяет статус СОУ в Администрировании и на ЭФ Состояние МТ по двум DTO BasicInfo и MainPageInfoContent.
"""
with _step(f"Сверка статуса СОУ: Администрирование vs Состояние МТ (tuId={tu_id}, '{tu_name}')"):
expected_enabled = sou_status == SouAdminStatus.RUNNING
with _step("Проверка согласованности статусов Администрирования и Состояния МТ"):
if is_in_basic_info == expected_enabled and is_on_main_page == expected_enabled:
return
admin_text = SouAdminStatus.report_text_by_value(sou_status.value)
basic_text = "СОУ запущена" if is_in_basic_info else "СОУ не запущена"
page_text = "СОУ запущена" if is_on_main_page else "СОУ не запущена"
fail(
f"Рассинхронизация для ТУ '{tu_name}' (tuId={tu_id}): "
f"Администрирование - {admin_text} ({sou_status.value}); "
f"Состояние МТ в BasicInfo - {basic_text}; "
f"Состояние МТ в MainPageInfoContent - {page_text}. "
f"Статусы в разных подписках не совпадают.",
pytrace=False,
)
def run_lds_command(
http_client: StandHttpClient,
request_name: str,
tu_id: int,
) -> None:
"""
Отправляет StopLdsRequest или LaunchLdsRequest и ждёт Completion с replyStatus=200.
"""
with _step(f"Команда {request_name} для tuId={tu_id}"):
response = http_client.post_request(request_name, {"tuId": tu_id})
with _step(f"Проверка: {request_name} завершился успешно (response.status_code=200)"):
if response.status_code != ReplyStatus.OK.value:
fail(
f"{request_name} завершился с replyStatus={response.status_code}, " f"ошибки: {response.text}",
pytrace=False,
)
async def poll_admin_tu_status(
http_client: StandHttpClient,
parser: WsMessageParser,
tu_id: int,
expected_status: SouAdminStatus,
total_wait_seconds: float = LdsCfgConst.POLL_TIMEOUT_SECONDS,
poll_interval_seconds: float = LdsCfgConst.POLL_INTERVAL_SECONDS,
) -> bool:
"""
Long-poll GetBasicInfoAdmin до смены статуса ТУ в Администрировании.
"""
status_label = SouAdminStatus.report_text_by_value(expected_status.value)
with _step(
f"Ожидание статуса '{status_label}' в Администрировании " f"(tuId={tu_id}, таймаут {int(total_wait_seconds)} с)"
):
deadline = asyncio.get_running_loop().time() + total_wait_seconds
while asyncio.get_running_loop().time() < deadline:
admin_reply = get_basic_info_admin(http_client, parser)
tus = admin_reply.replyContent.basicInfo.tus if admin_reply.replyContent else []
tu = next((item for item in tus if item.tuId == tu_id), None)
if tu and tu.status == expected_status.value:
return True
await asyncio.sleep(poll_interval_seconds)
return False
async def stop_tu_and_wait(
http_client: StandHttpClient,
parser: WsMessageParser,
tu_id: int,
tu_name: Optional[str] = None,
) -> None:
"""
Останавливает СОУ и ждёт статус STOPPED в Администрировании.
При таймауте - pytest.fail (блокирующая ошибка setup).
"""
label = f"tuId={tu_id}, tuName={tu_name!r}" if tu_name else f"tuId={tu_id}"
with _step(f"Остановка СОУ ({label})"):
run_lds_command(http_client, HttpConst.STOP_LDS_URL_PATH, tu_id)
if not await poll_admin_tu_status(http_client, parser, tu_id, SouAdminStatus.STOPPED):
fail(
f"СОУ не выключилась за {int(LdsCfgConst.POLL_TIMEOUT_SECONDS)} с: {label}",
pytrace=False,
)
async def launch_tu_and_wait(
http_client: StandHttpClient,
parser: WsMessageParser,
tu_id: int,
) -> bool:
"""
Запускает СОУ и ждёт статус RUNNING в Администрировании.
Возвращает True при успехе (для soft teardown).
"""
with _step(f"Запуск СОУ (tuId={tu_id})"):
run_lds_command(http_client, HttpConst.LAUNCH_LDS_URL_PATH, tu_id)
return await poll_admin_tu_status(http_client, parser, tu_id, SouAdminStatus.RUNNING)
async def stop_running_tus(
http_client: StandHttpClient,
parser: WsMessageParser,
tus: list[AdminTuInfo],
) -> None:
"""Последовательно останавливает включённые ТУ на стенде."""
if not tus:
return
logger.info("[SETUP] Остановка включённых ТУ на стенде: %s шт.", len(tus))
for tu in tus:
logger.info("[SETUP] Остановка ТУ на стенде: tuId=%s, tuName=%r", tu.tuId, tu.tuName)
await stop_tu_and_wait(http_client, parser, tu.tuId, tu.tuName)
async def restore_pre_run_tus(
http_client: StandHttpClient,
parser: WsMessageParser,
snapshot: list[dict[str, Any]],
exclude_tu_id: int,
) -> None:
"""
Включает обратно ТУ из снимка, кроме ТУ автотестов.
Ошибки по отдельным ТУ оформляются как ALERT без падения прогона.
"""
to_restore = [entry for entry in snapshot if entry.get("tuId") != exclude_tu_id]
if not to_restore:
return
logger.info(
"[TEARDOWN] Восстановление ТУ стенда: %s шт. (исключена ТУ автотестов tuId=%s)",
len(to_restore),
)
for entry in to_restore:
restore_tu_id = entry["tuId"]
restore_tu_name = entry.get("tuName", "")
logger.info(
"[TEARDOWN] Запуск ТУ из снимка: tuId=%s, tuName=%r",
restore_tu_id,
restore_tu_name,
)
if not await launch_tu_and_wait(http_client, parser, restore_tu_id):
attach_allure_alert(
f"Не удалось восстановить ТУ из снимка: tuId={restore_tu_id}, "
f"tuName={restore_tu_name!r}. Проверить вручную."
)
async def poll_basic_info_tu_presence(
http_client: StandHttpClient,
parser: WsMessageParser,
tu_id: int,
tu_name: str,
expect_present: bool,
total_wait_seconds: float = LdsCfgConst.POLL_TIMEOUT_SECONDS,
poll_interval_seconds: float = LdsCfgConst.POLL_INTERVAL_SECONDS,
) -> bool:
"""
Long-poll getBasicInfoRequest: ожидание появления или исчезновения ТУ в basicInfo.tus.
"""
action = "появления" if expect_present else "исчезновения"
with _step(
f"Ожидание {action} ТУ в BasicInfo " f"(tuId={tu_id}, tuName='{tu_name}', таймаут {int(total_wait_seconds)} с)"
):
deadline = asyncio.get_running_loop().time() + total_wait_seconds
while asyncio.get_running_loop().time() < deadline:
reply = get_basic_info(http_client, parser)
tus = reply.replyContent.basicInfo.tus if reply.replyContent else None
found = is_tu_in_basic_info(tus, tu_id, tu_name)
if expect_present and found:
return True
if not expect_present and not found:
return True
await asyncio.sleep(poll_interval_seconds)
if _configurator_flow_active:
logger.error(
"[LDS_CONFIGURATOR] Таймаут ожидания %s ТУ tuId=%s в BasicInfo за %s с",
action,
tu_id,
int(total_wait_seconds),
)
return False
async def poll_main_page_tu_presence(
ws_client: WebSocketClient,
tu_id: int,
expect_present: bool,
total_wait_seconds: float = LdsCfgConst.POLL_TIMEOUT_SECONDS,
poll_interval_seconds: float = LdsCfgConst.POLL_INTERVAL_SECONDS,
) -> bool:
"""
Long-poll MainPageInfoContent: ожидание появления или исчезновения ТУ в Состоянии МТ.
"""
action = "появления" if expect_present else "отсутствия"
with _step(f"Ожидание {action} ТУ в Состоянии МТ (tuId={tu_id}, таймаут {int(total_wait_seconds)} с)"):
ws_client.clear_queue()
await t_utils.connect(
ws_client,
LdsCfgConst.SUBSCRIBE_MAIN_PAGE_INFO_REQUEST,
{"tuIds": [tu_id], "additionalProperties": None},
)
deadline = asyncio.get_running_loop().time() + total_wait_seconds
while asyncio.get_running_loop().time() < deadline:
await asyncio.sleep(poll_interval_seconds)
batch = _drain_recv_queue(ws_client)
found = any(_tu_id_in_main_page_message(msg, tu_id) for msg in batch)
if expect_present and found:
return True
if not expect_present and not found:
return True
if not _configurator_flow_active:
t_utils.attach_ws_poll_failure(
[],
total_wait_seconds,
f"{LdsCfgConst.MAIN_PAGE_INFO_CONTENT} tuId={tu_id} present={expect_present}",
)
else:
logger.warning(
"[LDS_CONFIGURATOR] Таймаут ожидания %s ТУ tuId=%s за %s с",
action,
tu_id,
int(total_wait_seconds),
)
return False
def verify_get_tus_info(http_client: StandHttpClient, parser: WsMessageParser, tu_id: int) -> None:
"""
Проверяет, наличие ТУ и launchedAt в GetTusInformation.
"""
with _step(f"Запрос GetTusInformation для tuId={tu_id}"):
response = http_client.post_request(HttpConst.GET_TUS_INFORMATION_URL_PATH, {"tuIds": [tu_id]})
payload = t_utils.get_json_from_http_response(response)
reply: GetTusInformationReply = parser.parse_get_tus_information_msg(payload)
tus_info = reply.replyContent.tusInfo if reply.replyContent else []
tu_info = next((item for item in tus_info if item.tuId == tu_id), None)
with _step("Проверка: в ответе есть информация о запуске ТУ"):
if tu_info is None:
fail(f"GetTusInformationResponse не содержит tuId={tu_id}", pytrace=False)
launched_at = parser.timestamp_to_datetime(tu_info.launchedAt)
with _step(f"Проверка (GetTusInformation): Наличие времени запуска launchedAt: {launched_at} для tuId={tu_id}"):
if launched_at is None:
fail(f"GetTusInformationResponse: launchedAt отсутствует для tuId={tu_id}", pytrace=False)
def get_admin_tu_status(admin_reply: GetBasicInfoAdminReply, tu_id: int) -> Optional[SouAdminStatus]:
"""
Возвращает статус СОУ из GetBasicInfoAdmin для указанного tuId.
"""
tus = admin_reply.replyContent.basicInfo.tus if admin_reply.replyContent else []
tu = next((item for item in tus if item.tuId == tu_id), None)
if tu is None:
return None
try:
return SouAdminStatus(tu.status)
except ValueError:
return None
lds_status_report_xlsx_utils.py
"""
Утилиты для разбора xlsx-отчёта о режиме работы СОУ.
"""
from __future__ import annotations
from typing import List
from openpyxl.worksheet.worksheet import Worksheet
from constants.test_constants import ExportLdsStatusReportConstants as LdsReportConst
from utils.helpers import mode_duration_report_xlsx_utils as mode_duration_utils
_LDS_LAYOUT = mode_duration_utils.ModeDurationReportLayout.from_constants(LdsReportConst)
# Публичные алиасы типов отчёта СОУ (общая реализация — в mode_duration_report_xlsx_utils).
LdsStatusReportSectionRow = mode_duration_utils.ModeDurationReportSectionRow
LdsStatusReportParsed = mode_duration_utils.ModeDurationReportParsed
find_total_work_duration = mode_duration_utils.find_total_work_duration
format_duration_seconds = mode_duration_utils.format_duration_seconds
format_section_rows_for_allure = mode_duration_utils.format_mode_duration_section_rows_for_allure
is_duration_cell_filled = mode_duration_utils.is_duration_cell_filled
parse_duration_seconds = mode_duration_utils.parse_duration_seconds
def parse_lds_status_report_worksheet(
worksheet: Worksheet,
expected_section_names: List[str],
) -> LdsStatusReportParsed:
"""Разбирает лист xlsx-отчёта о режиме СОУ."""
return mode_duration_utils.parse_mode_duration_report_worksheet(
worksheet,
expected_section_names,
_LDS_LAYOUT,
)
__all__ = [
"LdsStatusReportParsed",
"LdsStatusReportSectionRow",
"find_total_work_duration",
"format_duration_seconds",
"format_section_rows_for_allure",
"is_duration_cell_filled",
"parse_duration_seconds",
"parse_lds_status_report_worksheet",
]
/
mode_duration_report_xlsx_utils.py
"""
Общая логика разбора xlsx-отчётов с таблицей участков и длительностями режимов.
Используется отчётами о режиме СОУ и о режиме работы МТ.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, time, timedelta
from typing import Dict, List, Optional, Protocol, Tuple
from openpyxl.worksheet.worksheet import Worksheet
from constants.test_constants import BaseTN3Constants as TestConst
from utils.helpers.report_xlsx_utils import (
ReportTitleInfo,
_stringify_cell,
build_column_cells,
get_report_column_headers,
parse_report_title,
)
_DURATION_PARTS_H_MM_SS = 3
_DURATION_PARTS_MM_SS = 2
class ModeDurationReportConstants(Protocol):
"""Структурные константы xlsx-отчёта с таблицей участков и длительностями режимов."""
REPORT_TITLE_ROW: int
REPORT_COLUMN_HEADERS_ROW: int
REPORT_DATA_FIRST_ROW: int
TOTAL_WORK_DURATION_LABEL: str
COL_SECTION: str
REPORT_HEADER_PERIOD_PATTERN: str
MODE_DURATION_COLUMNS: list
@dataclass(frozen=True)
class ModeDurationReportLayout:
"""Параметры разбора листа отчёта с длительностями режимов по участкам."""
report_title_row: int
report_column_headers_row: int
report_data_first_row: int
total_work_duration_label: str
col_section: str
header_period_pattern: str
mode_duration_columns: Tuple[str, ...]
@classmethod
def from_constants(cls, constants: ModeDurationReportConstants) -> ModeDurationReportLayout:
return cls(
report_title_row=constants.REPORT_TITLE_ROW,
report_column_headers_row=constants.REPORT_COLUMN_HEADERS_ROW,
report_data_first_row=constants.REPORT_DATA_FIRST_ROW,
total_work_duration_label=constants.TOTAL_WORK_DURATION_LABEL,
col_section=constants.COL_SECTION,
header_period_pattern=constants.REPORT_HEADER_PERIOD_PATTERN,
mode_duration_columns=tuple(constants.MODE_DURATION_COLUMNS),
)
@dataclass
class ModeDurationReportSectionRow:
"""Строка участка с длительностями режимов."""
row_index: int
section_name: str
cells: Dict[str, str] = field(default_factory=dict)
mode_duration_columns: Tuple[str, ...] = ()
@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 self.mode_duration_columns
}
@property
def modes_sum_seconds(self) -> int:
return sum(self.mode_durations_seconds.values())
@dataclass
class ModeDurationReportParsed:
"""Разобранный отчёт с таблицей участков и длительностями режимов."""
title_info: ReportTitleInfo
column_headers: List[str]
section_rows: List[ModeDurationReportSectionRow]
total_duration_seconds: Optional[int] = None
total_duration_raw: str = ""
total_label_row_index: Optional[int] = None
def parse_duration_seconds(value: object) -> Optional[int]:
"""Парсит длительность из ячейки (H:MM:SS, MM:SS или time/timedelta из Excel)."""
if value is None:
return None
if isinstance(value, timedelta):
return int(value.total_seconds())
if isinstance(value, time):
return value.hour * TestConst.SECONDS_PER_HOUR + value.minute * TestConst.SEC_PER_MIN + value.second
if isinstance(value, datetime):
return value.hour * TestConst.SECONDS_PER_HOUR + value.minute * TestConst.SEC_PER_MIN + value.second
duration_text = _stringify_cell(value).strip()
if not duration_text:
return None
duration_text_no_microseconds = duration_text.split(".")[0]
parts = duration_text_no_microseconds.split(":")
try:
if len(parts) == _DURATION_PARTS_H_MM_SS:
hours, minutes, seconds = (int(part) for part in parts)
return hours * TestConst.SECONDS_PER_HOUR + minutes * TestConst.SEC_PER_MIN + seconds
if len(parts) == _DURATION_PARTS_MM_SS:
minutes, seconds = (int(part) for part in parts)
return minutes * TestConst.SEC_PER_MIN + seconds
except ValueError:
return None
return None
def is_duration_cell_filled(value: object) -> bool:
"""Ячейка с длительностью заполнена (допускается 0:00:00)."""
return parse_duration_seconds(value) is not None
def format_duration_seconds(total_seconds: int) -> str:
"""Форматирует длительность в секундах в строку H:MM:SS."""
hours, remainder = divmod(total_seconds, TestConst.SECONDS_PER_HOUR)
minutes, seconds = divmod(remainder, TestConst.SEC_PER_MIN)
return f"{hours}:{minutes:02d}:{seconds:02d}"
def find_total_work_duration(
worksheet: Worksheet,
*,
data_first_row: int,
total_work_duration_label: str,
) -> Tuple[Optional[int], str, Optional[int]]:
"""
Ищет строку с меткой суммарного времени и парсит длительность рядом.
Возвращает (секунды, сырое значение ячейки, номер строки с меткой) или (None, "", None).
"""
for row_index, row_values in enumerate(
worksheet.iter_rows(min_row=data_first_row, values_only=True),
start=data_first_row,
):
for column_index, cell_value in enumerate(row_values):
if cell_value is None:
continue
cell_text = _stringify_cell(cell_value).strip()
if total_work_duration_label not in cell_text:
continue
duration_candidates = []
if column_index + 1 < len(row_values):
duration_candidates.append(row_values[column_index + 1])
if row_index + 1 <= worksheet.max_row:
duration_candidates.append(worksheet.cell(row=row_index + 1, column=column_index + 1).value)
for candidate in duration_candidates:
duration_seconds = parse_duration_seconds(candidate)
if duration_seconds is not None:
return duration_seconds, _stringify_cell(candidate).strip(), row_index
return None, "", row_index
return None, "", None
def parse_mode_duration_report_worksheet(
worksheet: Worksheet,
expected_section_names: List[str],
layout: ModeDurationReportLayout,
) -> ModeDurationReportParsed:
"""
Разбирает лист xlsx: шапка, колонки, строки участков и суммарное время.
В section_rows попадают только участки из expected_section_names (без учёта регистра).
"""
headers = get_report_column_headers(worksheet, layout.report_column_headers_row)
title_info = parse_report_title(
worksheet.cell(row=layout.report_title_row, column=1).value,
layout.header_period_pattern,
)
total_duration_seconds, total_duration_raw, total_label_row_index = find_total_work_duration(
worksheet,
data_first_row=layout.report_data_first_row,
total_work_duration_label=layout.total_work_duration_label,
)
section_rows: List[ModeDurationReportSectionRow] = []
expected_names_lower = {name.lower() for name in expected_section_names}
for row_index, row_values in enumerate(
worksheet.iter_rows(
min_row=layout.report_data_first_row,
max_col=len(headers) if headers else 5,
values_only=True,
),
start=layout.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(layout.col_section, "").strip()
if not section_name:
continue
if section_name.lower() not in expected_names_lower:
continue
section_rows.append(
ModeDurationReportSectionRow(
row_index=row_index,
section_name=section_name,
cells=cells,
mode_duration_columns=layout.mode_duration_columns,
)
)
return ModeDurationReportParsed(
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_mode_duration_section_rows_for_allure(
section_rows: List[ModeDurationReportSectionRow],
) -> 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)
mt_mode_report_xlsx_utils.py
"""
Утилиты для разбора xlsx-отчёта о режиме работы МТ.
"""
from __future__ import annotations
from typing import Dict, List
from openpyxl.worksheet.worksheet import Worksheet
from constants.test_constants import ExportMtModeReportConstants as MtReportConst
from utils.helpers import mode_duration_report_xlsx_utils as mode_duration_utils
from utils.helpers.report_xlsx_utils import sum_duration_columns_across_rows
_MT_LAYOUT = mode_duration_utils.ModeDurationReportLayout.from_constants(MtReportConst)
MtModeReportSectionRow = mode_duration_utils.ModeDurationReportSectionRow
MtModeReportParsed = mode_duration_utils.ModeDurationReportParsed
format_duration_seconds = mode_duration_utils.format_duration_seconds
format_mt_mode_section_rows_for_allure = mode_duration_utils.format_mode_duration_section_rows_for_allure
is_duration_cell_filled = mode_duration_utils.is_duration_cell_filled
parse_duration_seconds = mode_duration_utils.parse_duration_seconds
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-отчёта о режиме МТ."""
return mode_duration_utils.parse_mode_duration_report_worksheet(
worksheet,
expected_section_names,
_MT_LAYOUT,
)
__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",
]
pytest_auth
"""
Авторизация pytest-прогона: Keycloak access token и x-user-id для HTTP/WS.
Кэш кредов на уровне dataset (group_state)
"""
from __future__ import annotations
import os
import time
import pytest
import requests
from clients.http_client import StandHttpClient
from clients.keycloak_clients import KeycloakAuthError, KeycloakClient
from clients.testops_client import logger
from clients.websocket_client import WebSocketClient
from constants.architecture_constants import EnvKeyConstants as EnvConst
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.architecture_constants import WebSocketClientConstants as WSCliConst
# Один KeycloakClient на pytest-сессию; TTL токена проверяется внутри клиента.
_keycloak_client: KeycloakClient | None = None
def get_instance() -> str:
"""Имя стенда из переменной окружения STAND_NAME."""
instance = os.environ.get(EnvConst.STAND_NAME)
if not instance:
pytest.exit(f"Переменная окружения {EnvConst.STAND_NAME} не задана в .env")
return instance
def build_stand_host() -> str:
"""Хост api-gateway для HTTP и WSS запросов к стенду."""
instance = get_instance()
return (
f"{WSCliConst.SERVICE_NAME}.{WSCliConst.COMPONENT}-{instance}.{WSCliConst.ROOT_DOMAIN}/"
f"{WSCliConst.API_GATEWAY_PATH_SEGMENT}"
)
def _get_keycloak_client() -> KeycloakClient:
"""
Возвращает единственный KeycloakClient на время pytest-сессии.
Вызовы get_access_token() внутри одного клиента ждут истечения токена KeycloakClient._is_token_expired.
"""
global _keycloak_client
if _keycloak_client is None:
_keycloak_client = KeycloakClient(
url=os.environ.get(EnvConst.KEYCLOAK_SZI_URL),
client_id=os.environ.get(EnvConst.KEYCLOAK_CLIENT_ID),
client_secret=os.environ.get(EnvConst.KEYCLOAK_SZI_CLIENT_SECRET),
username=os.environ.get(EnvConst.KEYCLOAK_USERNAME),
password=os.environ.get(EnvConst.KEYCLOAK_PASSWORD),
)
return _keycloak_client
def _invalidate_keycloak_token() -> None:
"""
Сбрасывает закэшированный access token в KeycloakClient.
Вызывается при force_refresh=True (смена dataset), чтобы запросить новый JWT,
даже если предыдущий ещё в пределах TTL.
"""
client = _get_keycloak_client()
client._token = None
client._token_data = None
def get_token(max_retries: int = 12, backoff: float = 5.0, force_refresh: bool = False) -> str:
"""
Получает JWT access token из Keycloak с повторными попытками.
max_retries: сколько всего попыток
backoff: время в секундах между попытками
force_refresh: принудительно запросить новый token в Keycloak
"""
if force_refresh:
_invalidate_keycloak_token()
last_exc = None
keycloak = _get_keycloak_client()
for attempt in range(1, max_retries + 1):
try:
token = keycloak.get_access_token()
if not token:
raise KeycloakAuthError("Получен пустой access token")
return token
except KeycloakAuthError as e:
last_exc = e
logger.warning(f"[{attempt}/{max_retries}] KeycloakAuthError: {e}. Повтор через {backoff} сек.")
except Exception as e:
last_exc = e
logger.warning(f"[{attempt}/{max_retries}] Неожиданная ошибка: {e}. Повтор через {backoff} сек.")
if attempt < max_retries:
time.sleep(backoff)
# все попытки исчерпаны
logger.error(f"[KEYCLOAK] [ERROR] Не удалось получить токен после {max_retries} попыток: {last_exc}")
pytest.fail(f"[KEYCLOAK][ERROR] Не удалось получить токен после {max_retries} попыток: {last_exc}")
def clear_suite_auth(group_state: dict) -> None:
"""
Сбрасывает кэш auth в group_state перед инициализацией нового dataset.
Очищает stand_host, auth_token, x_user_id и метку auth_suite.
"""
group_state["stand_host"] = None
group_state["auth_token"] = None
group_state["x_user_id"] = None
group_state["auth_suite"] = None
def _fetch_x_user_id(http_client: StandHttpClient, max_retries: int = 12, backoff: float = 5.0) -> str:
"""
Запрашивает x-user-id через POST /apigateway/Ping с повторными попытками.
Значение нужно для параметра xUserId= в URL WebSocket-подключения.
Вызывается после старта api-gateway.
"""
http_client.suppress_recv_logging = True
last_exc = None
for attempt in range(1, max_retries + 1):
try:
response = http_client.post_request(HttpConst.PING_URL_PATH, {})
x_user_id = response.headers.get(HttpConst.X_USER_ID_KEY)
if not x_user_id:
raise ValueError("/Ping не вернул x-user-id")
logger.info("[AUTH] [OK] Получен x-user-id")
return x_user_id
except (requests.RequestException, ValueError) as e:
last_exc = e
logger.warning(f"[{attempt}/{max_retries}] /Ping: {e}. Повтор через {backoff} сек.")
except Exception as e:
last_exc = e
logger.warning(f"[{attempt}/{max_retries}] Неожиданная ошибка /Ping: {e}. Повтор через {backoff} сек.")
if attempt < max_retries:
time.sleep(backoff)
# все попытки исчерпаны
logger.error(f"[AUTH] [ERROR] Не удалось получить x-user-id после {max_retries} попыток: {last_exc}")
pytest.fail(f"[AUTH][ERROR] Не удалось получить x-user-id после {max_retries} попыток: {last_exc}")
def ensure_suite_auth(group_state: dict, suite_name: str) -> None:
"""
Получает и кэширует credentials набора: stand_host, auth_token, x-user-id.
Порядок: Keycloak token -> Ping к api-gateway (x-user-id). Оба запрашиваются
после setup_stand_for_imitator_run(), когда gateway уже доступен.
Повторный вызов для того же suite_name с валидным кэшем ничего не делает.
"""
if (
group_state.get("auth_suite") == suite_name
and group_state.get("auth_token")
and group_state.get("x_user_id")
and group_state.get("stand_host")
):
return
stand_host = build_stand_host()
auth_token = get_token(force_refresh=True)
http_client = StandHttpClient(stand_host, auth_token)
x_user_id = _fetch_x_user_id(http_client)
group_state["stand_host"] = stand_host
group_state["auth_token"] = auth_token
group_state["x_user_id"] = x_user_id
group_state["auth_suite"] = suite_name
logger.info("[AUTH] Инициализация auth для dataset '%s'", suite_name)
def ensure_auth_for_fixture(group_state: dict) -> None:
"""
Подстраховка для фикстур http_client / ws_client.
Основная инициализация auth выполняется в pytest_runtest_setup при смене dataset.
Если тест по какой-то причине стартует раньше (нет current_suite), здесь
выполняется инициализация по auth_suite или fallback '__default__'.
"""
suite_name = group_state.get("current_suite") or group_state.get("auth_suite") or "__default__"
ensure_suite_auth(group_state, suite_name)
def _require_auth(group_state: dict, *, require_x_user_id: bool = False) -> None:
"""
Проверяет, что auth уже инициализирован в group_state.
При отсутствии stand_host/auth_token (и x_user_id для WS) завершает прогон через pytest.exit.
"""
if not group_state.get("auth_token") or not group_state.get("stand_host"):
pytest.exit("[AUTH] auth не инициализирован - ожидается ensure_suite_auth()")
if require_x_user_id and not group_state.get("x_user_id"):
pytest.exit("[AUTH] x_user_id не инициализирован - ожидается ensure_suite_auth()")
def init_http_stand_client(group_state: dict) -> StandHttpClient:
"""Создает StandHttpClient. Креды из group_state (кэш на dataset)."""
_require_auth(group_state)
return StandHttpClient(group_state["stand_host"], group_state["auth_token"])
def init_ws_stand_client(group_state: dict) -> WebSocketClient:
"""Создает WebSocketClient. Креды из group_state (кэш на dataset)."""
_require_auth(group_state, require_x_user_id=True)
return WebSocketClient(
group_state["stand_host"],
group_state["auth_token"],
group_state["x_user_id"],
)
rejection_report_xlsx_utils.py
"""
Утилиты для разбора xlsx-отчёта об отбракованных входных данных.
"""
from __future__ import annotations
from dataclasses import dataclass, replace
from datetime import datetime, timedelta
from typing import Iterable, List, Optional, Tuple
from constants.enums import RejectionSensorTag
from constants.test_constants import ExportRejectedReportConstants as RejectedReportConst
from test_config.models_for_tests import RejectionReportRow, RejectionTestCase
from utils.helpers import report_xlsx_utils as report_utils
from utils.helpers.lds_status_report_xlsx_utils import format_duration_seconds, parse_duration_seconds
from utils.helpers.ws_test_utils import localize_as_moscow
MergeKey = Tuple[Optional[datetime], str, str, str, str]
@dataclass
class RejectionReportCaseCheck:
"""Подготовленные данные для проверки одного RejectionTestCase в xlsx-отчёте."""
case_label: str
tag_description: str
report_event: str
window_start: datetime
window_end: datetime
row_found: bool
found_row_summary: str
datetime_in_window: bool = False
datetime_actual_text: str = "(пусто)"
actual_duration_seconds: int = 0
expected_duration_seconds: int = 0
expected_duration_text: str = ""
pipe_section: str = ""
actual_signal_suffix: str = ""
expected_signal_suffix: str = ""
def prepare_rejection_report_case_checks(
monitored_rows: Iterable[RejectionReportRow],
rejection_cases: Iterable[RejectionTestCase],
imitator_start_time: datetime,
) -> List[RejectionReportCaseCheck]:
"""Собирает все вычисленные значения для проверки строк отчёта по кейсам набора."""
case_checks: List[RejectionReportCaseCheck] = []
for rejection_case in rejection_cases:
if rejection_case.expected_is_rejected:
report_event = expected_event_to_report_event(rejection_case.expected_event)
window_start, window_end = get_case_time_window(imitator_start_time, rejection_case)
case_label = f"события '{report_event}' - {rejection_case.sensor.description}"
expected_signal_suffix = report_signal_suffix_by_expected_name(rejection_case.expected_signal_name)
raw_case_rows = filter_rows_for_rejection_case(monitored_rows, rejection_case, imitator_start_time)
merged_case_rows = merge_rejection_rows(raw_case_rows)
primary_row = select_primary_merged_row(merged_case_rows)
if primary_row is None:
case_checks.append(
RejectionReportCaseCheck(
case_label=case_label,
tag_description=rejection_case.sensor.description,
report_event=report_event,
window_start=window_start,
window_end=window_end,
row_found=False,
found_row_summary="строка не найдена",
expected_signal_suffix=expected_signal_suffix,
)
)
continue
merge_key = build_merge_key(primary_row)
expected_duration_seconds = sum_duration_for_merge_key(raw_case_rows, merge_key)
pipe_section, actual_signal_suffix = split_object_column(primary_row.object_value)
datetime_in_window = is_datetime_within_closed_interval(
primary_row.datetime_value,
window_start,
window_end,
)
case_checks.append(
RejectionReportCaseCheck(
case_label=case_label,
tag_description=rejection_case.sensor.description,
report_event=report_event,
window_start=window_start,
window_end=window_end,
row_found=True,
found_row_summary=(
f"{primary_row.tag_value} | {primary_row.event_value} | {primary_row.datetime_value}"
),
datetime_in_window=datetime_in_window,
datetime_actual_text=str(primary_row.datetime_value) if primary_row.datetime_value else "(пусто)",
actual_duration_seconds=primary_row.duration_seconds,
expected_duration_seconds=expected_duration_seconds,
expected_duration_text=format_duration_seconds(expected_duration_seconds),
pipe_section=pipe_section,
actual_signal_suffix=actual_signal_suffix,
expected_signal_suffix=expected_signal_suffix,
)
)
return case_checks
def expected_event_to_report_event(expected_event: str) -> str:
"""Преобразует формулировку события из журнала в формулировку отчёта."""
return expected_event.replace("Отбраковка", "Отбракован", 1)
def report_signal_suffix_by_expected_name(expected_signal_name: str) -> str:
"""Возвращает суффикс сигнала в колонке 'Объект' отчёта по expected_signal_name из кейса."""
return RejectedReportConst.REPORT_SIGNAL_SUFFIX_BY_EXPECTED_NAME.get(
expected_signal_name,
expected_signal_name,
)
def split_object_column(object_value: str) -> tuple[str, str]:
"""
Разбирает колонку "Объект":
- до последней точки - участок трубопровода (имя объекта);
- после последней точки - название сигнала.
"""
if not object_value:
return "", ""
if RejectedReportConst.OBJECT_SIGNAL_SEPARATOR not in object_value:
return object_value.strip(), ""
pipe_section, signal_suffix = object_value.rsplit(
RejectedReportConst.OBJECT_SIGNAL_SEPARATOR,
RejectedReportConst.OBJECT_SIGNAL_RSPLIT_MAXSPLIT,
)
return pipe_section.strip(), signal_suffix.strip()
def is_datetime_within_closed_interval(
value: datetime,
interval_start: datetime,
interval_end: datetime,
) -> bool:
"""True, если value (в Europe/Moscow) попадает в закрытый интервал [interval_start, interval_end]."""
localized_value = localize_as_moscow(value)
return interval_start <= localized_value <= interval_end
def report_header_contains_expected_title(raw_title: str) -> bool:
"""Проверяет, что первая строка шапки xlsx содержит ожидаемый заголовок отчёта."""
title_lower = raw_title.lower()
return (
RejectedReportConst.REJECTED_REPORT_HEADER_TITLE_PART in title_lower
or RejectedReportConst.REJECTED_REPORT_HEADER_TITLE_PART_ALT in title_lower
)
def build_merge_key(row: RejectionReportRow) -> MergeKey:
"""Ключ объединения строк с одинаковым содержимым, кроме длительности."""
return (
row.datetime_value,
row.object_value,
row.event_value,
row.value_text,
row.tag_value,
)
def parse_rejection_report_row(row_index: int, cells: dict[str, str]) -> RejectionReportRow:
"""Собирает RejectionReportRow из словаря ячеек строки отчёта."""
duration_seconds = parse_duration_seconds(cells.get(RejectedReportConst.COL_DURATION)) or 0
return RejectionReportRow(
row_index=row_index,
datetime_value=report_utils.parse_report_datetime(cells.get(RejectedReportConst.COL_DATETIME)),
object_value=(cells.get(RejectedReportConst.COL_OBJECT) or "").strip(),
event_value=(cells.get(RejectedReportConst.COL_EVENT) or "").strip(),
value_text=(cells.get(RejectedReportConst.COL_VALUE) or "").strip(),
duration_seconds=duration_seconds,
tag_value=(cells.get(RejectedReportConst.COL_TAG) or "").strip(),
)
def iter_rejection_report_rows(worksheet) -> List[RejectionReportRow]:
"""Возвращает строки данных отчёта, начиная с третьей строки листа."""
headers = report_utils.get_report_column_headers(
worksheet,
headers_row=RejectedReportConst.REPORT_COLUMN_HEADERS_ROW,
)
if not headers:
return []
rows: List[RejectionReportRow] = []
for excel_row_index, row_values in enumerate(
worksheet.iter_rows(
min_row=RejectedReportConst.REPORT_DATA_FIRST_ROW,
max_col=len(headers),
values_only=True,
),
start=RejectedReportConst.REPORT_DATA_FIRST_ROW,
):
if not any(cell is not None and str(cell).strip() for cell in row_values):
continue
cells = report_utils.build_column_cells(row_values, headers)
rows.append(parse_rejection_report_row(excel_row_index, cells))
return rows
def filter_rows_by_monitored_tags(
rows: Iterable[RejectionReportRow],
monitored_tags: Iterable[RejectionSensorTag],
) -> List[RejectionReportRow]:
"""Оставляет только строки с тегами из RejectionSensorTag."""
allowed_tags = {tag.description for tag in monitored_tags}
return [row for row in rows if row.tag_value in allowed_tags]
def get_case_time_window(
imitator_start_time: datetime,
rejection_case: RejectionTestCase,
tolerance_seconds: int = RejectedReportConst.TIME_FILTER_TOLERANCE_SECONDS,
) -> tuple[datetime, datetime]:
"""Возвращает окно фильтрации строк отчёта для конкретного RejectionTestCase."""
imitator_msk = localize_as_moscow(imitator_start_time)
window_start = imitator_msk + timedelta(seconds=rejection_case.time_range_start_s - tolerance_seconds)
window_end = imitator_msk + timedelta(seconds=rejection_case.time_range_end_s + tolerance_seconds)
return window_start, window_end
def filter_rows_for_rejection_case(
rows: Iterable[RejectionReportRow],
rejection_case: RejectionTestCase,
imitator_start_time: datetime,
) -> List[RejectionReportRow]:
"""Фильтрует строки отчёта по тегу, событию и временному окну RejectionTestCase."""
report_event = expected_event_to_report_event(rejection_case.expected_event)
window_start, window_end = get_case_time_window(imitator_start_time, rejection_case)
filtered_rows: List[RejectionReportRow] = []
for row in rows:
if row.tag_value != rejection_case.sensor.description:
continue
if row.event_value != report_event:
continue
if row.datetime_value is None:
continue
if not is_datetime_within_closed_interval(row.datetime_value, window_start, window_end):
continue
filtered_rows.append(row)
return filtered_rows
def merge_rejection_rows(rows: Iterable[RejectionReportRow]) -> List[RejectionReportRow]:
"""
Объединяет полностью идентичные строки, суммируя длительность отбраковки.
"""
merged_rows: dict[MergeKey, RejectionReportRow] = {}
for row in rows:
merge_key = build_merge_key(row)
if merge_key not in merged_rows:
merged_rows[merge_key] = replace(row)
continue
merged_rows[merge_key].duration_seconds += row.duration_seconds
return list(merged_rows.values())
def select_primary_merged_row(merged_rows: List[RejectionReportRow]) -> Optional[RejectionReportRow]:
"""Выбирает основную строку отбраковки - с максимальной суммарной длительностью."""
if not merged_rows:
return None
return max(merged_rows, key=lambda row: row.duration_seconds)
def sum_duration_for_merge_key(rows: Iterable[RejectionReportRow], merge_key: MergeKey) -> int:
"""Суммирует длительности всех сырых строк с одинаковым ключом объединения."""
return sum(row.duration_seconds for row in rows if build_merge_key(row) == merge_key)
def format_rejection_rows_for_allure(rows: Iterable[RejectionReportRow]) -> str:
"""Форматирует строки отчёта для вложения в Allure."""
lines = []
for row in rows:
duration_text = format_duration_seconds(row.duration_seconds)
lines.append(
f"row#{row.row_index}: {row.datetime_value} | {row.object_value} | "
f"{row.event_value} | {row.value_text} | {duration_text} | {row.tag_value}"
)
return "\n".join(lines) if lines else "(нет строк)"
report_xlsx_utils.py
"""
Утилиты для разбора xlsx-отчётов и проверки их формата.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional
import allure
from openpyxl import load_workbook
from openpyxl.worksheet.worksheet import Worksheet
from constants.test_constants import BaseTN3Constants as TestConst
from constants.test_constants import ExportReportConstants as ReportConst
from utils.helpers.ws_test_utils import extract_first_number, localize_as_moscow
@dataclass
class ReportTitleInfo:
"""Разобранная шапка отчёта"""
raw_title: str
period_start: Optional[datetime] = None
period_end: Optional[datetime] = None
@dataclass
class LeakReportRow:
"""Разобранная строка данных по утечке"""
row_index: int
cells: Dict[str, str] = field(default_factory=dict)
@property
def datetime_value(self) -> Optional[datetime]:
return parse_report_datetime(self.cells.get(ReportConst.COL_DATETIME))
@property
def object_value(self) -> str:
return self.cells.get(ReportConst.COL_OBJECT, "")
@property
def lds_status(self) -> str:
return self.cells.get(ReportConst.COL_LDS_STATUS, "")
@property
def masking_info(self) -> str:
return self.cells.get(ReportConst.COL_MASK_INFO, "")
@property
def coordinate_meters(self) -> Optional[float]:
coordinate_km = extract_first_number(self.cells.get(ReportConst.COL_COORDINATE))
if coordinate_km is None:
return None
return coordinate_km * TestConst.KM_TO_METERS
@property
def leak_volume(self) -> Optional[float]:
return extract_first_number(self.cells.get(ReportConst.COL_LEAK_VOLUME))
@property
def mt_mode(self) -> str:
return self.cells.get(ReportConst.COL_MT_MODE, "")
def is_xlsx_file_bytes(file_bytes: Optional[bytes]) -> bool:
"""Проверяет zip-сигнатуру xlsx"""
if not file_bytes:
return False
return file_bytes.startswith(ReportConst.ZIP_SIGNATURE)
def is_xlsx_extension(file_name: str) -> bool:
"""Проверяет расширение .xlsx без учёта регистра."""
return file_name.lower().endswith(ReportConst.XLSX_EXTENSION)
def parse_report_datetime(value: object) -> Optional[datetime]:
"""Парсит дату/время из ячейки отчёта."""
if value is None:
return None
if isinstance(value, datetime):
return value
if isinstance(value, str):
try:
return datetime.strptime(value.strip(), ReportConst.REPORT_DATETIME_FORMAT)
except ValueError:
return None
return None
def _stringify_cell(value: object) -> str:
if value is None:
return ""
if isinstance(value, datetime):
return value.strftime(ReportConst.REPORT_DATETIME_FORMAT)
return str(value)
def normalize_report_period_naive(value: datetime) -> datetime:
"""Московское время без tzinfo и микросекунд - для сравнения периодов в отчёте."""
return localize_as_moscow(value).replace(microsecond=0, tzinfo=None)
def report_period_comparison_bounds(
period_start: datetime,
period_end: datetime,
tolerance_minutes: int = ReportConst.REPORT_PERIOD_TOLERANCE_MINUTES,
) -> tuple[datetime, datetime, datetime, datetime]:
"""
Границы периода с допуском +-tolerance_minutes для start и end отдельно.
Возвращает (start_lower, start_upper, end_lower, end_upper).
"""
start = normalize_report_period_naive(period_start)
end = normalize_report_period_naive(period_end)
delta = timedelta(minutes=tolerance_minutes)
return start - delta, start + delta, end - delta, end + delta
def build_export_report_file_name(
tu_description: str,
period_start: datetime,
period_end: datetime,
report_name_part: str = ReportConst.LEAKS_REPORT_NAME_PART,
name_tu_separator: str = " ",
) -> str:
"""
Имя xlsx при скачивании: '{название}{sep}{ТУ} DD.MM.YYYY HH_MM_SS - DD.MM.YYYY HH_MM_SS.xlsx'.
По умолчанию - отчёт об утечках.
"""
start_text = normalize_report_period_naive(period_start).strftime(ReportConst.REPORT_FILE_NAME_DATETIME_FORMAT)
end_text = normalize_report_period_naive(period_end).strftime(ReportConst.REPORT_FILE_NAME_DATETIME_FORMAT)
return (
f"{report_name_part}{name_tu_separator}{tu_description} {start_text} - {end_text}"
f"{ReportConst.XLSX_EXTENSION}"
)
def parse_period_from_export_file_name(
file_name: str,
period_pattern: str | None = None,
) -> tuple[Optional[datetime], Optional[datetime]]:
"""Извлекает границы периода из имени скачанного xlsx-файла."""
match = re.search(
period_pattern or ReportConst.REPORT_FILE_NAME_PERIOD_PATTERN,
file_name.strip(),
re.IGNORECASE,
)
if match is None:
return None, None
parse_format = ReportConst.REPORT_FILE_NAME_DATETIME_FORMAT.replace("_", ":")
def _parse_part(value: str) -> Optional[datetime]:
try:
return datetime.strptime(value.replace("_", ":"), parse_format)
except ValueError:
return None
return _parse_part(match.group("period_start")), _parse_part(match.group("period_end"))
def parse_report_title(
title_raw: object,
header_period_pattern: str | None = None,
) -> ReportTitleInfo:
"""
Парсит шапку отчёта с именованными группами period_start/period_end.
"""
title_str = _stringify_cell(title_raw)
pattern = header_period_pattern or ReportConst.REPORT_HEADER_PERIOD_PATTERN
match = re.search(pattern, title_str)
if match is None:
return ReportTitleInfo(raw_title=title_str)
return ReportTitleInfo(
raw_title=title_str,
period_start=parse_report_datetime(match.group("period_start")),
period_end=parse_report_datetime(match.group("period_end")),
)
def load_report_worksheet(file_path: Path) -> Optional[Worksheet]:
"""Открывает первый лист xlsx. При ошибке возвращает None."""
if not file_path.exists():
return None
try:
workbook = load_workbook(filename=str(file_path), read_only=True, data_only=True)
except Exception:
return None
sheet_names = workbook.sheetnames
if not sheet_names:
return None
return workbook[sheet_names[ReportConst.DEFAULT_SHEET_INDEX]]
def get_report_title_cell(worksheet: Worksheet) -> object:
return worksheet.cell(row=ReportConst.REPORT_TITLE_ROW, column=1).value
def get_report_column_headers(
worksheet: Worksheet,
headers_row: int = ReportConst.REPORT_COLUMN_HEADERS_ROW,
) -> List[str]:
"""Возвращает непустые заголовки колонок из указанной строки шапки."""
headers: List[str] = []
column_index = 1
while True:
cell_value = worksheet.cell(row=headers_row, column=column_index).value
if cell_value is None or not str(cell_value).strip():
break
headers.append(_stringify_cell(cell_value).strip())
column_index += 1
return headers
def build_column_cells(row_values: tuple, headers: List[str]) -> Dict[str, str]:
"""Собирает словарь {название колонки: значение ячейки} по строке данных."""
return {
header: _stringify_cell(row_values[column_index]) if column_index < len(row_values) else ""
for column_index, header in enumerate(headers)
}
def iter_report_data_rows(worksheet: Worksheet) -> List[LeakReportRow]:
"""
Возвращает строки данных по утечкам, начиная с REPORT_DATA_FIRST_ROW.
Пустые строки пропускаются.
"""
headers = get_report_column_headers(worksheet)
if not headers:
return []
rows: List[LeakReportRow] = []
for excel_row_index, row_values in enumerate(
worksheet.iter_rows(
min_row=ReportConst.REPORT_DATA_FIRST_ROW,
max_col=len(headers),
values_only=True,
),
start=ReportConst.REPORT_DATA_FIRST_ROW,
):
if not any(cell is not None and str(cell).strip() for cell in row_values):
continue
rows.append(
LeakReportRow(
row_index=excel_row_index,
cells=build_column_cells(row_values, headers),
)
)
return rows
def find_row_with_object(rows: List[LeakReportRow], object_substring: str) -> Optional[LeakReportRow]:
"""Ищет первую строку, где колонка 'Объект' содержит подстроку без учёта регистра"""
substring_lower = object_substring.lower()
for row in rows:
if substring_lower in row.object_value.lower():
return row
return None
def save_report_bytes_to_temp_file(
file_bytes: bytes,
prefix: str = "leaks_report_",
) -> Optional[Path]:
"""Сохраняет байты отчёта во временный xlsx-файл. При ошибке возвращает None."""
import tempfile
try:
with tempfile.NamedTemporaryFile(
suffix=ReportConst.XLSX_EXTENSION,
prefix=prefix,
delete=False,
) as temp_file:
temp_file.write(file_bytes)
return Path(temp_file.name)
except OSError:
return None
def attach_report_file_to_allure(file_path: Path, file_name: str) -> None:
"""Прикладывает xlsx к Allure при падении теста"""
try:
xlsx_type = allure.attachment_type.XLSX
except AttributeError:
xlsx_type = None
if xlsx_type is not None:
allure.attach.file(
str(file_path),
name=file_name,
attachment_type=xlsx_type,
extension="xlsx",
)
return
try:
with file_path.open("rb") as raw_file:
allure.attach(raw_file.read(), name=file_name, extension="xlsx")
except OSError:
pass
def read_worksheet_cell_value(
file_path: Path,
row: int,
column: int,
*,
data_only: bool = True,
sheet_index: int = ReportConst.DEFAULT_SHEET_INDEX,
) -> object:
"""
Читает значение ячейки из xlsx.
Для формул используем data_only=False, оказалось иначе openpyxl вернёт вычисленное значение.
"""
if not file_path.exists():
return None
try:
workbook = load_workbook(filename=str(file_path), read_only=True, data_only=data_only)
except Exception:
return None
sheet_names = workbook.sheetnames
if not sheet_names:
return None
worksheet = workbook[sheet_names[sheet_index]]
return worksheet.cell(row=row, column=column).value
def sum_duration_columns_across_rows(
section_rows: list,
mode_duration_columns: list[str],
) -> dict[str, int]:
"""Суммирует длительности по колонкам режимов для всех строк участков."""
totals = {column_name: 0 for column_name in mode_duration_columns}
for section_row in section_rows:
for column_name, duration_seconds in section_row.mode_durations_seconds.items():
totals[column_name] += duration_seconds
return totals
signal_unit_conversion_utils.py
import copy
from typing import Any
from constants.enums import MeasureConversionRule
from constants.test_constants import MeasureUnitConstants
def _resolve_unit_mapping(rule: MeasureConversionRule) -> tuple[str, str]:
if rule == MeasureConversionRule.MPA_MEASURE:
return MeasureUnitConstants.KG_CM_MEASURE, MeasureUnitConstants.MPA_MEASURE
if rule == MeasureConversionRule.KG_CM_MEASURE:
return MeasureUnitConstants.MPA_MEASURE, MeasureUnitConstants.KG_CM_MEASURE
raise ValueError(f"Неизвестное правило конвертации единиц измерения: {rule}")
def conversion_rules_need_update(rules_json: dict[str, Any], rule: MeasureConversionRule) -> bool:
"""
Проверяет, есть ли в файле единицы давления, которые нужно заменить по правилу набора.
"""
source_unit, _ = _resolve_unit_mapping(rule)
return any(signal.get("OriginUnit") == source_unit for signal in rules_json.get("Signals", []))
def apply_measure_conversion_rule(rules_json: dict[str, Any], rule: MeasureConversionRule) -> dict[str, Any]:
"""
Возвращает копию rules_json с заменой единиц давления и число заменённых сигналов.
Замена строго по полному совпадению OriginUnit с исходной единицей давления
(kgf/cm^2 <-> MPa). Остальные единицы (cSt, m^3/h, rpm и т.д.) не затрагиваются.
"""
source_unit, target_unit = _resolve_unit_mapping(rule)
result = copy.deepcopy(rules_json)
for signal in result.get("Signals", []):
origin_unit = signal.get("OriginUnit")
if origin_unit == source_unit:
signal["OriginUnit"] = target_unit
return result
vault_pytest_utils.py
"""
Интеграция Vault с setup/teardown.
Настраивает ProcessEmptyValuesRejection перед setup стенда и сбрасывает после
завершения набора отбраковки
"""
from __future__ import annotations
import logging
import os
from typing import Any
from constants.architecture_constants import EnvKeyConstants, VaultConstants
from infra.vault_config_manager import VaultConfigManager
logger = logging.getLogger(__name__)
VAULT_CREDENTIALS_MISSING_MESSAGE = (
"[SETUP] [ERROR] Vault не настроен для набора отбраковки (требуются VAULT_ADDR, ROLE_ID, SECRET_ID, STAND_NAME)"
)
def suite_requires_process_empty_values_rejection(suite_config: Any) -> bool:
"""
Проверяет, нужен ли для набора ProcessEmptyValuesRejection=true в Vault.
"""
return bool(getattr(suite_config, "requires_process_empty_values_rejection", False))
def has_vault_credentials() -> bool:
"""
Проверяет наличие обязательных переменных окружения для работы с Vault.
"""
return all(
os.environ.get(env_key)
for env_key in (
VaultConstants.VAULT_ADDR,
VaultConstants.ROLE_ID,
VaultConstants.SECRET_ID,
EnvKeyConstants.STAND_NAME,
)
)
def configure_vault_for_suite(suite_config: Any) -> tuple[bool, str | None]:
"""
Настраивает ProcessEmptyValuesRejection в Vault перед подготовкой стенда.
"""
requires_rejection = suite_requires_process_empty_values_rejection(suite_config)
if not has_vault_credentials():
if requires_rejection:
return False, VAULT_CREDENTIALS_MISSING_MESSAGE
logger.warning("[SETUP] [SKIP] Vault credentials не заданы - ProcessEmptyValuesRejection не изменен")
return False, None
enabled = requires_rejection
VaultConfigManager().set_process_empty_values_rejection(enabled)
logger.info(
"[VAULT] ProcessEmptyValuesRejection установлен в %s для набора",
"true" if enabled else "false",
)
return enabled, None
def reset_vault_process_empty_values_rejection() -> None:
"""
Сбрасывает ProcessEmptyValuesRejection в false.
"""
try:
if not has_vault_credentials():
return
manager = VaultConfigManager()
manager.ensure_disabled()
logger.info("[VAULT] ProcessEmptyValuesRejection сброшен в false")
except Exception as error:
logger.warning(
"[TEARDOWN] [ALERT] Не удалось сбросить ProcessEmptyValuesRejection в Vault: %s",
error,
)
ws_message_parser.py
from __future__ import annotations
import logging
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional, Type, TypeVar
from uuid import UUID
from zoneinfo import ZoneInfo
from allure import attach, attachment_type
from dacite import Config, DaciteError, from_dict
from msgpack import Timestamp
from pytest import fail
import models.subscribe_scheme_signals_state_model as signals_state_model
from constants.architecture_constants import WebSocketClientConstants as WS_Const
from constants.enums import ExportedDataType, ExportStatus, ReplyStatus
from constants.test_constants import BaseTN3Constants as Base_const
from models.acknowledge_leak_model import AcknowledgeLeakReply
from models.basic_info_model import BasicInfoReply
from models.export_reports_model import ReportDataExportedNotification
from models.get_basic_info_admin_model import GetBasicInfoAdminReply
from models.get_exported_files_list_model import GetExportedDataListReply
from models.get_input_signals_model import GetInputSignalsReply
from models.get_messages_model import GetMessagesReply
from models.get_output_signals_model import GetOutputSignalsReply
from models.get_tus_information_model import GetTusInformationReply
from models.imitate_signal_model import ImitateSignalReply
from models.launch_lds_model import LaunchLdsReply
from models.launch_pig_model import LaunchPigReply
from models.mask_lds_command_model import MaskLdsReply
from models.mask_signal_model import MaskSignalReply
from models.stop_lds_model import StopLdsReply
from models.subscribe_all_leaks_info_model import SubscribeAllLeaksInfoReply
from models.subscribe_balance_algorithm_results_model import SubscribeBalanceAlgorithmResultsReply
from models.subscribe_common_scheme_model import SubscribeCommonSchemeReply
from models.subscribe_input_signals_model import InputSignal, SubscribeInputSignalsContent, SubscribeInputSignalsReply
from models.subscribe_leaks_model import SubscribeLeaksReply
from models.subscribe_main_page_info_model import SubscribeMainPageInfoReply
from models.subscribe_main_page_signals_info_model import SubscribeMainPageSignalsInfoReply
from models.subscribe_output_signals_model import SubscribeOutputSignalsReply
from models.subscribe_tu_leaks_info_model import SubscribeTuLeaksInfoReply
from models.unimitate_signal_model import UnimitateSignalReply
from models.unmask_lds_command_model import UnmaskLdsReply
from models.unmask_signal_model import UnmaskSignalReply
from models.upload_exported_file_model import DownloadExportedDataReply
logger = logging.getLogger(__name__)
MessageType = TypeVar("MessageType") # создает типовую переменную для парсинга сообщений
ContentType = TypeVar("ContentType")
_SIGNAL_DATA_POSITION = 1
_MIN_SIGNAL_TUPLE_LENGTH = 2
class WsMessageParser:
"""
Парсинг websocket сообщений
"""
def __init__(self, dacite_config: Config = None):
self._dacite_config = dacite_config or self._get_default_config()
self.suppress_recv_logging: bool = False
@staticmethod
def timestamp_to_datetime(value: Any) -> Optional[datetime]:
"""
Преобразует время из формата пары msgpack. Timestamp и tz в datetime с timezone
"""
try:
if value is None:
return None
if isinstance(value, list) and len(value) == 2:
time_timestamp, timezone_offset = value
if isinstance(time_timestamp, Timestamp):
datetime_timezone = timezone(timedelta(minutes=timezone_offset))
return datetime.fromtimestamp(time_timestamp.seconds, datetime_timezone)
if isinstance(value, Timestamp):
return datetime.fromtimestamp(value.seconds, tz=timezone.utc)
if isinstance(value, str):
try:
return datetime.fromisoformat(value)
except ValueError:
return datetime.strptime(value.replace("Z", "+00:00"), Base_const.JOURNAL_TIME_FORMAT)
except (AttributeError, TypeError, ValueError) as error:
fail(f"Ошибка конвертации времени: {error}")
@staticmethod
def convert_to_uuid(value: Any) -> Optional[UUID]:
"""
Преобразует строку в UUID
"""
try:
if value is None:
return None
elif isinstance(value, UUID):
return value
elif isinstance(value, str):
return UUID(value)
except (AttributeError, TypeError, ValueError) as error:
fail(f"Ошибка конвертации UUID: {error}")
def parse_acknowledge_leak_msg(self, data: dict) -> AcknowledgeLeakReply:
"""
Парсит acknowledgeLeak сообщение
"""
return self._parse_message(data_class=AcknowledgeLeakReply, data=data)
def parse_all_leaks_info_msg(self, data: list) -> SubscribeAllLeaksInfoReply:
"""
Парсит allLeaksInfo сообщение
"""
return self._find_and_parse_message(data_class=SubscribeAllLeaksInfoReply, data=data)
def parse_basic_info_msg(self, data: dict) -> BasicInfoReply:
"""
Парсит basicInfo сообщение
"""
return self._parse_message(data_class=BasicInfoReply, data=data)
def parse_common_scheme_info_msg(self, data: list) -> SubscribeCommonSchemeReply:
"""
Парсит tuLeaksInfo сообщение
"""
return self._find_and_parse_message(data_class=SubscribeCommonSchemeReply, data=data)
def parse_imitate_signal_msg(self, data: dict) -> ImitateSignalReply:
"""
Парсит сообщение ImitateSignal
"""
return self._parse_message(data_class=ImitateSignalReply, data=data)
def parse_input_signals_info_msg(self, data: list) -> SubscribeInputSignalsReply:
"""
Парсит сообщение InputSignalsInfo
"""
payload = self.find_reply_status_in_ws_msg(data)
reply_content = payload.get('replyContent')
input_signals_list = reply_content.get('inputSignals')
parsed_payload = SubscribeInputSignalsReply(
replyStatus=payload.get('replyStatus'),
replyErrors=payload.get('replyErrors'),
replyContent=SubscribeInputSignalsContent(
tuId=reply_content.get('tuId'),
inputSignals=[
self._parse_message(InputSignal, item[_SIGNAL_DATA_POSITION])
for item in input_signals_list
if self._is_valid_signal_tuple(item)
],
),
)
if parsed_payload.replyErrors:
fail(f"Ошибка в сообщении типа InputSignalsInfo: {parsed_payload.replyErrors}")
return parsed_payload
def parse_input_signals_msg(self, data: list) -> GetInputSignalsReply:
"""
Парсит сообщение getInputSignals
"""
return self._find_and_parse_message(data_class=GetInputSignalsReply, data=data)
def parse_journal_msg(self, data: dict) -> GetMessagesReply:
"""
Парсит сообщение журнала messagesInfo
"""
return self._parse_message(data_class=GetMessagesReply, data=data)
def parse_leaks_content_msg(self, data: list) -> SubscribeLeaksReply:
"""
Парсит LeaksContent сообщение
"""
return self._find_and_parse_message(data_class=SubscribeLeaksReply, data=data)
def parse_main_page_msg(self, data: list) -> SubscribeMainPageInfoReply:
"""
Парсит сообщение mainPageInfo
"""
return self._find_and_parse_message(data_class=SubscribeMainPageInfoReply, data=data)
def parse_main_page_signals_msg(self, data: list) -> SubscribeMainPageSignalsInfoReply:
"""
Парсит сообщение mainPageSignalsInfo
"""
return self._find_and_parse_message(data_class=SubscribeMainPageSignalsInfoReply, data=data)
def parse_launch_pig_msg(self, data: list) -> LaunchPigReply:
"""
Парсит сообщение LaunchPig
"""
return self._find_and_parse_message(data_class=LaunchPigReply, data=data)
def parse_mask_signal_msg(self, data: list) -> MaskSignalReply:
"""
Парсит сообщение MaskSignal
"""
return self._find_and_parse_message(data_class=MaskSignalReply, data=data)
def parse_mask_lds_message(self, data: list) -> MaskLdsReply:
"""
Парсит сообщение maskLDSRequest
"""
return self._find_and_parse_message(data_class=MaskLdsReply, data=data)
def parse_unmask_lds_message(self, data: list) -> UnmaskLdsReply:
"""
Парсит сообщение UnmaskLDSRequest
"""
return self._find_and_parse_message(data_class=UnmaskLdsReply, data=data)
def parse_output_signals_info_msg(self, data: list) -> SubscribeOutputSignalsReply:
"""
Парсит OutputSignalsInfo сообщение
"""
return self._find_and_parse_message(data_class=SubscribeOutputSignalsReply, data=data)
def parse_output_signals_msg(self, data: dict):
"""
Парсит getOutputSignals сообщение
"""
return self._parse_message(data_class=GetOutputSignalsReply, data=data)
def parse_balance_algorithm_msg(self, data: list) -> SubscribeBalanceAlgorithmResultsReply:
"""
Парсит BalanceAlgorithmResults сообщение
"""
return self._find_and_parse_message(data_class=SubscribeBalanceAlgorithmResultsReply, data=data)
def parse_tu_leaks_info_msg(self, data: list) -> SubscribeTuLeaksInfoReply:
"""
Парсит tuLeaksInfo сообщение
"""
return self._find_and_parse_message(data_class=SubscribeTuLeaksInfoReply, data=data)
def parse_scheme_signals_state_msg(self, data: list) -> signals_state_model.SchemeSignalsStateReply:
"""
Парсит сообщение SchemeSignalsStateContent.
signalsStates приходят как [[signal_type, data_dict], ...] - конвертируем кортежи в словари.
"""
payload = self.find_reply_status_in_ws_msg(data)
reply_content = payload.get('replyContent') or {}
raw_signals = reply_content.get('signalsStates')
if raw_signals is None:
raw_signals = payload.get('signalsStates') or []
raw_to_states = reply_content.get('toStates')
if raw_to_states is None:
raw_to_states = payload.get('toStates') or []
tu_id = reply_content.get('tuId')
if tu_id is None:
tu_id = payload.get('tuId', 0)
parsed_payload = signals_state_model.SchemeSignalsStateReply(
replyStatus=payload.get('replyStatus', 0),
replyErrors=payload.get('replyErrors'),
replyContent=signals_state_model.SchemeSignalsStateContent(
tuId=tu_id,
signalsStates=self._parse_scheme_signal_states(raw_signals),
toStates=self._parse_scheme_to_states(raw_to_states),
),
)
if parsed_payload.replyErrors:
fail(f"Ошибка в сообщении типа SchemeSignalsStateContent: {parsed_payload.replyErrors}")
return parsed_payload
def parse_unimitate_signal_msg(self, data: dict) -> UnimitateSignalReply:
"""
Парсит сообщение UnimitateSignal
"""
return self._parse_message(data_class=UnimitateSignalReply, data=data)
def parse_unmask_signal_msg(self, data: list) -> UnmaskSignalReply:
"""
Парсит сообщение UnmaskSignal
"""
return self._find_and_parse_message(data_class=UnmaskSignalReply, data=data)
def parse_report_data_exported_notification_msg(self, data: list) -> ReportDataExportedNotification:
"""
Парсит пуш-нотификацию ReportDataExportedNotification о готовности отчёта.
"""
return self._find_and_parse_message(data_class=ReportDataExportedNotification, data=data)
def parse_exported_data_list_msg(self, data: dict) -> GetExportedDataListReply:
"""
Парсит ответ GetExportedDataListReply со списком сформированных файлов.
"""
return self._parse_message(data_class=GetExportedDataListReply, data=data)
def parse_download_exported_data_msg(self, data: list) -> DownloadExportedDataReply:
"""
Парсит ответ DownloadExportedDataReply со скачиваемым контентом файла (fileChunk).
"""
return self._find_and_parse_message(data_class=DownloadExportedDataReply, data=data)
def parse_get_basic_info_admin_msg(self, data: dict) -> GetBasicInfoAdminReply:
"""Парсит ответ GetBasicInfoAdminRequest."""
return self._parse_message(data_class=GetBasicInfoAdminReply, data=data)
def parse_launch_lds_msg(self, data: list) -> LaunchLdsReply:
"""Парсит Completion-ответ LaunchLdsRequest."""
return self._find_and_parse_message(data_class=LaunchLdsReply, data=data)
def parse_stop_lds_msg(self, data: list) -> StopLdsReply:
"""Парсит Completion-ответ StopLdsRequest."""
return self._find_and_parse_message(data_class=StopLdsReply, data=data)
def parse_get_tus_information_msg(self, data: dict) -> GetTusInformationReply:
"""Парсит ответ GetTusInformationRequest."""
return self._parse_message(data_class=GetTusInformationReply, data=data)
def _find_and_parse_message(
self,
data_class: Type[ContentType],
data: List[Any],
config: Optional[Config] = None,
) -> ContentType:
"""
Ищет объект с replyStatus в ws сообщении и парсит его
"""
payload = self.find_reply_status_in_ws_msg(data)
parsed_payload = self._parse_message(data_class=data_class, data=payload, config=config)
return parsed_payload
def _parse_message(
self,
data_class: Type[MessageType],
data: dict,
config: Optional[Config] = None,
) -> MessageType:
"""
Универсальная функция парсинга сообщений
"""
data_class_name = data_class.__name__
if not data:
fail(f"Пустое сообщение типа: {data_class_name}")
error_message = data.get('replyErrors', None)
if error_message:
fail(f"Ошибка в сообщении типа: {data_class_name} текст ошибки: {error_message}")
try:
message = from_dict(
data_class=data_class, data=data, config=config or self._dacite_config # type: ignore[arg-type]
)
if not self.suppress_recv_logging:
try:
attach(
str(message) + f" {datetime.now(ZoneInfo(WS_Const.ZONE_INFO))}",
name=data_class_name,
attachment_type=attachment_type.TEXT,
)
except (KeyError, RuntimeError) as error:
logger.debug("Allure attach пропущен: %s", error)
return message
except DaciteError as error:
fail(f"Ошибка парсинга сообщения типа: {data_class_name} текст ошибки: {error}")
def _parse_scheme_signal_states(self, raw_signals: list) -> list[signals_state_model.SignalState]:
"""Разворачивает signalsStates: поддерживает формат [type, dict] и уже готовые dict."""
signals_states = []
for item in raw_signals:
if self._is_valid_signal_tuple(item):
signals_states.append(self._parse_message(signals_state_model.SignalState, item[_SIGNAL_DATA_POSITION]))
elif isinstance(item, dict):
signals_states.append(self._parse_message(signals_state_model.SignalState, item))
return signals_states
def _parse_scheme_to_states(self, raw_to_states: list) -> list[signals_state_model.ToState]:
"""Разворачивает toStates: поддерживает формат [type, dict] и уже готовые dict."""
to_states = []
for item in raw_to_states:
if self._is_valid_signal_tuple(item):
to_states.append(self._parse_message(signals_state_model.ToState, item[_SIGNAL_DATA_POSITION]))
elif isinstance(item, dict):
to_states.append(self._parse_message(signals_state_model.ToState, item))
return to_states
@staticmethod
def _is_valid_signal_tuple(item: Any) -> bool:
"""Проверяет, что элемент - кортеж/список [signal_type, signal_data_dict]."""
return (
isinstance(item, (list, tuple))
and len(item) >= _MIN_SIGNAL_TUPLE_LENGTH
and isinstance(item[_SIGNAL_DATA_POSITION], dict)
)
@staticmethod
def _create_attach_from_ws_message(ws_message) -> None:
"""
Создает txt вложение ws сообщения в аллюр
"""
str_message = str(ws_message)
try:
attach(
str_message,
name=f"Распакованное сообщение от api-gateway {datetime.now(ZoneInfo(WS_Const.ZONE_INFO))}",
attachment_type=attachment_type.TEXT,
)
except (KeyError, RuntimeError) as error:
logger.debug("Allure attach пропущен: %s", error)
def _should_suppress_recv_attach(self) -> bool:
if self.suppress_recv_logging:
return True
from utils.helpers import lds_configurator_utils as lds_cfg
return lds_cfg.is_configurator_flow_active()
def find_reply_status_in_ws_msg(self, data: List[Any]) -> Optional[Dict[str, Any]]:
"""
Ищет объект с replyStatus в ws сообщении
"""
if not data:
fail("Пустое сообщение")
if not self._should_suppress_recv_attach():
self._create_attach_from_ws_message(data)
try:
for item in reversed(data):
# 1) Если сам элемент — словарь с replyStatus
if isinstance(item, dict) and 'replyStatus' in item:
return item
# 2) Если элемент — список / кортеж — проверяем все элементы в нём
if isinstance(item, (list, tuple)):
for elem in item:
if isinstance(elem, dict) and 'replyStatus' in elem:
return elem
except (AttributeError, KeyError, TypeError, RuntimeError, ValueError):
fail("Не удалось найти replyStatus в сообщении")
def _get_default_config(self) -> Config:
"""
Получает конфиг с правилами обработки полей
"""
# TODO добавить strict=True, после выполнения задачи LDS-8792
def _to_export_status(value: Any) -> ExportStatus:
return value if isinstance(value, ExportStatus) else ExportStatus(value)
def _to_exported_data_type(value: Any) -> ExportedDataType:
return value if isinstance(value, ExportedDataType) else ExportedDataType(value)
def _to_reply_status(value: Any) -> ReplyStatus:
return value if isinstance(value, ReplyStatus) else ReplyStatus(value)
return Config(
type_hooks={
UUID: self.convert_to_uuid,
datetime: self.timestamp_to_datetime,
ExportStatus: _to_export_status,
ExportedDataType: _to_exported_data_type,
ReplyStatus: _to_reply_status,
}
)
# Создает экземпляр класса для удобства импорта
ws_message_parser = WsMessageParser()
ws_test_utils.py
from __future__ import annotations
import asyncio
import pprint
import random
import re
from collections import Counter
from dataclasses import asdict, is_dataclass
from datetime import datetime, timedelta, timezone
from enum import IntEnum, IntFlag
from typing import Any, Callable, List, Optional, Protocol, Set, Type, TypeVar
from zoneinfo import ZoneInfo
import allure
from msgpack import Timestamp as MsgpackTimestamp
from pytest import fail
from requests import Response
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.architecture_constants import WebSocketClientConstants as WS_Const
from constants.enums import (
ConfirmationStatus,
ExportStatus,
LdsStatus,
LdsStatusDegradation,
LdsStatusFaulty,
LdsStatusInitialization,
LeakStatus,
ReplyStatus,
SignalType,
StationaryReason,
StationaryStatus,
StoppedPumpingReason,
UnStationaryReason,
)
from constants.test_constants import BaseTN3Constants as TestConst
from constants.test_constants import ExportReportConstants as ReportConst
from models.export_reports_model import ReportDataExportedContent, ReportDataExportedNotification
from models.get_messages_model import GetMessagesRequest
from models.subscribe_all_leaks_info_model import SubscribeAllLeaksInfoReply
from models.subscribe_common_scheme_model import DiagnosticArea, FlowArea
from models.subscribe_leaks_model import Leak
from models.subscribe_main_page_info_model import MainPageLeakInfo
from utils.helpers.ws_message_parser import ws_message_parser
from utils.msgpack_utils.message_filters import is_desired_invocation_id, is_desired_type
ObjectType = TypeVar("ObjectType") # создает типовую переменную для поиска объектов в списке
RandomObjectType = TypeVar("RandomObjectType")
Event = TypeVar("Event")
ParsedPayloadType = TypeVar("ParsedPayloadType")
InputSignalType = TypeVar("InputSignalType")
SiteSignalsMap = dict[tuple[int, int], dict[int, str]]
class ControlledSiteSignalMessage(Protocol):
controlledSiteId: int
segmentId: int
signals: list
def convert_leak_volume_m3(volume: float) -> float:
"""
Преобразует объем утечки в м3/час
"""
# Округляет результат для читабельности
return round(volume * TestConst.MASS_KG, 3)
def datetime_minus_seconds(datetime_obj: datetime, delta_s: int) -> datetime:
"""
Вычитает время в секундах из datetime
"""
return (datetime_obj - timedelta(seconds=delta_s)).replace(microsecond=0)
def calculate_leak_start_time(imitator_start_time: datetime, leak_interval_seconds: int) -> Optional[datetime]:
"""
Рассчитывает время начала утечки на основе времени старта имитатора.
:param imitator_start_time: datetime объект времени старта имитатора
:param leak_interval_seconds: интервал от старта до утечки в секундах (LEAK_START_INTERVAL)
:return: datetime время ожидаемого начала утечки
"""
if not imitator_start_time:
return None
return (imitator_start_time + timedelta(seconds=leak_interval_seconds)).replace(microsecond=0)
def calculate_leak_end_time(
imitator_start_time: datetime, leak_interval_seconds: int, allowed_diff_seconds: int
) -> Optional[datetime]:
"""
Рассчитывает крайнее время обнаружения утечки (с учётом допустимой погрешности).
:param imitator_start_time: datetime объект времени старта имитатора
:param leak_interval_seconds: интервал от старта до утечки в секундах (LEAK_START_INTERVAL)
:param allowed_diff_seconds: допустимая погрешность времени обнаружения (ALLOWED_TIME_DIFF_SECONDS)
:return: datetime крайнее время обнаружения утечки
"""
if not imitator_start_time:
return None
total_seconds = leak_interval_seconds + allowed_diff_seconds
return (imitator_start_time + timedelta(seconds=total_seconds)).replace(microsecond=0)
def get_leak_time_window(
imitator_start_time: datetime, leak_interval_seconds: int, allowed_diff_seconds: int, detected_at_tz=None
) -> tuple[datetime, datetime]:
"""
Возвращает временное окно для проверки времени обнаружения утечки.
:param imitator_start_time: datetime объект времени старта имитатора
:param leak_interval_seconds: интервал от старта до утечки в секундах
:param allowed_diff_seconds: допустимая погрешность времени обнаружения
:param detected_at_tz: timezone из времени обнаружения утечки (опционально)
:return: tuple (leak_start_time, leak_end_time) для использования в is_between проверке
"""
leak_start = calculate_leak_start_time(imitator_start_time, leak_interval_seconds)
leak_end = calculate_leak_end_time(imitator_start_time, leak_interval_seconds, allowed_diff_seconds)
# Если передан timezone, применяем его к временам для корректного сравнения
if detected_at_tz:
leak_start = leak_start.replace(tzinfo=detected_at_tz)
leak_end = leak_end.replace(tzinfo=detected_at_tz)
return leak_start, leak_end
def ensure_moscow_timezone(input_datetime: datetime) -> None | datetime:
"""
Конвертирует datetime в московское время, если оно не в московской таймзоне.
:param input_datetime: datetime объект
:return: datetime в московской таймзоне
"""
if input_datetime is None:
return input_datetime
# Если datetime без timezone - считаем что это UTC
if input_datetime.tzinfo is None:
input_datetime = input_datetime.replace(tzinfo=timezone.utc)
# Конвертируем в московское время
return input_datetime.astimezone(ZoneInfo(TestConst.ZONE_INFO))
def report_time_offset_hours(tz_name: str = TestConst.ZONE_INFO) -> Optional[int]:
"""
Смещение часового пояса (часы от UTC) для поля timeOffset в запросах отчётов.
"""
now = datetime.now(ZoneInfo(tz_name))
utc_offset = now.utcoffset()
if utc_offset is None:
return None
return int(utc_offset.total_seconds() // TestConst.SECONDS_PER_HOUR)
def localize_as_moscow(input_datetime: datetime) -> None | datetime:
"""
Присваивает datetime московский часовой пояс без сдвига времени.
Если datetime уже имеет timezone - конвертирует в московское время.
"""
if input_datetime is None:
return input_datetime
moscow_tz = ZoneInfo(TestConst.ZONE_INFO)
if input_datetime.tzinfo is None:
return input_datetime.replace(tzinfo=moscow_tz)
return input_datetime.astimezone(moscow_tz)
def format_datetime_moscow(value: Optional[datetime]) -> str:
"""Строковое представление datetime в Europe/Moscow для вложений Allure."""
if value is None:
return "None"
return str(localize_as_moscow(value))
def get_rejection_time_window(
imitator_start_time: datetime,
start_seconds: int | float,
reserve_seconds: int | float = 0,
) -> tuple[datetime, datetime]:
"""
Возвращает временное окно для проверки сообщения об отбраковке.
"""
imitator_msk = localize_as_moscow(imitator_start_time)
range_start = imitator_msk + timedelta(seconds=start_seconds - reserve_seconds)
range_end = localize_as_moscow(datetime.now())
return range_start, range_end
def find_rejection_journal_message(
messages_info: List[ObjectType],
tag: str,
range_start: datetime,
range_end: datetime,
technological_section: str,
expected_event: str,
) -> tuple[list[ObjectType], ObjectType | None]:
"""
Фильтрует сообщения журнала по tag и временному диапазону,
затем ищет целевое сообщение по technologicalSection и event.
"""
time_filtered = [
msg for msg in messages_info if msg.tag == tag and range_start <= ensure_moscow_timezone(msg.time) <= range_end
]
time_filtered.sort(key=lambda msg: ensure_moscow_timezone(msg.time), reverse=True)
target_msg = next(
(
msg
for msg in time_filtered
if msg.technologicalSection == technological_section and msg.event.rstrip() == expected_event
),
None,
)
return time_filtered, target_msg
def get_random_item(item_list: List[RandomObjectType]) -> RandomObjectType:
"""
Получает случайный объект из списка
"""
if not item_list:
return None
try:
return random.choice(item_list)
except (TypeError, ValueError):
return None
def get_longest_flow_area(flow_areas: List[FlowArea]) -> Optional[FlowArea]:
"""
Получает самый протяженный участок карты течения по количеству ДУ из списка всех участков
"""
if not flow_areas:
return None
try:
longest_flow_area = max(flow_areas, key=lambda flow_area: len(flow_area.diagnosticAreas))
return longest_flow_area
except (TypeError, ValueError):
return None
def get_longest_flow_area_by_pipes(flow_areas: List[FlowArea]) -> Optional[FlowArea]:
"""
Получает самый протяженный участок карты течения по количеству pipesIds
"""
if not flow_areas:
return None
try:
return max(flow_areas, key=lambda flow_area: len(flow_area.pipesIds or []))
except (TypeError, ValueError, AttributeError):
return None
def filter_flow_areas_with_available_flow(flow_areas: List[FlowArea]) -> List[FlowArea]:
"""
Оставляет только участки карты течения с доступным течением
"""
return [flow_area for flow_area in flow_areas if getattr(flow_area, "isFlowAvailable", False)]
def get_diagnostic_area_base_ids() -> list[int]:
"""
Возвращает список id базовых диагностических участков из константы DIAGNOSTIC_AREA_BASE_IDS
"""
return [pair[0] for pair in TestConst.DIAGNOSTIC_AREA_BASE_IDS.values()]
def diagnostic_area_title(du_id: int) -> str:
"""
Возвращает строковое представление ДУ: id и название из DIAGNOSTIC_AREA_BASE_IDS
"""
for name, pair in TestConst.DIAGNOSTIC_AREA_BASE_IDS.items():
if pair[0] == du_id:
return f"{du_id} ({name})"
return str(du_id)
def determine_stationary_status_by_majority(statuses: list[StationaryStatus]) -> Optional[StationaryStatus]:
"""
Определяет общий режим МТ по правилу большинства ДУ в одном статусе.
При равенстве количества выбирается статус с максимальным значением enum.
"""
if not statuses:
return None
counts = Counter(statuses)
max_count = max(counts.values())
candidates = [status for status, count in counts.items() if count == max_count]
return max(candidates, key=lambda status: status.value)
def collect_stationary_statuses_from_output_signals(
controlled_site_signals: list[ControlledSiteSignalMessage], controlled_sites_dict: dict
) -> tuple[list[tuple[str, StationaryStatus]], list[str]]:
"""
Собирает stationaryStatus (SignalType.REGLU) по всем участкам КП-КП из OutputSignalsInfo
"""
all_signals_dict = extract_signal_on_site_and_segment(controlled_site_signals)
site_stationary_statuses: list[tuple[str, StationaryStatus]] = []
missed_controlled_sites: list[str] = []
for name, value in controlled_sites_dict.items():
stationary_status = get_signal_value(all_signals_dict, value, SignalType.REGLU)
if stationary_status:
if isinstance(stationary_status, StationaryStatus):
site_stationary_statuses.append((name, stationary_status))
else:
missed_controlled_sites.append(name)
return site_stationary_statuses, missed_controlled_sites
def determine_lds_status_by_priority(lds_status_set: Set[int]) -> Optional[int]:
"""
Определяет режим работы СОУ по приоритету и наличию режимов работы у ДУ на самом протяженном участки карты течений
"""
lds_status_priority = [
LdsStatus.FAULTY.value,
LdsStatus.INITIALIZATION.value,
LdsStatus.DEGRADATION.value,
LdsStatus.SERVICEABLE.value,
]
if not lds_status_set:
return None
try:
for status in lds_status_priority:
if status in lds_status_set:
return status
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_signal_type_by_address_suffix(signals_list: list, address_suffix: str) -> Optional[int]:
"""
Ищет в списке сигналов тип сигнала по части адреса
"""
if not signals_list:
return None
try:
for sensor_signal in signals_list:
if sensor_signal.address is not None and str(sensor_signal.address).endswith(address_suffix):
return sensor_signal.signalType
return None
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_signal_val_by_signal_type(signals_list: list, signal_type: int) -> Optional[str]:
"""
Ищет в списке сигналов значение сигнала по типу
"""
if not signals_list:
return None
try:
for sensor_signal in signals_list:
if sensor_signal.signalType is not None and sensor_signal.signalType == signal_type:
return sensor_signal.value
return None
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_object_by_field(item_list: List[ObjectType], field_name: str, value: Any) -> ObjectType:
"""
Ищет объект в списке объектов по значению одного из полей объекта
"""
if not item_list:
return None
try:
return next((item for item in item_list if getattr(item, field_name) == value))
except Exception:
return None
def find_object_by_few_fields(item_list: List[ObjectType], fields_dict: dict) -> ObjectType:
"""
Ищет объект в списке объектов по значениям нескольких полей
"""
if not item_list:
return None
return next(
(item for item in item_list if all(getattr(item, field) == value for field, value in fields_dict.items())), None
)
def extract_signal_on_site_and_segment(
controlled_site_messages: list[ControlledSiteSignalMessage],
) -> SiteSignalsMap:
"""
Ищет сигналы для всех пар КП-КП
"""
result: SiteSignalsMap = {}
for site_obj in controlled_site_messages:
key = (site_obj.controlledSiteId, site_obj.segmentId)
result[key] = {signal.signalType: signal.value for signal in site_obj.signals}
return result
def get_signal_value(
all_signals_dict: SiteSignalsMap,
site_kp_kp: tuple[int, int],
signals_id: int,
) -> StationaryStatus | LdsStatus | int | None:
"""
Получает значение сигналов.
Приводит значение к типу инт.
Кладет в обертку значения режима МТ и режима СОУ
"""
signals_value = all_signals_dict.get(site_kp_kp, {}).get(signals_id)
try:
int_signals_value = int(signals_value)
except (TypeError, ValueError):
return None
if signals_id == SignalType.REGLU:
return StationaryStatus(int_signals_value) if signals_value else None
if signals_id == SignalType.REGSOU:
return LdsStatus(int_signals_value) if signals_value else None
return int_signals_value
def parse_journal_event(event_value: str) -> tuple[str | None, str | None]:
"""
Разделяет строку события на имя и причину, вложенную в скобки
"""
if not event_value or not isinstance(event_value, str):
return None, None
# Паттерн для "Состояние режима (Причина)"
pattern = r'^([^\(]+?)\s*\(([^\)]+)\)\s*$'
match = re.match(pattern, event_value)
if match:
mode_part = match.group(1).strip()
reason_part = match.group(2).strip()
return mode_part if mode_part else None, reason_part if reason_part else None
# Если значение события не соответствует паттерну, возвращаю текст, если нет текста, тогда None
return event_value.strip(), None
def get_signal(site_message, signal_type):
if not site_message:
return None
return next((s for s in site_message.signals if s.signalType == signal_type.value), None)
def get_value(obj):
return getattr(obj, "value", None)
def find_confirmed_leaks(item_list: List[Leak]) -> List[Leak]:
"""Ищет подтвержденные утечки"""
try:
return [
item
for item in item_list
if item.confirmationStatus == ConfirmationStatus.CONFIRMED.value and item.detectedAt is not None
]
except (AttributeError, KeyError, TypeError, ValueError):
return []
def find_confirmed_leaks_on_main_page(item_list: List[MainPageLeakInfo]) -> List[MainPageLeakInfo]:
"""Ищет подтвержденные утечки"""
try:
return [
item
for item in item_list
if item.leakStatus == LeakStatus.CONFIRMED.value and item.leakDetectedAt is not None
]
except (AttributeError, KeyError, TypeError, ValueError):
return []
def find_diagnostic_area_by_id(flow_areas: List[FlowArea], id_value: Optional[int]) -> Optional[DiagnosticArea]:
"""
Ищет ДУ по id в списке участков карты течений, исключает дубликаты по количеству pipeIds
"""
candidates = []
if not flow_areas or not id_value:
return None
try:
for flow_area in flow_areas:
for diagnostic_area in flow_area.diagnosticAreas:
if diagnostic_area.id == id_value:
candidates.append(diagnostic_area)
if not candidates:
return None
elif len(candidates) == 1:
return candidates[0]
else:
# Среди дубликатов ищет ДУ с наибольшим количеством pipeIds
return max(candidates, key=lambda candidate: len(candidate.pipeIds))
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_diagnostic_area_by_pipe_id(flow_areas: List[FlowArea], pipe_id: int) -> Optional[DiagnosticArea]:
"""
Ищет ДУ по pipe id в списке участков карты течений, исключает дубликаты по количеству pipeIds
"""
candidates = []
if not flow_areas:
return None
try:
for flow_area in flow_areas:
for diagnostic_area in flow_area.diagnosticAreas:
if diagnostic_area.pipeIds and pipe_id in diagnostic_area.pipeIds:
candidates.append(diagnostic_area)
if not candidates:
return None
elif len(candidates) == 1:
return candidates[0]
else:
# Среди дубликатов ищет ДУ с наибольшим количеством pipeIds
return max(candidates, key=lambda candidate: len(candidate.pipeIds))
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_diagnostic_areas_by_ids(flow_areas: List[FlowArea], id_list: List[int]) -> List[DiagnosticArea]:
"""
Получает список ДУ из списка flow_areas по списку id
"""
diagnostic_areas = [
result
for diagnostic_area_id in id_list
if (result := find_diagnostic_area_by_id(flow_areas, diagnostic_area_id)) is not None
]
return diagnostic_areas
def find_diagnostic_areas_by_pipe_ids(flow_areas: List[FlowArea], id_list: List[int]) -> List[DiagnosticArea]:
"""
Получает список ДУ из списка flow_areas по списку pipe id
"""
diagnostic_areas = [
result for pipe_id in id_list if (result := find_diagnostic_area_by_pipe_id(flow_areas, pipe_id)) is not None
]
return diagnostic_areas
def find_base_diagnostic_areas(flow_areas: List[FlowArea]) -> List[DiagnosticArea]:
"""
Получает список базовых ДУ из списка flow_areas
"""
return find_diagnostic_areas_by_ids(flow_areas, get_diagnostic_area_base_ids())
def find_leak_by_coordinate(
leaks_list: List[ObjectType], expected_coordinate: float, tolerance: float = TestConst.ALLOWED_DISTANCE_DIFF_METERS
) -> ObjectType:
"""
Ищет утечку в списке по координатам с допустимой погрешностью
"""
if not leaks_list:
return None
for leak in leaks_list:
leak_coordinate = getattr(leak, "leakCoordinate")
if leak_coordinate is None or "":
continue
if abs(leak_coordinate - expected_coordinate) <= tolerance:
return leak
return None
def to_moscow_timezone(date_str: str) -> Optional[datetime]:
"""
Преобразует строку времени в московское время
"""
if not date_str or not date_str.strip():
return None
try:
if date_str.startswith(("'", '"', '')) or date_str.endswith(("'", '"', '')):
date_str = date_str.strip().strip("'").strip('"')
date_utc = datetime.strptime(date_str, TestConst.OUTPUT_TIME_FORMAT).replace(tzinfo=timezone.utc)
return date_utc.astimezone(ZoneInfo(TestConst.ZONE_INFO))
except (AttributeError, TypeError, ValueError):
return None
def create_dict_from_dataclass(cls: Type, **kwargs) -> Optional[dict]:
"""Создает словарь из экземпляра dataclass c нужными параметрами"""
if not is_dataclass(cls):
return None
instance = cls(**kwargs)
return asdict(instance)
def datetime_to_msgpack_timestamp(dt: datetime) -> list:
"""Конвертирует datetime в формат [Timestamp(seconds, nanoseconds), tz_offset] для отправки на бэкенд."""
return [MsgpackTimestamp(seconds=int(dt.timestamp()), nanoseconds=0), 0]
def datetime_to_iso_format(dt: datetime) -> str:
"""Конвертирует datetime в строку 2026-07-27T21:00:00.000Z для отправки по http на бэкенд."""
utc_datetime = dt.astimezone(timezone.utc)
return utc_datetime.isoformat(timespec='milliseconds').replace('+00:00', 'Z')
def create_journal_req_body(**kwargs) -> dict:
"""Создает дефолтные параметры запроса к журналу"""
result = create_dict_from_dataclass(GetMessagesRequest, **kwargs)
period = result.get('periodTime')
if period:
for key in ('start', 'end'):
if isinstance(period.get(key), datetime):
period[key] = datetime_to_iso_format(period[key])
return result
def extract_first_number(value: object) -> Optional[float]:
"""
Извлекает первое число из ячейки (int/float/str вида '55.55 км', '111.11 м3/ч').
"""
if value is None:
return None
if isinstance(value, (int, float)) and not isinstance(value, bool):
return float(value)
if isinstance(value, str):
matches = re.findall(TestConst.DIGITS_WITH_DOT_PATTERN, value)
if matches:
try:
return float(matches[0].replace(",", "."))
except ValueError:
return None
return None
def attach_ws_poll_failure(
collected_messages: List[Any],
total_wait_seconds: float,
expected_message_type: str,
) -> None:
"""Краткая сводка и pprint каждого WS-сообщения при таймауте поллинга."""
allure.attach(
"\n".join(
[
f"Таймаут ожидания: {total_wait_seconds} с",
f"Ожидаемый тип сообщения: {expected_message_type}",
f"Всего сообщений за период поллинга: {len(collected_messages)}",
]
),
name="WS poll timeout",
attachment_type=allure.attachment_type.TEXT,
)
for msg in collected_messages:
allure.attach(
pprint.pformat(msg, width=120, sort_dicts=False),
name="received ws message",
attachment_type=allure.attachment_type.TEXT,
)
def _attach_ws_reply_parse_failure(
reply_payload: Optional[Any],
request_name: str,
error: BaseException,
invocation_id: str = "",
) -> None:
"""Прикрепляет к Allure ответ бэка при ошибке парсинга."""
allure.attach(
"\n".join(
[
f"Запрос: {request_name}",
f"invocation_id: {invocation_id}",
f"Ошибка: {error}",
]
),
name="WS parse failure",
attachment_type=allure.attachment_type.TEXT,
)
if reply_payload is not None:
allure.attach(
pprint.pformat(reply_payload, width=120, sort_dicts=False),
name="received ws message",
attachment_type=allure.attachment_type.TEXT,
)
def _drain_recv_queue(ws_client: WebSocketClient) -> List[Any]:
"""Забирает все сообщения из очереди ws без блокирующего receive_by_..."""
messages: List[Any] = []
while not ws_client.recv_queue.empty():
try:
messages.append(ws_client.recv_queue.get_nowait())
except asyncio.QueueEmpty:
break
return messages
def _find_valid_report_export_notification(
messages: List[Any],
parser,
notification_type: str,
) -> Optional[ReportDataExportedNotification]:
"""Ищет среди уже полученных сообщений успешную ReportDataExportedNotification."""
for msg in messages:
if not isinstance(msg, list) or not is_desired_type(msg, notification_type):
continue
try:
notification = parser.parse_report_data_exported_notification_msg(msg)
except (ValueError, TypeError, KeyError):
continue
if notification.replyStatus != ReplyStatus.OK.value:
continue
content: Optional[ReportDataExportedContent] = notification.replyContent
if content is None:
continue
if content.exportStatus != ExportStatus.DONE:
continue
return notification
return None
async def poll_for_report_export_notification(
ws_client: WebSocketClient,
parser,
total_wait_seconds: float,
poll_interval_seconds: float,
) -> Optional[Any]:
"""
Собирает сообщения из очереди ws и ищет ReportDataExportedNotification с успешным exportStatus.
При таймауте прикрепляет к Allure все полученные за период сообщения.
"""
deadline = asyncio.get_event_loop().time() + total_wait_seconds
collected_messages: List[Any] = []
ws_client.suppress_recv_logging = True
parser.suppress_recv_logging = True
try:
while asyncio.get_event_loop().time() < deadline:
await asyncio.sleep(poll_interval_seconds)
batch = _drain_recv_queue(ws_client)
collected_messages.extend(batch)
notification = _find_valid_report_export_notification(
batch, parser, ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION
)
if notification is not None:
return notification
finally:
collected_messages.extend(_drain_recv_queue(ws_client))
ws_client.suppress_recv_logging = False
parser.suppress_recv_logging = False
attach_ws_poll_failure(
collected_messages,
total_wait_seconds,
ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION,
)
return None
def _find_ws_reply_by_invocation_id(messages: List[Any], invocation_id: str, parser) -> Optional[list]:
"""
Ищет последний ответ с заданным invocation_id и телом replyStatus.
"""
reply_payload = None
for msg in messages:
if not isinstance(msg, list) or not is_desired_invocation_id(msg, invocation_id):
continue
if parser.find_reply_status_in_ws_msg(msg):
reply_payload = msg
return reply_payload
async def poll_for_exported_file(
http_client: StandHttpClient,
parser,
list_limit: int,
expected_data_type: Any,
name_substring: str,
tu_name_substring: str,
period_start: datetime,
period_end: datetime,
total_wait_seconds: float,
poll_interval_seconds: float,
period_tolerance_minutes: int = ReportConst.REPORT_PERIOD_TOLERANCE_MINUTES,
) -> Optional[Any]:
"""
Периодически шлёт GetExportedDataListRequest, забирает ответы из очереди
по invocation_id среди всех накопленных сообщений.
При таймауте или ошибке парсинга прикрепляет к Allure полученные ответы.
"""
deadline = asyncio.get_event_loop().time() + total_wait_seconds
last_items_count = -1
collected_messages: List[Any] = []
request_name = HttpConst.GET_EXPORTED_DATA_LIST_URL_PATH
request_payload = {"limit": list_limit}
parser.suppress_recv_logging = True
try:
while asyncio.get_event_loop().time() < deadline:
response = http_client.post_request(request_name, request_payload)
if response.json():
collected_messages.extend(response.json())
payload = get_json_from_http_response(response)
try:
parsed_payload = parser.parse_exported_data_list_msg(payload)
except Exception as error:
_attach_ws_reply_parse_failure(payload, request_name, error)
for msg in collected_messages:
allure.attach(
pprint.pformat(msg, width=120, sort_dicts=False),
name="received http message",
attachment_type=allure.attachment_type.TEXT,
)
fail(f"Не удалось разобрать ответ на {request_name}: {error}")
items = []
reply_content = getattr(parsed_payload, "replyContent", None)
if reply_content:
items = getattr(reply_content, "exportedData", [])
if len(items) != last_items_count:
allure.attach(
"\n".join(
f"id={item.id}, name={item.name}, type={item.exportedDataType}, "
f"start={format_datetime_moscow(item.start)}, end={format_datetime_moscow(item.end)}"
for item in items
),
name=f"Список сформированных файлов (всего: {len(items)})",
attachment_type=allure.attachment_type.TEXT,
)
last_items_count = len(items)
match = find_matching_exported_item(
items=items,
expected_data_type=expected_data_type,
name_substring=name_substring,
tu_name_substring=tu_name_substring,
period_start=period_start,
period_end=period_end,
period_tolerance_minutes=period_tolerance_minutes,
)
if match is not None:
return match
else:
await asyncio.sleep(poll_interval_seconds)
continue
finally:
parser.suppress_recv_logging = False
attach_ws_poll_failure(
collected_messages,
total_wait_seconds,
request_name,
)
return None
def _normalize_report_period_datetime(value: datetime) -> datetime:
"""Приводит datetime периода отчёта к московскому времени без микросекунд."""
return localize_as_moscow(value).replace(microsecond=0)
def _exported_item_period_matches(
item_start: datetime,
item_end: datetime,
period_start: datetime,
period_end: datetime,
tolerance_minutes: int,
) -> bool:
"""Проверяет start/end элемента списка в пределах периода запроса +- tolerance_minutes."""
item_start_norm = _normalize_report_period_datetime(item_start)
item_end_norm = _normalize_report_period_datetime(item_end)
period_start_norm = _normalize_report_period_datetime(period_start)
period_end_norm = _normalize_report_period_datetime(period_end)
delta = timedelta(minutes=tolerance_minutes)
return (period_start_norm - delta) <= item_start_norm <= (period_start_norm + delta) and (
period_end_norm - delta
) <= item_end_norm <= (period_end_norm + delta)
def _normalize_report_text_for_match(text: str) -> str:
return text.lower().replace("ё", "е")
def find_matching_exported_item(
items: List[Any],
expected_data_type: Any,
name_substring: str,
tu_name_substring: str,
period_start: datetime,
period_end: datetime,
period_tolerance_minutes: int = ReportConst.REPORT_PERIOD_TOLERANCE_MINUTES,
) -> Optional[Any]:
"""
Ищет элемент списка по типу, подстрокам в имени (отчёт + ТУ) и периоду start/end с допуском.
"""
name_substring_normalized = _normalize_report_text_for_match(name_substring)
tu_name_normalized = _normalize_report_text_for_match(tu_name_substring)
matched_items = []
for item in items:
if item.exportedDataType != expected_data_type:
continue
item_name_normalized = _normalize_report_text_for_match(item.name or "")
if name_substring_normalized not in item_name_normalized:
continue
if tu_name_normalized not in item_name_normalized:
continue
if item.start is None or item.end is None:
continue
if not _exported_item_period_matches(item.start, item.end, period_start, period_end, period_tolerance_minutes):
continue
matched_items.append(item)
if not matched_items:
return None
return max(matched_items, key=lambda exported_item: exported_item.id)
def parse_journal_msg_value(value: str) -> tuple:
"""Парсит поле value в сообщении журнала"""
try:
# ищет группы цифр с точкой в строке
matches = re.findall(TestConst.DIGITS_WITH_DOT_PATTERN, value)
coordinate, volume = (matches + [None, None])[:2]
if coordinate is not None:
try:
coordinate = float(coordinate)
except ValueError:
coordinate = None
if volume is not None:
try:
volume = float(volume)
except ValueError:
volume = None
return coordinate, volume
except (AttributeError, TypeError, ValueError):
return None, None
def parse_bit_flags(
value: int, enum_cls: Type[IntEnum | IntFlag], failures: Optional[List[str]] = None
) -> List[IntFlag]:
"""
Распаковка битовых флагов
"""
# 0 - это валидное состояние когда причин нет, в прошлой реализации тест бы падал если причин нет
# хотя это может быть ожидаемо, например при тестировании исправности или где-нибудь еще
if value == 0:
return []
found_flags = [flag for flag in enum_cls if value & flag.value]
known_bits = sum(flag.value for flag in found_flags)
if known_bits != value:
unknown_bits = value ^ known_bits
error_message = f"Неизвестные биты при распаковке {enum_cls.__name__}: {unknown_bits}"
if failures is not None:
failures.append(error_message)
else:
fail(f"Неизвестные биты при распаковке {enum_cls.__name__}: {unknown_bits}")
# та же сортировка только не цифры, а их текстовое значение
return sorted(found_flags, key=lambda flag: flag.value)
def get_json_from_http_response(response: Response) -> dict:
try:
return response.json()
except ValueError:
fail("Не удалось преобразовать тело ответа в JSON")
def get_reason_enum_by_lds_status(lds_status: int | LdsStatus, failures: Optional[List[str]] = None) -> Type[IntFlag]:
"""
Получение класса причин по режимам СОУ
"""
if isinstance(lds_status, int):
try:
lds_status = LdsStatus(lds_status)
except ValueError:
error_message = f"Неизвестный LdsStatus: {lds_status}"
if failures is not None:
failures.append(error_message)
else:
fail(error_message)
reason_by_lds_status = {
LdsStatus.FAULTY: LdsStatusFaulty,
LdsStatus.INITIALIZATION: LdsStatusInitialization,
LdsStatus.DEGRADATION: LdsStatusDegradation,
}
enum_class = reason_by_lds_status.get(lds_status)
if enum_class is None:
error_message = f"Для LdsStatus{lds_status.name} не определены причины"
if failures is not None:
failures.append(error_message)
else:
fail(error_message)
return enum_class
def get_reason_enum_by_stationary_status(
stationary_status: int | StationaryStatus, failures: Optional[List[str]] = None
) -> Type[IntFlag]:
"""
Получение класса причин по режимам МТ
"""
if isinstance(stationary_status, int):
try:
stationary_status = StationaryStatus(stationary_status)
except ValueError:
error_message = f"Неизвестный StationaryStatus: {stationary_status}"
if failures is not None:
failures.append(error_message)
else:
fail(error_message)
reason_by_stationary_status = {
StationaryStatus.STATIONARY: StationaryReason,
StationaryStatus.UNSTATIONARY: UnStationaryReason,
StationaryStatus.STOPPED: StoppedPumpingReason,
}
enum_class = reason_by_stationary_status.get(stationary_status)
if enum_class is None:
error_message = f"Для StationaryStatus{stationary_status.name} не определены причины"
if failures is not None:
failures.append(error_message)
else:
fail(error_message)
return enum_class
def parse_lds_status_reasons(lds_status: int, lds_status_reasons: int, failures: Optional[List[str]] = None):
"""
Получение списка ldsStatusReasons, соответствующего ldsStatus
"""
enum_cls = get_reason_enum_by_lds_status(lds_status, failures)
flags = enum_cls(lds_status_reasons)
if flags == str(lds_status_reasons):
error_message = f"Не удалось распаковать флаги: {lds_status_reasons} для {enum_cls.__name__}"
if failures is not None:
failures.append(error_message)
return flags
def _is_configurator_flow_active() -> bool:
from utils.helpers import lds_configurator_utils as lds_cfg
return lds_cfg.is_configurator_flow_active()
def parse_stationary_status_reasons(
stationary_status: int, stationary_status_reasons: int, failures: Optional[List[str]] = None
):
"""
Получение списка stationaryStatusReasons, соответствующего stationaryStatus
"""
enum_cls = get_reason_enum_by_stationary_status(stationary_status, failures)
flags = enum_cls(stationary_status_reasons)
if flags == str(stationary_status_reasons):
error_message = f"Не удалось распаковать флаги: {stationary_status_reasons} для {enum_cls.__name__}"
if failures is not None:
failures.append(error_message)
return flags
async def connect(ws_client: WebSocketClient, ws_invoke_type: str, ws_invoke_params: Any = None) -> None:
"""
Подключение к заданной подписке
"""
try:
if _is_configurator_flow_active():
await ws_client.invoke(ws_invoke_type, ws_invoke_params)
else:
with allure.step(f"Вызов {ws_invoke_type} c параметрами {ws_invoke_params}"):
await ws_client.invoke(ws_invoke_type, ws_invoke_params)
except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
if _is_configurator_flow_active():
raise
fail(f"Не удалось отправить сообщение типа: {ws_invoke_type} c параметрами {ws_invoke_params}. Ошибка: {error}")
async def connect_stream(
ws_client: WebSocketClient,
ws_invoke_type: str,
ws_invoke_params: Any = None,
purpose: str = "streaming-вызов WS",
) -> None:
"""
Streaming-вызов (StreamInvocation)
"""
try:
with allure.step(f"Streaming-вызов {ws_invoke_type} c параметрами {ws_invoke_params}"):
await ws_client.invoke_stream(ws_invoke_type, ws_invoke_params)
except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
fail(
f"Не удалось выполнить {purpose} ({ws_invoke_type}, StreamInvocation). "
f"Параметры запроса: {ws_invoke_params}. Ошибка соединения: {error}"
)
def _stream_completion_error(msg: Any, invocation_id: str) -> Optional[str]:
"""Текст ошибки из ответа SignalR Completion для данного invocation_id, если есть."""
if not isinstance(msg, list):
return None
if msg[0] != WS_Const.COMPLETION_MESSAGE_TYPE:
return None
if not is_desired_invocation_id(msg, invocation_id):
return None
if len(msg) <= WS_Const.COMPLETION_ERROR_MESSAGE_INDEX:
return None
error_text = msg[WS_Const.COMPLETION_ERROR_MESSAGE_INDEX]
if isinstance(error_text, str):
return error_text
return None
async def receive_download_exported_data_reply(
ws_client: WebSocketClient,
parser,
invocation_id: str,
request_name: str,
total_wait_seconds: float,
poll_interval_seconds: float = 0.5,
purpose: str = "скачивании xlsx-отчёта после выбора файла в списке сформированных отчётов",
) -> Any:
"""
Ожидает StreamItem с fileChunk после streaming DownloadExportedDataRequest.
"""
deadline = asyncio.get_event_loop().time() + total_wait_seconds
collected_messages: List[Any] = []
ws_client.suppress_recv_logging = True
parser.suppress_recv_logging = True
try:
while asyncio.get_event_loop().time() < deadline:
await asyncio.sleep(poll_interval_seconds)
batch = _drain_recv_queue(ws_client)
collected_messages.extend(batch)
for msg in batch:
stream_error = _stream_completion_error(msg, invocation_id)
if stream_error:
_attach_ws_reply_parse_failure(msg, request_name, RuntimeError(stream_error), invocation_id)
fail(
f"При {purpose} бэк вернул Completion с ошибкой "
f"({request_name}, invocation_id={invocation_id}): {stream_error}"
)
for msg in batch:
if (
not isinstance(msg, list)
or msg[0] != WS_Const.STREAM_ITEM_MESSAGE_TYPE
or not is_desired_invocation_id(msg, invocation_id)
):
continue
if parser.find_reply_status_in_ws_msg(msg) is None:
continue
try:
return parser.parse_download_exported_data_msg(msg)
except Exception as error:
_attach_ws_reply_parse_failure(msg, request_name, error, invocation_id)
fail(
f"При {purpose} получен StreamItem ({request_name}, invocation_id={invocation_id}), "
f"но не удалось разобрать ответ с fileChunk: {error}"
)
finally:
collected_messages.extend(_drain_recv_queue(ws_client))
ws_client.suppress_recv_logging = False
parser.suppress_recv_logging = False
attach_ws_poll_failure(collected_messages, total_wait_seconds, f"{request_name} (StreamItem)")
fail(
f"При {purpose} за {total_wait_seconds} с не получен StreamItem с fileChunk "
f"({request_name}, invocation_id={invocation_id}). Смотреть вложения received ws message"
)
async def connect_and_get_parsed_msg_by_tu_id(
tu_id: int,
ws_client: WebSocketClient,
ws_message_type: str,
ws_invoke_type: str,
ws_invoke_params: Any = None,
timeout: float = TestConst.BASIC_MESSAGE_TIMEOUT,
) -> SubscribeAllLeaksInfoReply:
"""
Подключается, ищет и парсит allLeaksInfo сообщение для конкретного ТУ
"""
await connect(ws_client, ws_invoke_type, ws_invoke_params)
async def get_parsed_msg():
"""
Ищет и парсит allLeaksInfo сообщение для конкретного ТУ
"""
while True:
payload = await ws_client.receive_by_type(ws_message_type, timeout=timeout)
parsed_payload = ws_message_parser.parse_all_leaks_info_msg(payload)
# Ищет сообщение с нужным ТУ
if parsed_payload.replyContent.tuId == tu_id:
return parsed_payload
try:
with allure.step(f"Получение сообщения с контентом типа: {ws_message_type} для ТУ {tu_id}"):
return await asyncio.wait_for(get_parsed_msg(), timeout=timeout)
except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
fail(f"Не удалось получить сообщение allLeaksInfo для ТУ {tu_id}. Ошибка: {error}")
async def connect_and_get_msg(
ws_client: WebSocketClient,
ws_invoke_type: str,
ws_invoke_params: Any = None,
receive_timeout: Optional[float] = None,
) -> list:
"""
Подключение типа get к заданной подписке и получение сообщения с заданным типом контента
"""
await connect(ws_client, ws_invoke_type, ws_invoke_params)
invocation_id = ws_client.invocation_id
timeout = receive_timeout if receive_timeout is not None else WS_Const.FILTERING_TIMEOUT
try:
if _is_configurator_flow_active():
payload = await ws_client.receive_by_invocation_id(invocation_id, timeout=timeout)
else:
with allure.step(f"Получение входящего сообщения c invocation_id: {invocation_id}"):
payload = await ws_client.receive_by_invocation_id(invocation_id, timeout=timeout)
return payload
except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
if _is_configurator_flow_active():
raise
fail(f"Не удалось получить сообщение типа: {ws_invoke_type}. Ошибка: {error}")
async def connect_and_poll_subscribed_msg(
ws_client: WebSocketClient,
ws_message_type: str,
ws_invoke_type: str,
ws_invoke_params: Any = None,
retries: int = TestConst.SUBSCRIBE_MESSAGE_POLL_ATTEMPTS,
) -> list:
"""
Подписывается и ищет нужное сообщение в потоке, с повторным переподключением ws клиента
"""
ws_client.suppress_recv_logging = True
try:
for attempt in range(1, retries + 1):
try:
with allure.step(
f"Получение сообщения с контентом типа: {ws_message_type} - попытка {attempt} из {retries}"
):
await connect(ws_client, ws_invoke_type, ws_invoke_params)
payload = await ws_client.receive_by_type(
ws_message_type, timeout=TestConst.POLL_BY_TIMEOUT_SECONDS
)
return payload
except asyncio.TimeoutError:
await ws_client.reconnect()
continue
except Exception:
await ws_client.reconnect()
continue
fail(f"Не удалось получить сообщение с контентом типа: {ws_message_type} за {retries} попыток")
except (asyncio.TimeoutError, OSError, ConnectionError, ConnectionResetError) as error:
fail(f"Не удалось получить сообщение типа: {ws_message_type}. Ошибка: {error}")
finally:
ws_client.suppress_recv_logging = False
async def connect_and_poll_subscribed_signal(
ws_client: WebSocketClient,
ws_message_type: str,
ws_invoke_type: str,
ws_invoke_params: Any,
sensor_id: int,
sensor_description: str,
parse_payload: Callable[[list], ParsedPayloadType],
extract_signals: Callable[[ParsedPayloadType], List[InputSignalType]],
*,
max_messages: int = TestConst.SUBSCRIBE_MESSAGE_POLL_ATTEMPTS,
timeout: float = TestConst.BASIC_MESSAGE_TIMEOUT,
) -> tuple[ParsedPayloadType, InputSignalType]:
"""
Подписывается один раз и читает до max_messages сообщений из потока,
пока не найдёт сигнал с нужным id.
"""
await connect(ws_client, ws_invoke_type, ws_invoke_params)
received_ids: Set[int] = set()
try:
for attempt in range(1, max_messages + 1):
with allure.step(
f"Получение сообщения с контентом типа: {ws_message_type} - попытка {attempt} из {max_messages}"
):
payload = await ws_client.receive_by_type(ws_message_type, timeout=timeout)
parsed = parse_payload(payload)
signals = extract_signals(parsed) or []
for signal in signals:
signal_id = getattr(signal, "id", None)
if signal_id is not None:
received_ids.add(signal_id)
target_signal = find_object_by_field(signals, "id", sensor_id)
if target_signal is not None:
allure.attach(
f"Сигнал id={sensor_id} ({sensor_description}) найден на попытке {attempt} из {max_messages}\n"
f"Всего сигналов в сообщении: {len(signals)}\n"
f"Полученные id: {sorted(received_ids)}",
name=f"Результат поиска сигнала в {ws_message_type}",
attachment_type=allure.attachment_type.TEXT,
)
return parsed, target_signal
received_ids_text = ", ".join(str(signal_id) for signal_id in sorted(received_ids)) or "нет"
fail(
f"Информация по датчику {sensor_description} (id={sensor_id}) не пришла в ответе {ws_message_type} "
f"после {max_messages} попыток чтения из потока подписки. Полученные id: {received_ids_text}"
)
except (asyncio.TimeoutError, OSError, ConnectionError, ConnectionResetError) as error:
fail(
f"Не удалось получить сообщение типа: {ws_message_type} для датчика {sensor_description} "
f"(id={sensor_id}). Ошибка: {error}"
)
async def connect_and_subscribe_msg(
ws_client: WebSocketClient,
ws_message_type: str,
ws_invoke_type: str,
ws_invoke_params: Any = None,
timeout: float = TestConst.BASIC_MESSAGE_TIMEOUT,
) -> list:
"""
Подключение типа subscribe к заданной подписке и получение сообщения с заданным типом контента
"""
await connect(ws_client, ws_invoke_type, ws_invoke_params)
try:
with allure.step(f"Получение сообщения с контентом типа: {ws_message_type}"):
payload = await ws_client.receive_by_type(ws_message_type, timeout=timeout)
return payload
except (asyncio.TimeoutError, OSError, ConnectionError, ConnectionResetError) as error:
fail(f"Не удалось получить сообщение типа: {ws_invoke_type}. Ошибка: {error}")
async def poll_balance_algorithm_diagnostic_areas(
ws_client: WebSocketClient,
ws_parser: ws_message_parser,
imitator_start_time: datetime,
end_time: datetime,
poll_interval: float,
) -> list:
"""
Опрашивает очередь ws_client на наличие BalanceAlgorithmResultsContent,
собирает и возвращает все diagnosticAreas из flowAreas.
"""
collected_diagnostic_areas = []
ws_client.suppress_recv_logging = True
ws_parser.suppress_recv_logging = True
try:
while datetime.now(tz=imitator_start_time.tzinfo) < end_time:
await asyncio.sleep(poll_interval)
latest_msg = None
while not ws_client.recv_queue.empty():
try:
msg = ws_client.recv_queue.get_nowait()
except asyncio.QueueEmpty:
break
if isinstance(msg, list) and is_desired_type(msg, "BalanceAlgorithmResultsContent"):
latest_msg = msg
if latest_msg is None:
continue
parsed_payload = ws_message_parser.parse_balance_algorithm_msg(latest_msg)
reply_content = parsed_payload.replyContent
if reply_content and reply_content.flowAreas:
for flow_area in reply_content.flowAreas:
if flow_area.diagnosticAreas:
collected_diagnostic_areas.extend(flow_area.diagnosticAreas)
finally:
ws_client.suppress_recv_logging = False
ws_parser.suppress_recv_logging = False
return collected_diagnostic_areas
def get_leak_diagnostic_area_samples(
collected_diagnostic_areas: list,
leak_diagnostic_area_name: str,
total_wait: int,
) -> list:
"""
Проверяет наличие diagnosticAreas и возвращает подмножество
для ДУ с заданным leak_diagnostic_area_id. Падает, если данные не найдены.
"""
if not collected_diagnostic_areas:
fail(f"За {total_wait} секунд не пришло ни одной diagnosticArea в BalanceAlgorithmResultsContent")
leak_diagnostic_area_samples = [
diagnostic_area
for diagnostic_area in collected_diagnostic_areas
if diagnostic_area.name == leak_diagnostic_area_name
]
if not leak_diagnostic_area_samples:
fail(f"За {total_wait} секунд не пришло ни одного сообщения для ДУ с name={leak_diagnostic_area_name}.")
return leak_diagnostic_area_samples
def moscow_now() -> datetime:
"""Текущее время в часовом поясе Europe/Moscow."""
return datetime.now(ZoneInfo(TestConst.ZONE_INFO))
conftest.py
from __future__ import annotations
import asyncio
import pprint
import random
import re
from collections import Counter
from dataclasses import asdict, is_dataclass
from datetime import datetime, timedelta, timezone
from enum import IntEnum, IntFlag
from typing import Any, Callable, List, Optional, Protocol, Set, Type, TypeVar
from zoneinfo import ZoneInfo
import allure
from msgpack import Timestamp as MsgpackTimestamp
from pytest import fail
from requests import Response
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.architecture_constants import WebSocketClientConstants as WS_Const
from constants.enums import (
ConfirmationStatus,
ExportStatus,
LdsStatus,
LdsStatusDegradation,
LdsStatusFaulty,
LdsStatusInitialization,
LeakStatus,
ReplyStatus,
SignalType,
StationaryReason,
StationaryStatus,
StoppedPumpingReason,
UnStationaryReason,
)
from constants.test_constants import BaseTN3Constants as TestConst
from constants.test_constants import ExportReportConstants as ReportConst
from models.export_reports_model import ReportDataExportedContent, ReportDataExportedNotification
from models.get_messages_model import GetMessagesRequest
from models.subscribe_all_leaks_info_model import SubscribeAllLeaksInfoReply
from models.subscribe_common_scheme_model import DiagnosticArea, FlowArea
from models.subscribe_leaks_model import Leak
from models.subscribe_main_page_info_model import MainPageLeakInfo
from utils.helpers.ws_message_parser import ws_message_parser
from utils.msgpack_utils.message_filters import is_desired_invocation_id, is_desired_type
ObjectType = TypeVar("ObjectType") # создает типовую переменную для поиска объектов в списке
RandomObjectType = TypeVar("RandomObjectType")
Event = TypeVar("Event")
ParsedPayloadType = TypeVar("ParsedPayloadType")
InputSignalType = TypeVar("InputSignalType")
SiteSignalsMap = dict[tuple[int, int], dict[int, str]]
class ControlledSiteSignalMessage(Protocol):
controlledSiteId: int
segmentId: int
signals: list
def convert_leak_volume_m3(volume: float) -> float:
"""
Преобразует объем утечки в м3/час
"""
# Округляет результат для читабельности
return round(volume * TestConst.MASS_KG, 3)
def datetime_minus_seconds(datetime_obj: datetime, delta_s: int) -> datetime:
"""
Вычитает время в секундах из datetime
"""
return (datetime_obj - timedelta(seconds=delta_s)).replace(microsecond=0)
def calculate_leak_start_time(imitator_start_time: datetime, leak_interval_seconds: int) -> Optional[datetime]:
"""
Рассчитывает время начала утечки на основе времени старта имитатора.
:param imitator_start_time: datetime объект времени старта имитатора
:param leak_interval_seconds: интервал от старта до утечки в секундах (LEAK_START_INTERVAL)
:return: datetime время ожидаемого начала утечки
"""
if not imitator_start_time:
return None
return (imitator_start_time + timedelta(seconds=leak_interval_seconds)).replace(microsecond=0)
def calculate_leak_end_time(
imitator_start_time: datetime, leak_interval_seconds: int, allowed_diff_seconds: int
) -> Optional[datetime]:
"""
Рассчитывает крайнее время обнаружения утечки (с учётом допустимой погрешности).
:param imitator_start_time: datetime объект времени старта имитатора
:param leak_interval_seconds: интервал от старта до утечки в секундах (LEAK_START_INTERVAL)
:param allowed_diff_seconds: допустимая погрешность времени обнаружения (ALLOWED_TIME_DIFF_SECONDS)
:return: datetime крайнее время обнаружения утечки
"""
if not imitator_start_time:
return None
total_seconds = leak_interval_seconds + allowed_diff_seconds
return (imitator_start_time + timedelta(seconds=total_seconds)).replace(microsecond=0)
def get_leak_time_window(
imitator_start_time: datetime, leak_interval_seconds: int, allowed_diff_seconds: int, detected_at_tz=None
) -> tuple[datetime, datetime]:
"""
Возвращает временное окно для проверки времени обнаружения утечки.
:param imitator_start_time: datetime объект времени старта имитатора
:param leak_interval_seconds: интервал от старта до утечки в секундах
:param allowed_diff_seconds: допустимая погрешность времени обнаружения
:param detected_at_tz: timezone из времени обнаружения утечки (опционально)
:return: tuple (leak_start_time, leak_end_time) для использования в is_between проверке
"""
leak_start = calculate_leak_start_time(imitator_start_time, leak_interval_seconds)
leak_end = calculate_leak_end_time(imitator_start_time, leak_interval_seconds, allowed_diff_seconds)
# Если передан timezone, применяем его к временам для корректного сравнения
if detected_at_tz:
leak_start = leak_start.replace(tzinfo=detected_at_tz)
leak_end = leak_end.replace(tzinfo=detected_at_tz)
return leak_start, leak_end
def ensure_moscow_timezone(input_datetime: datetime) -> None | datetime:
"""
Конвертирует datetime в московское время, если оно не в московской таймзоне.
:param input_datetime: datetime объект
:return: datetime в московской таймзоне
"""
if input_datetime is None:
return input_datetime
# Если datetime без timezone - считаем что это UTC
if input_datetime.tzinfo is None:
input_datetime = input_datetime.replace(tzinfo=timezone.utc)
# Конвертируем в московское время
return input_datetime.astimezone(ZoneInfo(TestConst.ZONE_INFO))
def report_time_offset_hours(tz_name: str = TestConst.ZONE_INFO) -> Optional[int]:
"""
Смещение часового пояса (часы от UTC) для поля timeOffset в запросах отчётов.
"""
now = datetime.now(ZoneInfo(tz_name))
utc_offset = now.utcoffset()
if utc_offset is None:
return None
return int(utc_offset.total_seconds() // TestConst.SECONDS_PER_HOUR)
def localize_as_moscow(input_datetime: datetime) -> None | datetime:
"""
Присваивает datetime московский часовой пояс без сдвига времени.
Если datetime уже имеет timezone - конвертирует в московское время.
"""
if input_datetime is None:
return input_datetime
moscow_tz = ZoneInfo(TestConst.ZONE_INFO)
if input_datetime.tzinfo is None:
return input_datetime.replace(tzinfo=moscow_tz)
return input_datetime.astimezone(moscow_tz)
def format_datetime_moscow(value: Optional[datetime]) -> str:
"""Строковое представление datetime в Europe/Moscow для вложений Allure."""
if value is None:
return "None"
return str(localize_as_moscow(value))
def get_rejection_time_window(
imitator_start_time: datetime,
start_seconds: int | float,
reserve_seconds: int | float = 0,
) -> tuple[datetime, datetime]:
"""
Возвращает временное окно для проверки сообщения об отбраковке.
"""
imitator_msk = localize_as_moscow(imitator_start_time)
range_start = imitator_msk + timedelta(seconds=start_seconds - reserve_seconds)
range_end = localize_as_moscow(datetime.now())
return range_start, range_end
def find_rejection_journal_message(
messages_info: List[ObjectType],
tag: str,
range_start: datetime,
range_end: datetime,
technological_section: str,
expected_event: str,
) -> tuple[list[ObjectType], ObjectType | None]:
"""
Фильтрует сообщения журнала по tag и временному диапазону,
затем ищет целевое сообщение по technologicalSection и event.
"""
time_filtered = [
msg for msg in messages_info if msg.tag == tag and range_start <= ensure_moscow_timezone(msg.time) <= range_end
]
time_filtered.sort(key=lambda msg: ensure_moscow_timezone(msg.time), reverse=True)
target_msg = next(
(
msg
for msg in time_filtered
if msg.technologicalSection == technological_section and msg.event.rstrip() == expected_event
),
None,
)
return time_filtered, target_msg
def get_random_item(item_list: List[RandomObjectType]) -> RandomObjectType:
"""
Получает случайный объект из списка
"""
if not item_list:
return None
try:
return random.choice(item_list)
except (TypeError, ValueError):
return None
def get_longest_flow_area(flow_areas: List[FlowArea]) -> Optional[FlowArea]:
"""
Получает самый протяженный участок карты течения по количеству ДУ из списка всех участков
"""
if not flow_areas:
return None
try:
longest_flow_area = max(flow_areas, key=lambda flow_area: len(flow_area.diagnosticAreas))
return longest_flow_area
except (TypeError, ValueError):
return None
def get_longest_flow_area_by_pipes(flow_areas: List[FlowArea]) -> Optional[FlowArea]:
"""
Получает самый протяженный участок карты течения по количеству pipesIds
"""
if not flow_areas:
return None
try:
return max(flow_areas, key=lambda flow_area: len(flow_area.pipesIds or []))
except (TypeError, ValueError, AttributeError):
return None
def filter_flow_areas_with_available_flow(flow_areas: List[FlowArea]) -> List[FlowArea]:
"""
Оставляет только участки карты течения с доступным течением
"""
return [flow_area for flow_area in flow_areas if getattr(flow_area, "isFlowAvailable", False)]
def get_diagnostic_area_base_ids() -> list[int]:
"""
Возвращает список id базовых диагностических участков из константы DIAGNOSTIC_AREA_BASE_IDS
"""
return [pair[0] for pair in TestConst.DIAGNOSTIC_AREA_BASE_IDS.values()]
def diagnostic_area_title(du_id: int) -> str:
"""
Возвращает строковое представление ДУ: id и название из DIAGNOSTIC_AREA_BASE_IDS
"""
for name, pair in TestConst.DIAGNOSTIC_AREA_BASE_IDS.items():
if pair[0] == du_id:
return f"{du_id} ({name})"
return str(du_id)
def determine_stationary_status_by_majority(statuses: list[StationaryStatus]) -> Optional[StationaryStatus]:
"""
Определяет общий режим МТ по правилу большинства ДУ в одном статусе.
При равенстве количества выбирается статус с максимальным значением enum.
"""
if not statuses:
return None
counts = Counter(statuses)
max_count = max(counts.values())
candidates = [status for status, count in counts.items() if count == max_count]
return max(candidates, key=lambda status: status.value)
def collect_stationary_statuses_from_output_signals(
controlled_site_signals: list[ControlledSiteSignalMessage], controlled_sites_dict: dict
) -> tuple[list[tuple[str, StationaryStatus]], list[str]]:
"""
Собирает stationaryStatus (SignalType.REGLU) по всем участкам КП-КП из OutputSignalsInfo
"""
all_signals_dict = extract_signal_on_site_and_segment(controlled_site_signals)
site_stationary_statuses: list[tuple[str, StationaryStatus]] = []
missed_controlled_sites: list[str] = []
for name, value in controlled_sites_dict.items():
stationary_status = get_signal_value(all_signals_dict, value, SignalType.REGLU)
if stationary_status:
if isinstance(stationary_status, StationaryStatus):
site_stationary_statuses.append((name, stationary_status))
else:
missed_controlled_sites.append(name)
return site_stationary_statuses, missed_controlled_sites
def determine_lds_status_by_priority(lds_status_set: Set[int]) -> Optional[int]:
"""
Определяет режим работы СОУ по приоритету и наличию режимов работы у ДУ на самом протяженном участки карты течений
"""
lds_status_priority = [
LdsStatus.FAULTY.value,
LdsStatus.INITIALIZATION.value,
LdsStatus.DEGRADATION.value,
LdsStatus.SERVICEABLE.value,
]
if not lds_status_set:
return None
try:
for status in lds_status_priority:
if status in lds_status_set:
return status
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_signal_type_by_address_suffix(signals_list: list, address_suffix: str) -> Optional[int]:
"""
Ищет в списке сигналов тип сигнала по части адреса
"""
if not signals_list:
return None
try:
for sensor_signal in signals_list:
if sensor_signal.address is not None and str(sensor_signal.address).endswith(address_suffix):
return sensor_signal.signalType
return None
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_signal_val_by_signal_type(signals_list: list, signal_type: int) -> Optional[str]:
"""
Ищет в списке сигналов значение сигнала по типу
"""
if not signals_list:
return None
try:
for sensor_signal in signals_list:
if sensor_signal.signalType is not None and sensor_signal.signalType == signal_type:
return sensor_signal.value
return None
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_object_by_field(item_list: List[ObjectType], field_name: str, value: Any) -> ObjectType:
"""
Ищет объект в списке объектов по значению одного из полей объекта
"""
if not item_list:
return None
try:
return next((item for item in item_list if getattr(item, field_name) == value))
except Exception:
return None
def find_object_by_few_fields(item_list: List[ObjectType], fields_dict: dict) -> ObjectType:
"""
Ищет объект в списке объектов по значениям нескольких полей
"""
if not item_list:
return None
return next(
(item for item in item_list if all(getattr(item, field) == value for field, value in fields_dict.items())), None
)
def extract_signal_on_site_and_segment(
controlled_site_messages: list[ControlledSiteSignalMessage],
) -> SiteSignalsMap:
"""
Ищет сигналы для всех пар КП-КП
"""
result: SiteSignalsMap = {}
for site_obj in controlled_site_messages:
key = (site_obj.controlledSiteId, site_obj.segmentId)
result[key] = {signal.signalType: signal.value for signal in site_obj.signals}
return result
def get_signal_value(
all_signals_dict: SiteSignalsMap,
site_kp_kp: tuple[int, int],
signals_id: int,
) -> StationaryStatus | LdsStatus | int | None:
"""
Получает значение сигналов.
Приводит значение к типу инт.
Кладет в обертку значения режима МТ и режима СОУ
"""
signals_value = all_signals_dict.get(site_kp_kp, {}).get(signals_id)
try:
int_signals_value = int(signals_value)
except (TypeError, ValueError):
return None
if signals_id == SignalType.REGLU:
return StationaryStatus(int_signals_value) if signals_value else None
if signals_id == SignalType.REGSOU:
return LdsStatus(int_signals_value) if signals_value else None
return int_signals_value
def parse_journal_event(event_value: str) -> tuple[str | None, str | None]:
"""
Разделяет строку события на имя и причину, вложенную в скобки
"""
if not event_value or not isinstance(event_value, str):
return None, None
# Паттерн для "Состояние режима (Причина)"
pattern = r'^([^\(]+?)\s*\(([^\)]+)\)\s*$'
match = re.match(pattern, event_value)
if match:
mode_part = match.group(1).strip()
reason_part = match.group(2).strip()
return mode_part if mode_part else None, reason_part if reason_part else None
# Если значение события не соответствует паттерну, возвращаю текст, если нет текста, тогда None
return event_value.strip(), None
def get_signal(site_message, signal_type):
if not site_message:
return None
return next((s for s in site_message.signals if s.signalType == signal_type.value), None)
def get_value(obj):
return getattr(obj, "value", None)
def find_confirmed_leaks(item_list: List[Leak]) -> List[Leak]:
"""Ищет подтвержденные утечки"""
try:
return [
item
for item in item_list
if item.confirmationStatus == ConfirmationStatus.CONFIRMED.value and item.detectedAt is not None
]
except (AttributeError, KeyError, TypeError, ValueError):
return []
def find_confirmed_leaks_on_main_page(item_list: List[MainPageLeakInfo]) -> List[MainPageLeakInfo]:
"""Ищет подтвержденные утечки"""
try:
return [
item
for item in item_list
if item.leakStatus == LeakStatus.CONFIRMED.value and item.leakDetectedAt is not None
]
except (AttributeError, KeyError, TypeError, ValueError):
return []
def find_diagnostic_area_by_id(flow_areas: List[FlowArea], id_value: Optional[int]) -> Optional[DiagnosticArea]:
"""
Ищет ДУ по id в списке участков карты течений, исключает дубликаты по количеству pipeIds
"""
candidates = []
if not flow_areas or not id_value:
return None
try:
for flow_area in flow_areas:
for diagnostic_area in flow_area.diagnosticAreas:
if diagnostic_area.id == id_value:
candidates.append(diagnostic_area)
if not candidates:
return None
elif len(candidates) == 1:
return candidates[0]
else:
# Среди дубликатов ищет ДУ с наибольшим количеством pipeIds
return max(candidates, key=lambda candidate: len(candidate.pipeIds))
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_diagnostic_area_by_pipe_id(flow_areas: List[FlowArea], pipe_id: int) -> Optional[DiagnosticArea]:
"""
Ищет ДУ по pipe id в списке участков карты течений, исключает дубликаты по количеству pipeIds
"""
candidates = []
if not flow_areas:
return None
try:
for flow_area in flow_areas:
for diagnostic_area in flow_area.diagnosticAreas:
if diagnostic_area.pipeIds and pipe_id in diagnostic_area.pipeIds:
candidates.append(diagnostic_area)
if not candidates:
return None
elif len(candidates) == 1:
return candidates[0]
else:
# Среди дубликатов ищет ДУ с наибольшим количеством pipeIds
return max(candidates, key=lambda candidate: len(candidate.pipeIds))
except (AttributeError, KeyError, RuntimeError, TypeError, ValueError):
return None
def find_diagnostic_areas_by_ids(flow_areas: List[FlowArea], id_list: List[int]) -> List[DiagnosticArea]:
"""
Получает список ДУ из списка flow_areas по списку id
"""
diagnostic_areas = [
result
for diagnostic_area_id in id_list
if (result := find_diagnostic_area_by_id(flow_areas, diagnostic_area_id)) is not None
]
return diagnostic_areas
def find_diagnostic_areas_by_pipe_ids(flow_areas: List[FlowArea], id_list: List[int]) -> List[DiagnosticArea]:
"""
Получает список ДУ из списка flow_areas по списку pipe id
"""
diagnostic_areas = [
result for pipe_id in id_list if (result := find_diagnostic_area_by_pipe_id(flow_areas, pipe_id)) is not None
]
return diagnostic_areas
def find_base_diagnostic_areas(flow_areas: List[FlowArea]) -> List[DiagnosticArea]:
"""
Получает список базовых ДУ из списка flow_areas
"""
return find_diagnostic_areas_by_ids(flow_areas, get_diagnostic_area_base_ids())
def find_leak_by_coordinate(
leaks_list: List[ObjectType], expected_coordinate: float, tolerance: float = TestConst.ALLOWED_DISTANCE_DIFF_METERS
) -> ObjectType:
"""
Ищет утечку в списке по координатам с допустимой погрешностью
"""
if not leaks_list:
return None
for leak in leaks_list:
leak_coordinate = getattr(leak, "leakCoordinate")
if leak_coordinate is None or "":
continue
if abs(leak_coordinate - expected_coordinate) <= tolerance:
return leak
return None
def to_moscow_timezone(date_str: str) -> Optional[datetime]:
"""
Преобразует строку времени в московское время
"""
if not date_str or not date_str.strip():
return None
try:
if date_str.startswith(("'", '"', '')) or date_str.endswith(("'", '"', '')):
date_str = date_str.strip().strip("'").strip('"')
date_utc = datetime.strptime(date_str, TestConst.OUTPUT_TIME_FORMAT).replace(tzinfo=timezone.utc)
return date_utc.astimezone(ZoneInfo(TestConst.ZONE_INFO))
except (AttributeError, TypeError, ValueError):
return None
def create_dict_from_dataclass(cls: Type, **kwargs) -> Optional[dict]:
"""Создает словарь из экземпляра dataclass c нужными параметрами"""
if not is_dataclass(cls):
return None
instance = cls(**kwargs)
return asdict(instance)
def datetime_to_msgpack_timestamp(dt: datetime) -> list:
"""Конвертирует datetime в формат [Timestamp(seconds, nanoseconds), tz_offset] для отправки на бэкенд."""
return [MsgpackTimestamp(seconds=int(dt.timestamp()), nanoseconds=0), 0]
def datetime_to_iso_format(dt: datetime) -> str:
"""Конвертирует datetime в строку 2026-07-27T21:00:00.000Z для отправки по http на бэкенд."""
utc_datetime = dt.astimezone(timezone.utc)
return utc_datetime.isoformat(timespec='milliseconds').replace('+00:00', 'Z')
def create_journal_req_body(**kwargs) -> dict:
"""Создает дефолтные параметры запроса к журналу"""
result = create_dict_from_dataclass(GetMessagesRequest, **kwargs)
period = result.get('periodTime')
if period:
for key in ('start', 'end'):
if isinstance(period.get(key), datetime):
period[key] = datetime_to_iso_format(period[key])
return result
def extract_first_number(value: object) -> Optional[float]:
"""
Извлекает первое число из ячейки (int/float/str вида '55.55 км', '111.11 м3/ч').
"""
if value is None:
return None
if isinstance(value, (int, float)) and not isinstance(value, bool):
return float(value)
if isinstance(value, str):
matches = re.findall(TestConst.DIGITS_WITH_DOT_PATTERN, value)
if matches:
try:
return float(matches[0].replace(",", "."))
except ValueError:
return None
return None
def attach_ws_poll_failure(
collected_messages: List[Any],
total_wait_seconds: float,
expected_message_type: str,
) -> None:
"""Краткая сводка и pprint каждого WS-сообщения при таймауте поллинга."""
allure.attach(
"\n".join(
[
f"Таймаут ожидания: {total_wait_seconds} с",
f"Ожидаемый тип сообщения: {expected_message_type}",
f"Всего сообщений за период поллинга: {len(collected_messages)}",
]
),
name="WS poll timeout",
attachment_type=allure.attachment_type.TEXT,
)
for msg in collected_messages:
allure.attach(
pprint.pformat(msg, width=120, sort_dicts=False),
name="received ws message",
attachment_type=allure.attachment_type.TEXT,
)
def _attach_ws_reply_parse_failure(
reply_payload: Optional[Any],
request_name: str,
error: BaseException,
invocation_id: str = "",
) -> None:
"""Прикрепляет к Allure ответ бэка при ошибке парсинга."""
allure.attach(
"\n".join(
[
f"Запрос: {request_name}",
f"invocation_id: {invocation_id}",
f"Ошибка: {error}",
]
),
name="WS parse failure",
attachment_type=allure.attachment_type.TEXT,
)
if reply_payload is not None:
allure.attach(
pprint.pformat(reply_payload, width=120, sort_dicts=False),
name="received ws message",
attachment_type=allure.attachment_type.TEXT,
)
def _drain_recv_queue(ws_client: WebSocketClient) -> List[Any]:
"""Забирает все сообщения из очереди ws без блокирующего receive_by_..."""
messages: List[Any] = []
while not ws_client.recv_queue.empty():
try:
messages.append(ws_client.recv_queue.get_nowait())
except asyncio.QueueEmpty:
break
return messages
def _find_valid_report_export_notification(
messages: List[Any],
parser,
notification_type: str,
) -> Optional[ReportDataExportedNotification]:
"""Ищет среди уже полученных сообщений успешную ReportDataExportedNotification."""
for msg in messages:
if not isinstance(msg, list) or not is_desired_type(msg, notification_type):
continue
try:
notification = parser.parse_report_data_exported_notification_msg(msg)
except (ValueError, TypeError, KeyError):
continue
if notification.replyStatus != ReplyStatus.OK.value:
continue
content: Optional[ReportDataExportedContent] = notification.replyContent
if content is None:
continue
if content.exportStatus != ExportStatus.DONE:
continue
return notification
return None
async def poll_for_report_export_notification(
ws_client: WebSocketClient,
parser,
total_wait_seconds: float,
poll_interval_seconds: float,
) -> Optional[Any]:
"""
Собирает сообщения из очереди ws и ищет ReportDataExportedNotification с успешным exportStatus.
При таймауте прикрепляет к Allure все полученные за период сообщения.
"""
deadline = asyncio.get_event_loop().time() + total_wait_seconds
collected_messages: List[Any] = []
ws_client.suppress_recv_logging = True
parser.suppress_recv_logging = True
try:
while asyncio.get_event_loop().time() < deadline:
await asyncio.sleep(poll_interval_seconds)
batch = _drain_recv_queue(ws_client)
collected_messages.extend(batch)
notification = _find_valid_report_export_notification(
batch, parser, ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION
)
if notification is not None:
return notification
finally:
collected_messages.extend(_drain_recv_queue(ws_client))
ws_client.suppress_recv_logging = False
parser.suppress_recv_logging = False
attach_ws_poll_failure(
collected_messages,
total_wait_seconds,
ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION,
)
return None
def _find_ws_reply_by_invocation_id(messages: List[Any], invocation_id: str, parser) -> Optional[list]:
"""
Ищет последний ответ с заданным invocation_id и телом replyStatus.
"""
reply_payload = None
for msg in messages:
if not isinstance(msg, list) or not is_desired_invocation_id(msg, invocation_id):
continue
if parser.find_reply_status_in_ws_msg(msg):
reply_payload = msg
return reply_payload
async def poll_for_exported_file(
http_client: StandHttpClient,
parser,
list_limit: int,
expected_data_type: Any,
name_substring: str,
tu_name_substring: str,
period_start: datetime,
period_end: datetime,
total_wait_seconds: float,
poll_interval_seconds: float,
period_tolerance_minutes: int = ReportConst.REPORT_PERIOD_TOLERANCE_MINUTES,
) -> Optional[Any]:
"""
Периодически шлёт GetExportedDataListRequest, забирает ответы из очереди
по invocation_id среди всех накопленных сообщений.
При таймауте или ошибке парсинга прикрепляет к Allure полученные ответы.
"""
deadline = asyncio.get_event_loop().time() + total_wait_seconds
last_items_count = -1
collected_messages: List[Any] = []
request_name = HttpConst.GET_EXPORTED_DATA_LIST_URL_PATH
request_payload = {"limit": list_limit}
parser.suppress_recv_logging = True
try:
while asyncio.get_event_loop().time() < deadline:
response = http_client.post_request(request_name, request_payload)
if response.json():
collected_messages.extend(response.json())
payload = get_json_from_http_response(response)
try:
parsed_payload = parser.parse_exported_data_list_msg(payload)
except Exception as error:
_attach_ws_reply_parse_failure(payload, request_name, error)
for msg in collected_messages:
allure.attach(
pprint.pformat(msg, width=120, sort_dicts=False),
name="received http message",
attachment_type=allure.attachment_type.TEXT,
)
fail(f"Не удалось разобрать ответ на {request_name}: {error}")
items = []
reply_content = getattr(parsed_payload, "replyContent", None)
if reply_content:
items = getattr(reply_content, "exportedData", [])
if len(items) != last_items_count:
allure.attach(
"\n".join(
f"id={item.id}, name={item.name}, type={item.exportedDataType}, "
f"start={format_datetime_moscow(item.start)}, end={format_datetime_moscow(item.end)}"
for item in items
),
name=f"Список сформированных файлов (всего: {len(items)})",
attachment_type=allure.attachment_type.TEXT,
)
last_items_count = len(items)
match = find_matching_exported_item(
items=items,
expected_data_type=expected_data_type,
name_substring=name_substring,
tu_name_substring=tu_name_substring,
period_start=period_start,
period_end=period_end,
period_tolerance_minutes=period_tolerance_minutes,
)
if match is not None:
return match
else:
await asyncio.sleep(poll_interval_seconds)
continue
finally:
parser.suppress_recv_logging = False
attach_ws_poll_failure(
collected_messages,
total_wait_seconds,
request_name,
)
return None
def _normalize_report_period_datetime(value: datetime) -> datetime:
"""Приводит datetime периода отчёта к московскому времени без микросекунд."""
return localize_as_moscow(value).replace(microsecond=0)
def _exported_item_period_matches(
item_start: datetime,
item_end: datetime,
period_start: datetime,
period_end: datetime,
tolerance_minutes: int,
) -> bool:
"""Проверяет start/end элемента списка в пределах периода запроса +- tolerance_minutes."""
item_start_norm = _normalize_report_period_datetime(item_start)
item_end_norm = _normalize_report_period_datetime(item_end)
period_start_norm = _normalize_report_period_datetime(period_start)
period_end_norm = _normalize_report_period_datetime(period_end)
delta = timedelta(minutes=tolerance_minutes)
return (period_start_norm - delta) <= item_start_norm <= (period_start_norm + delta) and (
period_end_norm - delta
) <= item_end_norm <= (period_end_norm + delta)
def _normalize_report_text_for_match(text: str) -> str:
return text.lower().replace("ё", "е")
def find_matching_exported_item(
items: List[Any],
expected_data_type: Any,
name_substring: str,
tu_name_substring: str,
period_start: datetime,
period_end: datetime,
period_tolerance_minutes: int = ReportConst.REPORT_PERIOD_TOLERANCE_MINUTES,
) -> Optional[Any]:
"""
Ищет элемент списка по типу, подстрокам в имени (отчёт + ТУ) и периоду start/end с допуском.
"""
name_substring_normalized = _normalize_report_text_for_match(name_substring)
tu_name_normalized = _normalize_report_text_for_match(tu_name_substring)
matched_items = []
for item in items:
if item.exportedDataType != expected_data_type:
continue
item_name_normalized = _normalize_report_text_for_match(item.name or "")
if name_substring_normalized not in item_name_normalized:
continue
if tu_name_normalized not in item_name_normalized:
continue
if item.start is None or item.end is None:
continue
if not _exported_item_period_matches(item.start, item.end, period_start, period_end, period_tolerance_minutes):
continue
matched_items.append(item)
if not matched_items:
return None
return max(matched_items, key=lambda exported_item: exported_item.id)
def parse_journal_msg_value(value: str) -> tuple:
"""Парсит поле value в сообщении журнала"""
try:
# ищет группы цифр с точкой в строке
matches = re.findall(TestConst.DIGITS_WITH_DOT_PATTERN, value)
coordinate, volume = (matches + [None, None])[:2]
if coordinate is not None:
try:
coordinate = float(coordinate)
except ValueError:
coordinate = None
if volume is not None:
try:
volume = float(volume)
except ValueError:
volume = None
return coordinate, volume
except (AttributeError, TypeError, ValueError):
return None, None
def parse_bit_flags(
value: int, enum_cls: Type[IntEnum | IntFlag], failures: Optional[List[str]] = None
) -> List[IntFlag]:
"""
Распаковка битовых флагов
"""
# 0 - это валидное состояние когда причин нет, в прошлой реализации тест бы падал если причин нет
# хотя это может быть ожидаемо, например при тестировании исправности или где-нибудь еще
if value == 0:
return []
found_flags = [flag for flag in enum_cls if value & flag.value]
known_bits = sum(flag.value for flag in found_flags)
if known_bits != value:
unknown_bits = value ^ known_bits
error_message = f"Неизвестные биты при распаковке {enum_cls.__name__}: {unknown_bits}"
if failures is not None:
failures.append(error_message)
else:
fail(f"Неизвестные биты при распаковке {enum_cls.__name__}: {unknown_bits}")
# та же сортировка только не цифры, а их текстовое значение
return sorted(found_flags, key=lambda flag: flag.value)
def get_json_from_http_response(response: Response) -> dict:
try:
return response.json()
except ValueError:
fail("Не удалось преобразовать тело ответа в JSON")
def get_reason_enum_by_lds_status(lds_status: int | LdsStatus, failures: Optional[List[str]] = None) -> Type[IntFlag]:
"""
Получение класса причин по режимам СОУ
"""
if isinstance(lds_status, int):
try:
lds_status = LdsStatus(lds_status)
except ValueError:
error_message = f"Неизвестный LdsStatus: {lds_status}"
if failures is not None:
failures.append(error_message)
else:
fail(error_message)
reason_by_lds_status = {
LdsStatus.FAULTY: LdsStatusFaulty,
LdsStatus.INITIALIZATION: LdsStatusInitialization,
LdsStatus.DEGRADATION: LdsStatusDegradation,
}
enum_class = reason_by_lds_status.get(lds_status)
if enum_class is None:
error_message = f"Для LdsStatus{lds_status.name} не определены причины"
if failures is not None:
failures.append(error_message)
else:
fail(error_message)
return enum_class
def get_reason_enum_by_stationary_status(
stationary_status: int | StationaryStatus, failures: Optional[List[str]] = None
) -> Type[IntFlag]:
"""
Получение класса причин по режимам МТ
"""
if isinstance(stationary_status, int):
try:
stationary_status = StationaryStatus(stationary_status)
except ValueError:
error_message = f"Неизвестный StationaryStatus: {stationary_status}"
if failures is not None:
failures.append(error_message)
else:
fail(error_message)
reason_by_stationary_status = {
StationaryStatus.STATIONARY: StationaryReason,
StationaryStatus.UNSTATIONARY: UnStationaryReason,
StationaryStatus.STOPPED: StoppedPumpingReason,
}
enum_class = reason_by_stationary_status.get(stationary_status)
if enum_class is None:
error_message = f"Для StationaryStatus{stationary_status.name} не определены причины"
if failures is not None:
failures.append(error_message)
else:
fail(error_message)
return enum_class
def parse_lds_status_reasons(lds_status: int, lds_status_reasons: int, failures: Optional[List[str]] = None):
"""
Получение списка ldsStatusReasons, соответствующего ldsStatus
"""
enum_cls = get_reason_enum_by_lds_status(lds_status, failures)
flags = enum_cls(lds_status_reasons)
if flags == str(lds_status_reasons):
error_message = f"Не удалось распаковать флаги: {lds_status_reasons} для {enum_cls.__name__}"
if failures is not None:
failures.append(error_message)
return flags
def _is_configurator_flow_active() -> bool:
from utils.helpers import lds_configurator_utils as lds_cfg
return lds_cfg.is_configurator_flow_active()
def parse_stationary_status_reasons(
stationary_status: int, stationary_status_reasons: int, failures: Optional[List[str]] = None
):
"""
Получение списка stationaryStatusReasons, соответствующего stationaryStatus
"""
enum_cls = get_reason_enum_by_stationary_status(stationary_status, failures)
flags = enum_cls(stationary_status_reasons)
if flags == str(stationary_status_reasons):
error_message = f"Не удалось распаковать флаги: {stationary_status_reasons} для {enum_cls.__name__}"
if failures is not None:
failures.append(error_message)
return flags
async def connect(ws_client: WebSocketClient, ws_invoke_type: str, ws_invoke_params: Any = None) -> None:
"""
Подключение к заданной подписке
"""
try:
if _is_configurator_flow_active():
await ws_client.invoke(ws_invoke_type, ws_invoke_params)
else:
with allure.step(f"Вызов {ws_invoke_type} c параметрами {ws_invoke_params}"):
await ws_client.invoke(ws_invoke_type, ws_invoke_params)
except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
if _is_configurator_flow_active():
raise
fail(f"Не удалось отправить сообщение типа: {ws_invoke_type} c параметрами {ws_invoke_params}. Ошибка: {error}")
async def connect_stream(
ws_client: WebSocketClient,
ws_invoke_type: str,
ws_invoke_params: Any = None,
purpose: str = "streaming-вызов WS",
) -> None:
"""
Streaming-вызов (StreamInvocation)
"""
try:
with allure.step(f"Streaming-вызов {ws_invoke_type} c параметрами {ws_invoke_params}"):
await ws_client.invoke_stream(ws_invoke_type, ws_invoke_params)
except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
fail(
f"Не удалось выполнить {purpose} ({ws_invoke_type}, StreamInvocation). "
f"Параметры запроса: {ws_invoke_params}. Ошибка соединения: {error}"
)
def _stream_completion_error(msg: Any, invocation_id: str) -> Optional[str]:
"""Текст ошибки из ответа SignalR Completion для данного invocation_id, если есть."""
if not isinstance(msg, list):
return None
if msg[0] != WS_Const.COMPLETION_MESSAGE_TYPE:
return None
if not is_desired_invocation_id(msg, invocation_id):
return None
if len(msg) <= WS_Const.COMPLETION_ERROR_MESSAGE_INDEX:
return None
error_text = msg[WS_Const.COMPLETION_ERROR_MESSAGE_INDEX]
if isinstance(error_text, str):
return error_text
return None
async def receive_download_exported_data_reply(
ws_client: WebSocketClient,
parser,
invocation_id: str,
request_name: str,
total_wait_seconds: float,
poll_interval_seconds: float = 0.5,
purpose: str = "скачивании xlsx-отчёта после выбора файла в списке сформированных отчётов",
) -> Any:
"""
Ожидает StreamItem с fileChunk после streaming DownloadExportedDataRequest.
"""
deadline = asyncio.get_event_loop().time() + total_wait_seconds
collected_messages: List[Any] = []
ws_client.suppress_recv_logging = True
parser.suppress_recv_logging = True
try:
while asyncio.get_event_loop().time() < deadline:
await asyncio.sleep(poll_interval_seconds)
batch = _drain_recv_queue(ws_client)
collected_messages.extend(batch)
for msg in batch:
stream_error = _stream_completion_error(msg, invocation_id)
if stream_error:
_attach_ws_reply_parse_failure(msg, request_name, RuntimeError(stream_error), invocation_id)
fail(
f"При {purpose} бэк вернул Completion с ошибкой "
f"({request_name}, invocation_id={invocation_id}): {stream_error}"
)
for msg in batch:
if (
not isinstance(msg, list)
or msg[0] != WS_Const.STREAM_ITEM_MESSAGE_TYPE
or not is_desired_invocation_id(msg, invocation_id)
):
continue
if parser.find_reply_status_in_ws_msg(msg) is None:
continue
try:
return parser.parse_download_exported_data_msg(msg)
except Exception as error:
_attach_ws_reply_parse_failure(msg, request_name, error, invocation_id)
fail(
f"При {purpose} получен StreamItem ({request_name}, invocation_id={invocation_id}), "
f"но не удалось разобрать ответ с fileChunk: {error}"
)
finally:
collected_messages.extend(_drain_recv_queue(ws_client))
ws_client.suppress_recv_logging = False
parser.suppress_recv_logging = False
attach_ws_poll_failure(collected_messages, total_wait_seconds, f"{request_name} (StreamItem)")
fail(
f"При {purpose} за {total_wait_seconds} с не получен StreamItem с fileChunk "
f"({request_name}, invocation_id={invocation_id}). Смотреть вложения received ws message"
)
async def connect_and_get_parsed_msg_by_tu_id(
tu_id: int,
ws_client: WebSocketClient,
ws_message_type: str,
ws_invoke_type: str,
ws_invoke_params: Any = None,
timeout: float = TestConst.BASIC_MESSAGE_TIMEOUT,
) -> SubscribeAllLeaksInfoReply:
"""
Подключается, ищет и парсит allLeaksInfo сообщение для конкретного ТУ
"""
await connect(ws_client, ws_invoke_type, ws_invoke_params)
async def get_parsed_msg():
"""
Ищет и парсит allLeaksInfo сообщение для конкретного ТУ
"""
while True:
payload = await ws_client.receive_by_type(ws_message_type, timeout=timeout)
parsed_payload = ws_message_parser.parse_all_leaks_info_msg(payload)
# Ищет сообщение с нужным ТУ
if parsed_payload.replyContent.tuId == tu_id:
return parsed_payload
try:
with allure.step(f"Получение сообщения с контентом типа: {ws_message_type} для ТУ {tu_id}"):
return await asyncio.wait_for(get_parsed_msg(), timeout=timeout)
except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
fail(f"Не удалось получить сообщение allLeaksInfo для ТУ {tu_id}. Ошибка: {error}")
async def connect_and_get_msg(
ws_client: WebSocketClient,
ws_invoke_type: str,
ws_invoke_params: Any = None,
receive_timeout: Optional[float] = None,
) -> list:
"""
Подключение типа get к заданной подписке и получение сообщения с заданным типом контента
"""
await connect(ws_client, ws_invoke_type, ws_invoke_params)
invocation_id = ws_client.invocation_id
timeout = receive_timeout if receive_timeout is not None else WS_Const.FILTERING_TIMEOUT
try:
if _is_configurator_flow_active():
payload = await ws_client.receive_by_invocation_id(invocation_id, timeout=timeout)
else:
with allure.step(f"Получение входящего сообщения c invocation_id: {invocation_id}"):
payload = await ws_client.receive_by_invocation_id(invocation_id, timeout=timeout)
return payload
except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
if _is_configurator_flow_active():
raise
fail(f"Не удалось получить сообщение типа: {ws_invoke_type}. Ошибка: {error}")
async def connect_and_poll_subscribed_msg(
ws_client: WebSocketClient,
ws_message_type: str,
ws_invoke_type: str,
ws_invoke_params: Any = None,
retries: int = TestConst.SUBSCRIBE_MESSAGE_POLL_ATTEMPTS,
) -> list:
"""
Подписывается и ищет нужное сообщение в потоке, с повторным переподключением ws клиента
"""
ws_client.suppress_recv_logging = True
try:
for attempt in range(1, retries + 1):
try:
with allure.step(
f"Получение сообщения с контентом типа: {ws_message_type} - попытка {attempt} из {retries}"
):
await connect(ws_client, ws_invoke_type, ws_invoke_params)
payload = await ws_client.receive_by_type(
ws_message_type, timeout=TestConst.POLL_BY_TIMEOUT_SECONDS
)
return payload
except asyncio.TimeoutError:
await ws_client.reconnect()
continue
except Exception:
await ws_client.reconnect()
continue
fail(f"Не удалось получить сообщение с контентом типа: {ws_message_type} за {retries} попыток")
except (asyncio.TimeoutError, OSError, ConnectionError, ConnectionResetError) as error:
fail(f"Не удалось получить сообщение типа: {ws_message_type}. Ошибка: {error}")
finally:
ws_client.suppress_recv_logging = False
async def connect_and_poll_subscribed_signal(
ws_client: WebSocketClient,
ws_message_type: str,
ws_invoke_type: str,
ws_invoke_params: Any,
sensor_id: int,
sensor_description: str,
parse_payload: Callable[[list], ParsedPayloadType],
extract_signals: Callable[[ParsedPayloadType], List[InputSignalType]],
*,
max_messages: int = TestConst.SUBSCRIBE_MESSAGE_POLL_ATTEMPTS,
timeout: float = TestConst.BASIC_MESSAGE_TIMEOUT,
) -> tuple[ParsedPayloadType, InputSignalType]:
"""
Подписывается один раз и читает до max_messages сообщений из потока,
пока не найдёт сигнал с нужным id.
"""
await connect(ws_client, ws_invoke_type, ws_invoke_params)
received_ids: Set[int] = set()
try:
for attempt in range(1, max_messages + 1):
with allure.step(
f"Получение сообщения с контентом типа: {ws_message_type} - попытка {attempt} из {max_messages}"
):
payload = await ws_client.receive_by_type(ws_message_type, timeout=timeout)
parsed = parse_payload(payload)
signals = extract_signals(parsed) or []
for signal in signals:
signal_id = getattr(signal, "id", None)
if signal_id is not None:
received_ids.add(signal_id)
target_signal = find_object_by_field(signals, "id", sensor_id)
if target_signal is not None:
allure.attach(
f"Сигнал id={sensor_id} ({sensor_description}) найден на попытке {attempt} из {max_messages}\n"
f"Всего сигналов в сообщении: {len(signals)}\n"
f"Полученные id: {sorted(received_ids)}",
name=f"Результат поиска сигнала в {ws_message_type}",
attachment_type=allure.attachment_type.TEXT,
)
return parsed, target_signal
received_ids_text = ", ".join(str(signal_id) for signal_id in sorted(received_ids)) or "нет"
fail(
f"Информация по датчику {sensor_description} (id={sensor_id}) не пришла в ответе {ws_message_type} "
f"после {max_messages} попыток чтения из потока подписки. Полученные id: {received_ids_text}"
)
except (asyncio.TimeoutError, OSError, ConnectionError, ConnectionResetError) as error:
fail(
f"Не удалось получить сообщение типа: {ws_message_type} для датчика {sensor_description} "
f"(id={sensor_id}). Ошибка: {error}"
)
async def connect_and_subscribe_msg(
ws_client: WebSocketClient,
ws_message_type: str,
ws_invoke_type: str,
ws_invoke_params: Any = None,
timeout: float = TestConst.BASIC_MESSAGE_TIMEOUT,
) -> list:
"""
Подключение типа subscribe к заданной подписке и получение сообщения с заданным типом контента
"""
await connect(ws_client, ws_invoke_type, ws_invoke_params)
try:
with allure.step(f"Получение сообщения с контентом типа: {ws_message_type}"):
payload = await ws_client.receive_by_type(ws_message_type, timeout=timeout)
return payload
except (asyncio.TimeoutError, OSError, ConnectionError, ConnectionResetError) as error:
fail(f"Не удалось получить сообщение типа: {ws_invoke_type}. Ошибка: {error}")
async def poll_balance_algorithm_diagnostic_areas(
ws_client: WebSocketClient,
ws_parser: ws_message_parser,
imitator_start_time: datetime,
end_time: datetime,
poll_interval: float,
) -> list:
"""
Опрашивает очередь ws_client на наличие BalanceAlgorithmResultsContent,
собирает и возвращает все diagnosticAreas из flowAreas.
"""
collected_diagnostic_areas = []
ws_client.suppress_recv_logging = True
ws_parser.suppress_recv_logging = True
try:
while datetime.now(tz=imitator_start_time.tzinfo) < end_time:
await asyncio.sleep(poll_interval)
latest_msg = None
while not ws_client.recv_queue.empty():
try:
msg = ws_client.recv_queue.get_nowait()
except asyncio.QueueEmpty:
break
if isinstance(msg, list) and is_desired_type(msg, "BalanceAlgorithmResultsContent"):
latest_msg = msg
if latest_msg is None:
continue
parsed_payload = ws_message_parser.parse_balance_algorithm_msg(latest_msg)
reply_content = parsed_payload.replyContent
if reply_content and reply_content.flowAreas:
for flow_area in reply_content.flowAreas:
if flow_area.diagnosticAreas:
collected_diagnostic_areas.extend(flow_area.diagnosticAreas)
finally:
ws_client.suppress_recv_logging = False
ws_parser.suppress_recv_logging = False
return collected_diagnostic_areas
def get_leak_diagnostic_area_samples(
collected_diagnostic_areas: list,
leak_diagnostic_area_name: str,
total_wait: int,
) -> list:
"""
Проверяет наличие diagnosticAreas и возвращает подмножество
для ДУ с заданным leak_diagnostic_area_id. Падает, если данные не найдены.
"""
if not collected_diagnostic_areas:
fail(f"За {total_wait} секунд не пришло ни одной diagnosticArea в BalanceAlgorithmResultsContent")
leak_diagnostic_area_samples = [
diagnostic_area
for diagnostic_area in collected_diagnostic_areas
if diagnostic_area.name == leak_diagnostic_area_name
]
if not leak_diagnostic_area_samples:
fail(f"За {total_wait} секунд не пришло ни одного сообщения для ДУ с name={leak_diagnostic_area_name}.")
return leak_diagnostic_area_samples
def moscow_now() -> datetime:
"""Текущее время в часовом поясе Europe/Moscow."""
return datetime.now(ZoneInfo(TestConst.ZONE_INFO))
