https://pastein.ru/t/s7T
Загрузка данных
"""
Вспомогательные функции setup/teardown СОУ через раздел Администрирование.
"""
from __future__ import annotations
import asyncio
import logging
import time
from contextlib import contextmanager
from datetime import datetime, timedelta
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(2)
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_all_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),
exclude_tu_id,
)
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_launched_at(
http_client: StandHttpClient,
parser: WsMessageParser,
tu_id: int,
launch_checkpoint: datetime,
) -> None:
"""
Проверяет, что launchedAt в GetTusInformation не раньше момента LaunchLds с допуском.
"""
tolerance = timedelta(seconds=LdsCfgConst.LAUNCHED_AT_TOLERANCE_SECONDS)
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("Проверка: поле launchedAt заполнено"):
if launched_at is None:
fail(f"GetTusInformationResponse: launchedAt отсутствует для tuId={tu_id}", pytrace=False)
launched_at_msk = t_utils.localize_as_moscow(launched_at)
checkpoint_msk = t_utils.localize_as_moscow(launch_checkpoint)
compare_msg = (
f"launchedAt: {t_utils.format_datetime_moscow(launched_at_msk)}\n"
f"checkpoint: {t_utils.format_datetime_moscow(checkpoint_msk)}\n"
f"tolerance: {LdsCfgConst.LAUNCHED_AT_TOLERANCE_SECONDS} с"
)
if _configurator_flow_active:
logger.info("[LDS_CONFIGURATOR] Сравнение времени запуска СОУ:\n%s", compare_msg)
else:
allure.attach(
compare_msg,
name="Сравнение времени запуска СОУ",
attachment_type=allure.attachment_type.TEXT,
)
with _step("Проверка: launchedAt не раньше момента команды 'Запустить СОУ' (с допуском)"):
if launched_at_msk < checkpoint_msk - tolerance:
fail(
f"Время запуска СОУ на бэкенде ({t_utils.format_datetime_moscow(launched_at_msk)}) "
f"раньше момента команды 'Запустить СОУ' "
f"({t_utils.format_datetime_moscow(checkpoint_msk)}) "
f"с учётом допуска {int(LdsCfgConst.LAUNCHED_AT_TOLERANCE_SECONDS)} с",
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
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
from constants.enums import ExportedDataType, ExportStatus, ReplyStatus
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):
return datetime.fromisoformat(value)
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: list) -> GetExportedDataListReply:
"""
Парсит ответ GetExportedDataListReply со списком сформированных файлов.
"""
return self._find_and_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(WebSocketClientConstants.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 find_reply_status_in_ws_msg(data: List[Any]) -> Optional[Dict[str, Any]]:
"""
Ищет объект с replyStatus в ws сообщении
"""
if not data:
fail("Пустое сообщение")
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
from __future__ import annotations
import asyncio
import pprint
import random
import re
from dataclasses import asdict, is_dataclass
from datetime import datetime, timedelta, timezone
from enum import IntEnum, IntFlag
from typing import Any, Callable, List, Optional, 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.websocket_client import WebSocketClient
from constants.architecture_constants import WebSocketClientConstants as WS_Const
from constants.enums import (
ConfirmationStatus,
ExportStatus,
LdsStatus,
LdsStatusDegradation,
LdsStatusFaulty,
LdsStatusInitialization,
LeakStatus,
ReplyStatus,
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")
SignalType = TypeVar("SignalType")
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 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_a_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 parse_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, TestConst.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 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_msgpack_timestamp(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],
invocation_id: str,
request_name: str,
error: BaseException,
) -> 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(
ws_client: WebSocketClient,
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 = ReportConst.GET_EXPORTED_DATA_LIST_REQUEST
ws_client.suppress_recv_logging = True
parser.suppress_recv_logging = True
try:
while asyncio.get_event_loop().time() < deadline:
drained_before_request = _drain_recv_queue(ws_client)
collected_messages.extend(drained_before_request)
await connect(
ws_client,
request_name,
{"limit": list_limit},
)
invocation_id = ws_client.invocation_id
await asyncio.sleep(poll_interval_seconds)
batch = _drain_recv_queue(ws_client)
collected_messages.extend(batch)
list_reply_payload = _find_ws_reply_by_invocation_id(batch, invocation_id, parser)
if list_reply_payload is None:
continue
try:
parsed_payload = parser.parse_exported_data_list_msg(list_reply_payload)
except Exception as error:
_attach_ws_reply_parse_failure(list_reply_payload, invocation_id, request_name, error)
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,
)
fail(f"Не удалось разобрать ответ на {request_name}: {error}")
items = []
if parsed_payload.replyContent is not None:
items = parsed_payload.replyContent.exportedData or []
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
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,
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, invocation_id, request_name, RuntimeError(stream_error))
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, invocation_id, request_name, error)
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_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[SignalType]],
*,
max_messages: int = TestConst.SUBSCRIBE_MESSAGE_POLL_ATTEMPTS,
timeout: float = TestConst.BASIC_MESSAGE_TIMEOUT,
) -> tuple[ParsedPayloadType, SignalType]:
"""
Подписывается один раз и читает до 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
import asyncio
import glob
import os
import shutil
import threading
import time
from typing import Optional
import allure
import pytest
import pytest_asyncio
from clients.http_client import StandHttpClient
from clients.keycloak_clients import KeycloakAuthError, KeycloakClient
from clients.testops_client import AllureResultsUploader, 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 ImitatorConstants as ImConst
from constants.architecture_constants import WebSocketClientConstants as WSCliConst
from constants.enums import RejectionSensorTag
from constants.test_constants import BaseTN3Constants
from infra.stand_setup_manager import StandSetupManager
from test_config.datasets import get_config_by_name
from test_scenarios import lds_configurator_scenarios
from utils.helpers import lds_configurator_utils as lds_cfg_utils
from utils.helpers.ws_message_parser import ws_message_parser as lds_ws_parser
def pytest_addoption(parser):
"""
Добавляет кастомные опции командной строки pytest.
"""
parser.addoption(
"--suites",
action="store",
default=None,
help="Запустить только указанные наборы данных. Пример: --suites=select_4,select_19_20",
)
def _find_config_by_suite_name(suite_name: str):
"""Находит конфиг по имени набора данных."""
try:
return get_config_by_name(suite_name)
except ValueError:
return None
@pytest.fixture(autouse=True)
def allure_suite_hierarchy(request):
"""
Автоматически устанавливает иерархию Allure для группировки тестов по наборам данных.
В Allure отчёте тесты группируются:
- Parent Suite: SingleLeakSuite / MultiLeakSuite (тип набора)
- Suite: select_4 / select_6 / ... (имя набора данных)
Работает как с параметризованными тестами (config в параметрах),
так и с обычными тестами (через маркер test_suite_name).
"""
config = None
suite_name = None
# Пробуем получить конфиг из параметризации
if hasattr(request, 'fixturenames') and 'config' in request.fixturenames:
try:
config = request.getfixturevalue('config')
suite_name = config.suite_name
except Exception:
pass
# Если не нашли, пробуем найти конфиг по маркеру test_suite_name
if not config:
marker = request.node.get_closest_marker('test_suite_name')
if marker:
suite_name = marker.args[0]
config = _find_config_by_suite_name(suite_name)
if config and suite_name:
parent_suite = "MultiLeakSuite" if config.has_multiple_leaks else "SingleLeakSuite"
allure.dynamic.parent_suite(parent_suite)
allure.dynamic.suite(suite_name)
def pytest_configure(config):
"""
Храним состояние сессии
"""
config.addinivalue_line("markers", "critical_stop: если тест упал, останавливаем дальнейшее выполнение сессии")
config.group_state = {
"current_suite": None,
"suite_start_time": None,
"stand_manager": None,
"imitator_start_time": None, # datetime объект времени старта имитатора для расчёта интервалов утечек
"use_lds_configurator": True,
"resolved_tu_id": None,
"admin_tu_name": None,
"pre_run_running_tus": None,
"suite_infra_ready": False,
}
def _update_sensor_ids(stand_manager: StandSetupManager) -> None:
"""
Для тестов датчиков обновляет sensor_id по address из конфигурации стенда.
"""
sensor_ids_by_address = stand_manager.get_sensor_ids_by_address()
BaseTN3Constants.SENSOR_IDS_BY_ADDRESS.update(sensor_ids_by_address)
RejectionSensorTag.update_ids_from_config(sensor_ids_by_address)
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
"""
Делает падение критического теста с маркировкой critical_stop однозначным:
- рисуем fail для теста
- после него прекращаем запуск остальных тестов
"""
outcome = yield
report = outcome.get_result()
if report.when == "call" and report.failed and item.get_closest_marker("critical_stop"):
item.session.shouldstop = f"Критическая проверка упала: {item.nodeid}"
# ===== Маппинг имён тестов на атрибуты конфига для получения маркеров =====
# Используется для добавления offset и test_case_id маркеров во время сбора тестов
# Smoke-тесты уровня набора (маркеры из SmokeSuiteConfig)
SMOKE_SUITE_LEVEL_MAPPING = {
'test_basic_info': 'basic_info_test',
'test_journal_info': 'journal_info_test',
'test_imitate_pressure_sensor_signal': 'imitate_pressure_sensor_signal_test',
'test_imitate_flowmeter_signal': 'imitate_flowmeter_signal_test',
'test_lds_status_initialization': 'lds_status_initialization_test',
'test_lds_status_init_in_journal': 'lds_status_init_in_journal_test',
'test_main_page_info': 'main_page_info_test',
'test_main_page_info_signals': 'main_page_info_signals_test',
'test_mask_signal': 'mask_signal_test',
'test_mask_info_in_journal': 'mask_info_in_journal_test',
'test_lds_status_initialization_out': 'lds_status_initialization_out_test',
'test_lds_status_init_out_in_journal': 'lds_status_init_out_in_journal_test',
'test_main_page_info_unstationary': 'main_page_info_unstationary_test',
'test_mask_du_on_mini_scheme': 'mask_du_on_mini_scheme_test',
'test_unmask_du_on_mini_scheme': 'unmask_du_on_mini_scheme_test',
'test_lds_status_after_confirming_leak': 'lds_status_after_confirming_leak_test',
'test_lds_status_completed_leak': 'lds_status_completed_leak_test',
'test_diagnostics_of_signals_after_initialization': 'diagnostics_of_signals_after_initialization_test',
'test_mode_mt_in_journal': 'mode_mt_in_journal_test',
'test_export_lds_status_report': 'export_lds_status_report_test',
'test_export_mt_mode_report': 'export_mt_mode_report_test',
}
# Regress-тесты режимов СОУ (маркеры из LDSStatusConfig)
LDS_STATUS_SUITE_LEVEL_MAPPING = {
'test_lds_status_basic_info': 'lds_status_basic_info_test',
'test_lds_status_init_accumulation_data': 'init_accumulation_data_test',
'test_lds_status_init_accumulation_data_in_journal': 'init_accumulation_data_in_journal_test',
'test_lds_status_init_cold_start': 'init_cold_start_test',
'test_lds_status_init_cold_start_in_journal': 'init_cold_start_in_journal_test',
'test_lds_status_init_exiting_faulty': 'init_exiting_faulty_test',
'test_lds_status_init_switching_shut_off': 'init_switching_shut_off_test',
'test_lds_status_init_switching_shut_off_in_journal': 'init_switching_shut_off_in_journal_test',
'test_lds_status_serviceable_after_cold_start': 'serviceable_after_cold_start_test',
'test_lds_status_serviceable_after_cold_start_in_journal': 'serviceable_after_cold_start_in_journal_test',
'test_lds_status_serviceable_after_switching_shut_off': 'serviceable_after_switching_shut_off_test',
'test_lds_status_serviceable_after_switching_shut_off_in_journal': 'serviceable_after_switching_shut_off_in_journal_test', # noqa: E501
'test_lds_status_serviceable_after_deg_absence_min_pressure_sensors': 'serviceable_after_deg_absence_min_pressure_sensors_test', # noqa: E501
'test_lds_status_serviceable_after_deg_additive_injectors_operation': 'serviceable_after_deg_additive_injectors_operation_test', # noqa: E501
'test_lds_status_serviceable_after_deg_exceeding_distance_between_flow_meters': 'serviceable_after_deg_exceeding_distance_between_flow_meters_test', # noqa: E501
'test_lds_status_serviceable_after_deg_starting_pumping_out_pumps': 'serviceable_after_deg_starting_pumping_out_pumps_test', # noqa: E501
'test_lds_status_serviceable_after_deg_faulty_pressure_sensors_at_pump': 'serviceable_after_deg_faulty_pressure_sensors_at_pump_test', # noqa: E501
'test_lds_status_serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal': 'serviceable_after_deg_faulty_pressure_sensors_at_pump_in_journal_test', # noqa: E501
'test_lds_status_serviceable_after_faulty': 'serviceable_after_faulty_test',
'test_lds_status_degradation_additive_injectors_operation': 'deg_additive_injectors_operation_test',
'test_lds_status_degradation_exceeding_distance_between_pressure_sensors': 'deg_exceeding_distance_between_pressure_sensors_test', # noqa: E501
'test_lds_status_degradation_exceeding_distance_between_pressure_sensors_in_journal': 'deg_exceeding_distance_between_pressure_sensors_in_journal_test', # noqa: E501
'test_lds_status_degradation_absence_min_pressure_sensors': 'deg_absence_min_pressure_sensors_test',
'test_lds_status_degradation_faulty_pressure_sensors_at_pump_station': 'deg_faulty_pressure_sensors_at_pump_station_test', # noqa: E501
'test_lds_status_degradation_faulty_pressure_sensors_at_pump_station_in_journal': 'deg_faulty_pressure_sensors_at_pump_station_in_journal_test', # noqa: E501
'test_lds_status_degradation_gravity_section_pumping': 'deg_gravity_section_pumping_test',
'test_lds_status_degradation_gravity_section_pumping_in_stopping': 'deg_gravity_section_pumping_in_stopping_test',
'test_lds_status_degradation_gravity_section_pumping_in_stopping_in_journal': 'deg_gravity_section_pumping_in_stopping_in_journal_test', # noqa: E501
'test_lds_status_degradation_pig_sensor_passage': 'deg_pig_sensor_passage_test',
'test_lds_status_degradation_starting_pumping_out_pumps': 'deg_starting_pumping_out_pumps_test',
'test_lds_status_degradation_exceeding_distance_between_flow_meters': 'deg_exceeding_distance_between_flow_meters_test', # noqa: E501
'test_lds_status_degradation_rejection_temperature_sensor_on_du_2': 'deg_rejection_temperature_sensor_on_du_2_test',
'test_lds_status_degradation_rejection_temperature_sensor_on_du_3': 'deg_rejection_temperature_sensor_on_du_3_test',
'test_lds_status_degradation_rejection_temperature_sensor_on_du_5': 'deg_rejection_temperature_sensor_on_du_5_test',
'test_lds_status_degradation_rejection_density_and_viscosity_on_du_2': 'deg_rejection_density_and_viscosity_on_du_2_test', # noqa: E501
'test_lds_status_degradation_rejection_density_and_viscosity_on_du_3': 'deg_rejection_density_and_viscosity_on_du_3_test', # noqa: E501
'test_lds_status_degradation_rejection_density_and_viscosity_on_du_5': 'deg_rejection_density_and_viscosity_on_du_5_test', # noqa: E501
'test_lds_status_faulty_absence_min_flow_meters_continuous': 'faulty_absence_min_flow_meters_continuous_test',
'test_lds_status_faulty_absence_min_flow_meters': 'faulty_absence_min_flow_meters_test',
'test_lds_status_faulty_absence_min_pressure_sensors': 'faulty_absence_min_pressure_sensors_test',
'test_lds_status_faulty_absence_min_pressure_sensors_in_journal': 'faulty_absence_min_pressure_sensors_in_journal_test', # noqa: E501
}
# Тесты уровня утечки (маркеры из LeakTestConfig - параметр leak)
LEAK_LEVEL_TEST_MAPPING = {
'test_leaks_content': 'leaks_content_test',
'test_all_leaks_info': 'all_leaks_info_test',
'test_tu_leaks_info': 'tu_leaks_info_test',
'test_leak_info_in_journal': 'leak_info_in_journal',
'test_possible_leak_in_journal': 'possible_leak_in_journal_test',
'test_acknowledge_leak_info': 'acknowledge_leak_test',
'test_acknowledge_leak_in_journal': 'acknowledge_leak_in_journal_test',
'test_output_signals': 'output_signals_test',
'test_lds_status_during_leak': 'lds_status_during_leak_test',
'test_balance_algorithm_leak_waiting': 'balance_algorithm_leak_waiting_test',
'test_balance_algorithm_leak_detected': 'balance_algorithm_leak_detected_test',
'test_the_leak_is_complete_on_kg': 'the_leak_is_complete_on_kg_test',
'test_leak_is_complete_in_output_signals': 'leak_is_complete_in_output_signals_test',
'test_complete_tu_leaks_info_content': 'complete_tu_leaks_info_content_test',
'test_all_leaks_is_empty': 'all_leaks_is_empty_test',
'test_leak_is_confirm_on_main_page': 'leak_is_confirm_on_main_page_test',
'test_leak_is_complete_on_main_page': 'leak_is_complete_on_main_page_test',
'test_balance_algorithm_leak_completed': 'balance_algorithm_leak_completed_test',
'test_completed_leak_info_in_journal': 'completed_leak_info_in_journal_test',
'test_export_leaks_report': 'export_leaks_report_test',
}
STATIONARY_STATUS_SUITE_LEVEL_MAPPING = {
'test_stationary_status_basic_info': 'stationary_status_basic_info_test',
'test_stationary_status_check_with_reasons': 'stationary_status_check_with_reasons_test',
'test_stationary_status_in_journal': 'stationary_status_in_journal_test',
}
# Тесты уровня отбраковки (маркеры из RejectionTestCase - параметр rejection_case)
IS_REJECTED_LEVEL_TEST_MAPPING = {
'test_rejection_input_signals': 'rejection_input_signals_test',
'test_rejection_journal': 'rejection_journal_test',
'test_rejection_main_page': 'rejection_main_page_test',
'test_rejection_scheme_signals_state': 'rejection_scheme_signals_state_test',
}
# Suite-level тесты отбраковки (маркеры из IsRejectedConfig - параметр config)
IS_REJECTED_SUITE_LEVEL_MAPPING = {
'test_rejection_report': 'rejection_report_test',
}
# Мержим все вместе чтобы не переписывать логику коллектора айтемов (тестов)
SUITE_LEVEL_TEST_MAPPING = {
**SMOKE_SUITE_LEVEL_MAPPING,
**LDS_STATUS_SUITE_LEVEL_MAPPING,
**STATIONARY_STATUS_SUITE_LEVEL_MAPPING,
**IS_REJECTED_SUITE_LEVEL_MAPPING,
}
def _get_test_markers_config(item, test_name):
"""
Получает конфигурацию с маркерами для теста.
Для leak-level тестов: маркеры берутся из параметра leak
Для suite-level тестов: маркеры берутся из config
:return: CaseMarkers объект или None
"""
if not hasattr(item, 'callspec'):
return None
params = item.callspec.params
# Проверяем, есть ли параметр leak (для leak-level тестов)
if 'leak' in params and test_name in LEAK_LEVEL_TEST_MAPPING:
leak = params['leak']
attr_name = LEAK_LEVEL_TEST_MAPPING[test_name]
return getattr(leak, attr_name, None)
# Проверяем, есть ли параметр rejection_case для тестов отбраковки
if 'rejection_case' in params and test_name in IS_REJECTED_LEVEL_TEST_MAPPING:
rejection_case = params['rejection_case']
attr_name = IS_REJECTED_LEVEL_TEST_MAPPING[test_name]
return getattr(rejection_case, attr_name, None)
# Suite-level тесты отбраковки (без rejection_case)
if 'config' in params and 'rejection_case' not in params and test_name in IS_REJECTED_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = IS_REJECTED_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
# Для suite-level тестов берём из config
if 'config' in params:
if test_name in SMOKE_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = SMOKE_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
if test_name in LDS_STATUS_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = LDS_STATUS_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
if test_name in STATIONARY_STATUS_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = STATIONARY_STATUS_SUITE_LEVEL_MAPPING[test_name]
return getattr(suite_config, attr_name, None)
return None
def pytest_collection_modifyitems(session, config, items):
"""
1. Фильтрует тесты по --suites (если указано)
2. Исключает тесты, у которых конфиг = None (тест отключён для этого набора данных)
3. Добавляет маркеры offset и test_case_id из конфига к каждому параметризованному тесту
4. Сортирует тесты по test_suite_name для группировки по наборам данных
"""
# Получаем список выбранных наборов из --suites
suites_option = config.getoption("--suites")
selected_suites = None
if suites_option:
# Парсим список наборов: "select_4,select_19_20" -> ["select_4", "select_19_20"]
selected_suites = [s.strip().lower() for s in suites_option.split(",")]
selected_items = []
deselected_items = []
for item in items:
# Фильтрация по --suites
if selected_suites:
suite_marker = item.get_closest_marker("test_suite_name")
if suite_marker:
suite_name = suite_marker.args[0].lower()
# Проверяем, содержит ли имя набора одну из выбранных подстрок
if not any(selected in suite_name for selected in selected_suites):
deselected_items.append(item)
continue
# Получаем имя функции теста (без параметров)
test_name = item.originalname or item.name.split('[')[0]
# Получаем конфиг с маркерами для теста
test_config = _get_test_markers_config(item, test_name)
if test_config is not None:
# Добавляем маркер offset
if hasattr(test_config, 'offset') and test_config.offset is not None:
item.add_marker(pytest.mark.offset(test_config.offset))
# Добавляем маркер test_case_id
if hasattr(test_config, 'test_case_id') and test_config.test_case_id is not None:
item.add_marker(pytest.mark.test_case_id(test_config.test_case_id))
elif (
test_name in SUITE_LEVEL_TEST_MAPPING
or test_name in LEAK_LEVEL_TEST_MAPPING
or test_name in IS_REJECTED_LEVEL_TEST_MAPPING
):
# Конфиг теста = None - исключаем тест из прогона
deselected_items.append(item)
continue
selected_items.append(item)
# Уведомляем pytest об исключённых тестах
if deselected_items:
config.hook.pytest_deselected(items=deselected_items)
# Заменяем список тестов на отфильтрованный
items[:] = selected_items
# Сортировка тестов по test_suite_name и offset
# Цель: обеспечить запуск тестов строго по offset строго внутри набора данных
# При равных offset сохраняем исходный порядок коллекции, чтобы порядок параметризации не перескакивал
def suite_offset_key(item):
"""
Сортировка тестов по test_suite_name и offset (без падения на None).
"""
test_suite_name_marker = item.get_closest_marker("test_suite_name")
test_suite_name = test_suite_name_marker.args[0] if test_suite_name_marker else ""
offset_marker = item.get_closest_marker("offset")
if offset_marker:
try:
offset_value = float(offset_marker.args[0])
except Exception:
offset_value = float("inf")
else:
offset_value = float("inf")
original_index = getattr(item, "_collection_index", 0)
# Возвращаем тройку ключей сортировки
# 1) test_suite_name - группировка по набору
# 2) offset_value - порядок внутри набора по времени
# 3) original_index - стабильность при равных offset
return test_suite_name, offset_value, original_index
# Сохраняем исходный порядок коллекции для стабильной сортировки
for index, item in enumerate(items):
item._collection_index = index
# по кортежу питон сортирует слева направо, благодаря этому сортировка по offset идет строго внутри test_suite_name
items.sort(key=suite_offset_key)
for item in items:
if hasattr(item, "_collection_index"):
delattr(item, "_collection_index")
@pytest.fixture(autouse=True)
def allure_tms_link(request):
"""
Allure TMS‑линки по test_case_id
"""
if test_case_id_marker := request.node.get_closest_marker("test_case_id"):
test_case_id = test_case_id_marker.args[0]
allure.dynamic.link(
f"https://{os.environ['TESTOPS_BASE_URL']}/testcases?selected_id={test_case_id}",
name=f"TestCase-{test_case_id}",
link_type="tms",
)
@pytest.fixture(autouse=True)
def offset_wait(request):
"""
Offset‑ожидание перед каждым тестом относительно фактического старта core
"""
if offset_marker := request.node.get_closest_marker("offset"):
offset_sec = float(offset_marker.args[0]) * BaseTN3Constants.SEC_PER_MIN
start = request.config.group_state["suite_start_time"] or 0
elapsed = time.monotonic() - start
to_wait = max(0, offset_sec - elapsed)
if to_wait:
time.sleep(to_wait)
def compute_imitator_duration(item, current_test_suite: str) -> float:
"""
Вычисляет длительность для имитатора (в минутах).
Правило:
- Собирает все тесты (item.session.items) с меткой test_suite_name == current_test_suite
- Извлекает все значения @pytest.mark.offset(...) (в минутах)
- Если offsets найдены: возвращает max(offsets) + IMITATOR_FINISH_DELAY задержка остановки имитатора
- Иначе: если у текущего item есть @pytest.mark.imitator_duration — используется как fallback и логируется
- Если ничего не найдено — pytest.fail с понятным текстом
"""
suite_items = [
suite_item
for suite_item in item.session.items
if (marker := suite_item.get_closest_marker("test_suite_name")) and marker.args[0] == current_test_suite
]
offsets = []
for suite_item in suite_items:
offset_marker = suite_item.get_closest_marker("offset")
if offset_marker:
try:
offsets.append(float(offset_marker.args[0]))
except Exception:
continue
if offsets:
max_offset = max(offsets)
imitator_duration = float(max_offset) + ImConst.IMITATOR_FINISH_DELAY_MINUTE
return imitator_duration
else:
# fallback- если все еще задан старый маркер imitator_duration, то используем его
if imitator_mark := item.get_closest_marker("imitator_duration"):
imitator_duration = float(imitator_mark.args[0])
logger.warning(
"[DEPRECATED] использован pytest.mark.imitator_duration()"
f"рекомендуется убрать и полагаться на max_offset + {ImConst.IMITATOR_FINISH_DELAY_MINUTE}"
)
return imitator_duration
pytest.fail(
"Не удалось вычислить imitator_duration: в тестовом модуле одновременно отсутствуют "
"и @pytest.mark.offset(), и pytest.mark.imitator_duration()"
)
@pytest.fixture(autouse=True)
def require_suite_infra(request):
"""
Блокирует тесты набора, если infra-setup не завершился успешно.
"""
if not request.node.get_closest_marker("test_suite_name"):
return
cfg = request.config.group_state
if cfg.get("current_suite") and not cfg.get("suite_infra_ready"):
pytest.exit(
"[SETUP] [ERROR] Инфраструктура набора не готова (suite_infra_ready=False). " "Тесты не запускаются."
)
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_setup(item):
"""
Перезапуск имитатора при смене test_suite_name
"""
cfg = item.config.group_state
test_suite_marker = item.get_closest_marker("test_suite_name")
if not test_suite_marker:
pytest.fail("Тест без @pytest.mark.test_suite_name")
current_test_suite = test_suite_marker.args[0]
if current_test_suite != cfg["current_suite"]:
# stop old
_run_lds_configurator_teardown_if_needed(cfg)
if stand_manager := cfg["stand_manager"]:
stand_manager.stop_imitator_wrapper()
try:
stand_manager.restore_signal_unit_conversion_rules()
except Exception:
logger.exception(
"[ERROR] [SETUP] Ошибка при восстановлении signal_unit_conversion_rules.json "
"перед запуском нового набора"
)
if not os.environ.get("RUN_WITHOUT_TESTOPS", "False").lower() == "true":
# При запуске с TestOps удаляет данные прогона
stand_manager.server_test_data_remover()
# start new
cfg["current_suite"] = current_test_suite
cfg["suite_start_time"] = None
cfg["suite_infra_ready"] = False
data_id = item.get_closest_marker("test_suite_data_id").args[0]
test_data_name = item.get_closest_marker("test_data_name").args[0]
# legacy: id из enum TU для имитатора (tn{id}_tags.txt), не resolved_tu_id из Администрирования
tu_id = item.get_closest_marker("tu_id").args[0]
imitator_duration = compute_imitator_duration(item, current_test_suite)
suite_config = _find_config_by_suite_name(current_test_suite)
measure_conversion_rules = suite_config.measure_conversion_rules if suite_config is not None else None
if suite_config is not None and suite_config.use_lds_configurator:
if not suite_config.admin_tu_name.strip():
pytest.exit(
f"[SETUP] [ERROR] Набор '{suite_config.suite_name}': admin_tu_name обязателен "
"при use_lds_configurator=True"
)
stand_manager = StandSetupManager(
duration_m=imitator_duration,
test_data_id=data_id,
test_data_name=test_data_name,
tu_id=tu_id,
measure_conversion_rules=measure_conversion_rules,
)
cfg["stand_manager"] = stand_manager
try:
stand_manager.check_opc_server_status()
except RuntimeError as error:
msg = (
"[SETUP] [ERROR] OPC сервер недоступен. Имитатор и автотесты не запущены. "
f"Ошибка при проверке статуса OPC: {error}"
)
allure.attach(msg, name="OPC сервер недоступен", attachment_type=allure.attachment_type.TEXT)
pytest.exit(msg)
try:
stand_manager.setup_stand_for_imitator_run()
except Exception as error:
pytest.exit(f"[SETUP] [ERROR] ошибка при подготовке стенда: {error}")
try:
_update_sensor_ids(stand_manager)
except Exception as error:
pytest.exit(f"[SETUP] [ERROR] ошибка обновления id датчиков отбраковки из конфигурации: {error}")
if suite_config is not None and suite_config.use_lds_configurator:
try:
_run_lds_admin_setup(suite_config, cfg)
except BaseException as error:
pytest.exit(f"[SETUP] [ERROR] LDS Configurator admin setup: {error}")
imitator_thread = threading.Thread(
target=stand_manager.start_imitator, name=f"imitator->{current_test_suite}", daemon=True
)
core_thread = threading.Thread(target=stand_manager.start_core)
try:
imitator_thread.start()
except Exception as error:
pytest.exit(f"[SETUP] [ERROR] ошибка запуска имитатора: {error}")
time.sleep(ImConst.CORE_START_DELAY_S)
try:
cfg["suite_start_time"] = time.monotonic()
core_thread.start()
core_thread.join(timeout=5)
except Exception as error:
pytest.exit(f"[SETUP] [ERROR] ошибка запуска СORE контейнеров: {error}")
# Сохраняем время старта имитатора для расчёта интервалов утечек в тестах
cfg["imitator_start_time"] = stand_manager.start_time
if suite_config is not None and suite_config.use_lds_configurator:
if suite_config.admin_tu is None:
pytest.exit(
f"[SETUP] [ERROR] Набор '{suite_config.suite_name}': admin_tu обязателен "
"при use_lds_configurator=True"
)
try:
_run_lds_verify_after_core(suite_config)
except BaseException as error:
# Имитатор остановится в pytest_sessionfinish через stop_imitator_wrapper
pytest.exit(f"[SETUP] [ERROR] LDS Configurator проверка после запуска lds-core: {error}")
cfg["suite_infra_ready"] = True
yield # pytest продолжит выполнение теста
def _run_lds_configurator_ws(coro_factory) -> None:
"""Запускает async WS-сценарий lds-configurator"""
lds_cfg_utils.set_configurator_flow_active(True)
lds_ws_parser.suppress_recv_logging = True
try:
asyncio.run(coro_factory())
finally:
lds_ws_parser.suppress_recv_logging = False
lds_cfg_utils.set_configurator_flow_active(False)
def _run_lds_admin_setup(suite_config, group_state: dict) -> None:
"""
Setup СОУ через Администрирование до старта имитатора.
"""
async def _admin_setup() -> None:
http_client = init_http_stand_client()
await lds_configurator_scenarios.lds_configurator_admin_setup(http_client, suite_config, group_state)
_run_lds_configurator_ws(_admin_setup)
def _run_lds_verify_after_core(suite_config) -> None:
"""
WS-проверка готовности стенда после запуска lds-core.
"""
async def _verify() -> None:
websocket_client = init_ws_stand_client()
http_client = init_http_stand_client()
async with websocket_client as ws_client:
ws_client.suppress_recv_logging = True
await lds_configurator_scenarios.lds_configurator_verify_after_core(ws_client, http_client, suite_config)
_run_lds_configurator_ws(_verify)
def _run_lds_configurator_teardown_if_needed(cfg: dict) -> None:
"""
Выполняет WS-teardown СОУ при смене набора или завершении сессии.
Ошибки не прерывают pytest: логируются и оформляются как ALERT внутри сценария.
Состояние group_state сбрасывается в finally.
"""
if not cfg.get("use_lds_configurator") or not cfg.get("resolved_tu_id"):
return
tu_id = cfg["resolved_tu_id"]
admin_tu_name = cfg.get("admin_tu_name") or ""
pre_run_running_tus = cfg.get("pre_run_running_tus") or []
async def _teardown() -> None:
websocket_client = init_ws_stand_client()
http_client = init_http_stand_client()
async with websocket_client as ws_client:
ws_client.suppress_recv_logging = True
await lds_configurator_scenarios.lds_configurator_teardown(
ws_client, http_client, tu_id, admin_tu_name, pre_run_running_tus
)
try:
_run_lds_configurator_ws(_teardown)
except BaseException as error:
logger.warning(
"[TEARDOWN] [ALERT] LDS Configurator teardown завершился с ошибкой для tuId=%s: %s",
tu_id,
error,
)
finally:
cfg["resolved_tu_id"] = None
cfg["use_lds_configurator"] = False
cfg["admin_tu_name"] = None
cfg["pre_run_running_tus"] = None
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_teardown(item, nextitem):
"""
Teardown имитатора при выходе из группы
"""
yield
cfg = item.config.group_state
next_marker = nextitem.get_closest_marker("test_suite_name") if nextitem else None
next_suite = next_marker.args[0] if next_marker else None
if next_suite != cfg["current_suite"]:
if stand_manager := cfg["stand_manager"]:
stand_manager.stop_imitator_wrapper()
try:
stand_manager.restore_signal_unit_conversion_rules()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при восстановлении signal_unit_conversion_rules.json")
if not os.environ.get("RUN_WITHOUT_TESTOPS", "False").lower() == "true":
# При запуске с TestOps удаляет данные прогона
stand_manager.server_test_data_remover()
cfg["stand_manager"] = None
cfg["current_suite"] = None
cfg["suite_start_time"] = None
cfg["imitator_start_time"] = None
cfg["suite_infra_ready"] = False
# опционально дождаться завершения потока (если не daemon) — безопасный join
imitator_thread = cfg.get("imitator_thread")
if imitator_thread and not getattr(imitator_thread, "daemon", False):
try:
imitator_thread.join(timeout=5)
except RuntimeError:
logger.exception("Ошибка при join() фона имитатора")
def get_instance() -> str:
"""
Получает имя стенда
"""
instance = os.environ.get(EnvConst.STAND_NAME)
if not instance:
pytest.exit(f"Переменная окружения {EnvConst.STAND_NAME} не задана в .env")
return instance
def build_stand_host() -> str:
"""
Создает URL для ws и http запросов к стенду
"""
instance = get_instance()
ws_host = (
f"{WSCliConst.SERVICE_NAME}.{WSCliConst.COMPONENT}-{instance}.{WSCliConst.ROOT_DOMAIN}/"
f"{WSCliConst.API_GATEWAY_PATH_SEGMENT}"
)
return ws_host
def get_token(max_retries: int = 10, backoff: float = 5.0) -> str:
"""
:param max_retries: сколько всего попыток (включая первую)
:param backoff: время в секундах между попытками
"""
last_exc = None
keycloak = 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),
)
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 get_connection_params() -> tuple:
"""
Получает адрес стенда и токен авторизации
"""
stand_host = build_stand_host()
auth_token = get_token()
return stand_host, auth_token
def init_http_stand_client() -> StandHttpClient:
"""
Создает экземпляр класса StandHttpClient для запросов к стенду
"""
stand_host, auth_token = get_connection_params()
http_client = StandHttpClient(stand_host, auth_token)
return http_client
def get_x_user_id() -> Optional[str]:
"""
Получает x-user-id из headers Ping запроса
"""
http_client = init_http_stand_client()
try:
response = http_client.post_request(HttpConst.PING_URL_PATH, {})
logger.info("[HTTP_CLIENT] [OK] Успешно получен x-user-id")
return response.headers.get(HttpConst.X_USER_ID_KEY)
except Exception as error:
logger.error(f"[HTTP_CLIENT] [ERROR] Не удалось получить x-user-id: {error}")
pytest.exit(f"[HTTP_CLIENT] [ERROR] Не удалось получить x-user-id: {error}")
def init_ws_stand_client() -> WebSocketClient:
"""
Создает экземпляр класса WebSocketClient для запросов к стенду
"""
ws_host, token = get_connection_params()
x_user_id = get_x_user_id()
ws_client = WebSocketClient(ws_host, token, x_user_id)
return ws_client
@pytest_asyncio.fixture
async def ws_client():
"""
Фикстура для работы с websocket клиентом
:return: Объект wss соединения
"""
ws_client = init_ws_stand_client()
async with ws_client as client:
yield client
@pytest.fixture
def http_client():
"""
Фикстура для работы с http клиентом
:return: экземпляр класса для выполнения http запросов к стенду
"""
http_client = init_http_stand_client()
yield http_client
@pytest.fixture
def imitator_start_time(request):
"""
Фикстура для получения времени старта имитатора (datetime объект).
Используется для точного расчёта времени обнаружения утечек:
- leak_start_time = imitator_start_time + timedelta(seconds=LEAK_START_INTERVAL)
- leak_end_time = imitator_start_time + timedelta(seconds=LEAK_START_INTERVAL + ALLOWED_TIME_DIFF_SECONDS)
"""
start_time = request.config.group_state.get("imitator_start_time")
if start_time is None:
pytest.fail("imitator_start_time не установлен. Убедитесь что тест запущен после инициализации имитатора.")
return start_time
def pytest_sessionfinish(session, exitstatus):
"""
В завершении сессии — отправляем единый Allure‑отчёт в TestOps.
"""
# 1) teardown стенда: LDS Configurator + остановка имитатора
try:
group_state = getattr(session.config, "group_state", {})
_run_lds_configurator_teardown_if_needed(group_state)
stand_manager = group_state.get("stand_manager")
if stand_manager:
try:
stand_manager.stop_imitator_wrapper()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при остановке имитатора")
try:
stand_manager.restore_signal_unit_conversion_rules()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при восстановлении signal_unit_conversion_rules.json")
try:
stand_manager.server_test_data_remover()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при удалении тестового набора данных со стенда")
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при получении stand_manager из group_state")
# 2) Выгрузка allure-results в TestOps
try:
uploader = AllureResultsUploader()
logger.info("[INFO] [TEARDOWN] Uploading Allure results to TestOps")
uploader.upload_allure_results()
except Exception:
logger.exception("[ERROR] [TEARDOWN] Ошибка при выгрузке allure-results в TestOps")
# 3) Удаление локальных архивов с данными
shutil.rmtree("allure-results")
project_root = os.path.dirname(os.path.abspath(__file__))
files_for_drop = glob.glob(os.path.join(project_root, "*.tar.gz"))
if not files_for_drop:
logger.warning("[WARNING] [TEARDOWN] Не нашлось архивов .tar.gz с данным для удаления")
else:
for file in files_for_drop:
os.remove(file)
