https://pastein.ru/t/bGT
Загрузка данных
"""
Сценарии setup/teardown СОУ через раздел Администрирование (LDS Configurator).
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Optional
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 SouAdminStatus
from constants.test_constants import LdsConfiguratorConstants as LdsCfgConst
from test_config.models_for_tests import BaseSuiteConfig
from utils.helpers import lds_configurator_utils as lds_utils
from utils.helpers.ws_message_parser import ws_message_parser as parser
logger = logging.getLogger(__name__)
def _save_group_state(group_state: Optional[Dict[str, Any]], cfg: BaseSuiteConfig, tu_id: int) -> None:
"""
Сохраняет resolved tu_id и флаги в group_state для teardown в conftest.
"""
if group_state is None:
return
group_state["use_lds_configurator"] = cfg.use_lds_configurator
group_state["resolved_tu_id"] = tu_id
group_state["admin_tu_name"] = cfg.admin_tu_name
def _save_pre_run_running_tus(
group_state: Optional[Dict[str, Any]],
snapshot: list[Dict[str, Any]],
) -> None:
"""Сохраняет список включённых ТУ стенда до запуска автотестов."""
if group_state is None:
return
group_state["pre_run_running_tus"] = snapshot
async def lds_configurator_admin_setup(
http_client: StandHttpClient,
cfg: BaseSuiteConfig,
group_state: Optional[Dict[str, Any]] = None,
) -> None:
"""
Холодный запуск СОУ через Администрирование до старта имитатора.
1. Получить tu_id по admin_tu_name из GetBasicInfoAdmin.
2. Снять слепок всех RUNNING ТУ и остановить их на стенде.
3. LaunchLdsRequest и ожидание status=включена.
4. Подтвердить launchedAt в GetTusInformation.
"""
target_tu_id: int
logger.info("[SETUP] Получение ТУ '%s' из Администрирования", cfg.admin_tu_name)
admin_reply = lds_utils.get_basic_info_admin_with_retry(http_client, parser)
target_tu = lds_utils.find_tu_by_name(admin_reply, cfg.admin_tu_name)
running_tus = lds_utils.extract_running_tus(admin_reply)
lds_utils.validate_admin_tu(target_tu)
target_tu_id = target_tu.tuId
target_tu_name = target_tu.tuName
target_tu_status = target_tu.status
cfg.resolved_tu_id = target_tu_id
_save_group_state(group_state, cfg, target_tu_id)
logger.info(
"[SETUP] Найден целевой ТУ: tuId=%s, tuName=%r, status=%s (%s)",
target_tu_id,
target_tu_name,
SouAdminStatus(target_tu.status),
SouAdminStatus.report_text_by_value(target_tu.status),
)
pre_run_snapshot = lds_utils.running_tus_to_snapshot(running_tus)
_save_pre_run_running_tus(group_state, pre_run_snapshot)
if target_tu_status == SouAdminStatus.RUNNING.value:
if len(running_tus) == 1:
_save_pre_run_running_tus(group_state, []) # Сохраняет пустой список ТУ, для запрета перезапуска
logger.info(
"[SETUP] Запущен ранее только целевой ТУ: tuId=%s, tuName=%r, status=%s (%s)",
target_tu_id,
target_tu_name,
SouAdminStatus(target_tu_status),
SouAdminStatus.report_text_by_value(target_tu_status),
)
else:
logger.info(
"[SETUP] Список включённых ТУ на стенде: %s шт. %s",
len(pre_run_snapshot),
pre_run_snapshot,
)
# Получает список запущенных ТУ, кроме целевого
extra_running_tus = [item for item in running_tus if item.tuId != target_tu_id]
await lds_utils.stop_running_tus(http_client, parser, extra_running_tus)
if not await lds_utils.poll_admin_tu_status(http_client, parser, target_tu_id, SouAdminStatus.RUNNING):
fail(
f"Не удалось оставить только целевой ТУ tuId={target_tu_id} в статусе {SouAdminStatus.RUNNING} ",
pytrace=False,
)
else:
logger.info(
"[SETUP] Список включённых ТУ на стенде: %s шт. %s",
len(pre_run_snapshot),
pre_run_snapshot,
)
await lds_utils.stop_running_tus(http_client, parser, running_tus)
logger.info("[SETUP] Холодный запуск СОУ (LaunchLdsRequest) для tuId=%s", target_tu_id)
lds_utils.run_lds_command(http_client, HttpConst.LAUNCH_LDS_URL_PATH, target_tu_id)
logger.info("[SETUP] Ожидание включения СОУ в Администрировании")
if not await lds_utils.poll_admin_tu_status(http_client, parser, target_tu_id, SouAdminStatus.RUNNING):
fail(
"Не удалось запустить целевой ТУ: статус в Администрировании не стал 'включена' за 2 минуты",
pytrace=False,
)
logger.info("[SETUP] Подтверждение наличия времени запуска (GetTusInformation)")
lds_utils.verify_get_tus_info(http_client, parser, target_tu_id)
logger.info("[LDS_CONFIGURATOR] [SETUP] [OK] Запуск СОУ через Администрирование до старта имитатора. Успех!")
async def lds_configurator_verify_after_core(
ws_client: WebSocketClient,
http_client: StandHttpClient,
cfg: BaseSuiteConfig,
) -> None:
"""
Проверка готовности стенда после запуска lds-core.
1. Актуальный статус СОУ из Администрирования.
2. Ожидание согласованного состояния ТУ на Состоянии МТ в BasicInfo из общего запаса времени.
3. Ожидание согласованного состояния ТУ на Состоянии МТ в MainPageInfoContent - остаток от общего запаса времени.
4. Сверка статуса СОУ: Администрирование vs Состояние МТ.
"""
tu_id = cfg.tu_id
tu_name = cfg.tu_name
logger.info("[SETUP] Получение актуального статуса СОУ для tuId=%s", tu_id)
admin_reply = lds_utils.get_basic_info_admin_with_retry(http_client, parser)
sou_status = lds_utils.get_admin_tu_status(admin_reply, tu_id)
if sou_status is None:
fail(
f"ТУ tuId={tu_id} ('{tu_name}') не найден в GetBasicInfoAdminResponse",
pytrace=False,
)
expect_enabled = sou_status == SouAdminStatus.RUNNING
action = "появления" if expect_enabled else "отсутствия"
loop = asyncio.get_running_loop()
if expect_enabled:
ui_sync_time_total = LdsCfgConst.VERIFY_UI_SYNC_TIME_SECONDS
admin_label = "Admin=RUNNING"
else:
ui_sync_time_total = LdsCfgConst.POLL_TIMEOUT_SECONDS
admin_label = "Admin=STOPPED"
ui_sync_time_seconds = int(ui_sync_time_total)
ui_sync_deadline = loop.time() + ui_sync_time_total
basic_timeout = ui_sync_deadline - loop.time()
logger.info(
"[SETUP] %s, запас времени %s с, ожидание %s ТУ в BasicInfo (до %s с)",
admin_label,
ui_sync_time_seconds,
action,
int(basic_timeout),
)
basic_info_poll_ok = await lds_utils.poll_basic_info_tu_presence(
http_client, parser, tu_id, tu_name, expect_present=expect_enabled, total_wait_seconds=basic_timeout
)
if not basic_info_poll_ok:
if expect_enabled:
fail(
f"СОУ не отображается на Состоянии МТ в BasicInfo: ТУ не появилась за {ui_sync_time_seconds} c "
f"после запуска core",
pytrace=False,
)
fail(
"СОУ отображается на Состоянии МТ в BasicInfo при статусе 'выключена' в Администрировании",
pytrace=False,
)
main_timeout = ui_sync_deadline - loop.time()
if main_timeout <= 0:
fail(
f"Весь запас времени {ui_sync_time_seconds} для синхронизации подписок израсходован на ожидание BasicInfo"
"на MainPage времени не осталось"
)
logger.info(
"[SETUP] Ожидание %s ТУ в Состоянии МТ (остаток запаса времени %s с)",
action,
int(main_timeout),
)
main_page_poll_ok = await lds_utils.poll_main_page_tu_presence(
ws_client, tu_id, expect_present=expect_enabled, total_wait_seconds=main_timeout
)
if not main_page_poll_ok:
if expect_enabled:
fail(
"СОУ не отображается в Состоянии МТ: ТУ не появилась за оставшееся время запаса для синхронизации"
f"({ui_sync_time_seconds} c после запуска core)",
pytrace=False,
)
fail(
"СОУ отображается в Состоянии МТ при статусе 'выключена' в Администрировании",
pytrace=False,
)
logger.info("[SETUP] Сверка статуса СОУ: Администрирование vs Состояние МТ")
lds_utils.check_sou_status_sync(sou_status, expect_enabled, expect_enabled, tu_id, tu_name)
logger.info("[LDS_CONFIGURATOR] [SETUP] [OK] Проверка готовности стенда после запуска lds-core. Успех!")
async def lds_configurator_teardown(
http_client: StandHttpClient,
tu_id: int,
admin_tu_name: str,
pre_run_running_tus: Optional[list[Dict[str, Any]]] = None,
) -> None:
"""
Teardown набора: остановка СОУ автотестов и восстановление ТУ стенда из снимка.
Некритичные отклонения логируются без падения прогона.
"""
snapshot = pre_run_running_tus or []
running_tu = {"tuId": tu_id, "tuName": admin_tu_name}
try:
logger.info("[TEARDOWN] Проверка статуса СОУ (tuId=%s, «%s»)", tu_id, admin_tu_name)
admin_reply = lds_utils.get_basic_info_admin_with_retry(http_client, parser)
sou_status = lds_utils.get_admin_tu_status(admin_reply, tu_id)
if sou_status == SouAdminStatus.RUNNING and snapshot and running_tu not in snapshot:
logger.info("[TEARDOWN] Остановка СОУ (StopLdsRequest) для tuId=%s", tu_id)
lds_utils.run_lds_command(http_client, HttpConst.STOP_LDS_URL_PATH, tu_id)
logger.info("[TEARDOWN] Ожидание выключения СОУ в Администрировании")
if not await lds_utils.poll_admin_tu_status(http_client, parser, tu_id, SouAdminStatus.STOPPED):
lds_utils.attach_allure_alert(
f"СОУ не выключилась за 2 минуты после StopLdsRequest. "
f"tuId={tu_id}, adminTuName={admin_tu_name!r}. Проверить вручную."
)
else:
lds_utils.attach_allure_alert(
f"[SKIP] tuId={tu_id}, adminTuName='{admin_tu_name} не в статусе 'включена' "
"или была включена до старта прогона. Остановка пропущена."
)
if snapshot:
await lds_utils.restore_pre_run_tus(http_client, parser, snapshot, tu_id)
logger.info("[LDS_CONFIGURATOR] [TEARDOWN] [OK] Восстановление ТУ. Успех!")
except BaseException as error:
logger.warning(
"[TEARDOWN] [ALERT] LDS Configurator teardown: %s: %r. tuId=%s, adminTuName=%r",
type(error).__name__,
error,
tu_id,
admin_tu_name,
)
lds_utils.attach_allure_alert(
f"Ошибка LDS Configurator teardown: {type(error).__name__}: {error!r}. "
f"tuId={tu_id}, adminTuName={admin_tu_name!r}"
)
lds-autotests/utils/helpers/lds_configurator_utils.py
"""
Вспомогательные функции setup/teardown СОУ через раздел Администрирование.
"""
from __future__ import annotations
import asyncio
import logging
import time
from contextlib import contextmanager
from typing import Any, Optional
import allure
from _pytest.outcomes import Failed
from pytest import fail
from clients.http_client import StandHttpClient
from clients.websocket_client import WebSocketClient
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.enums import ReplyStatus, SouAdminStatus
from constants.test_constants import LdsConfiguratorConstants as LdsCfgConst
from models.basic_info_model import BasicInfoReply, BasicTUInfo
from models.get_basic_info_admin_model import AdminTuInfo, GetBasicInfoAdminReply
from models.get_tus_information_model import GetTusInformationReply
from utils.helpers import ws_test_utils as t_utils
from utils.helpers.ws_message_parser import WsMessageParser
from utils.msgpack_utils.message_filters import is_desired_type
logger = logging.getLogger(__name__)
_configurator_flow_active: bool = False
def set_configurator_flow_active(enabled: bool) -> None:
"""Включает логирование setup/teardown при запуске СОУ через Администрирование"""
global _configurator_flow_active
_configurator_flow_active = enabled
def is_configurator_flow_active() -> bool:
"""True во время setup/teardown/verify СОУ через Администрирование"""
return _configurator_flow_active
def _fail_or_raise(message: str) -> None:
"""В configurator flow - RuntimeError для retry/teardown; иначе pytest.fail."""
if _configurator_flow_active:
raise RuntimeError(message)
fail(message, pytrace=False)
@contextmanager
def _step(name: str):
if _configurator_flow_active:
logger.info("[LDS_CONFIGURATOR] %s", name)
yield
else:
with allure.step(name):
yield
def attach_allure_alert(message: str) -> None:
"""
Публикует предупреждение в Allure и лог без падения прогона.
Используется в teardown при некритичных отклонениях.
"""
logger.warning("[LDS_CONFIGURATOR] %s", message)
if not _configurator_flow_active:
allure.attach(message, name="ALERT", attachment_type=allure.attachment_type.TEXT)
def get_basic_info(http_client: StandHttpClient, parser: WsMessageParser) -> BasicInfoReply:
"""
Выполняет getBasicInfoRequest и парсит ответ BasicInfoContent.
"""
response = http_client.post_request(HttpConst.GET_BASIC_INFO_URL_PATH, {})
payload = t_utils.get_json_from_http_response(response)
return parser.parse_basic_info_msg(payload)
def is_tu_in_basic_info(tus: Optional[list[BasicTUInfo]], tu_id: int, tu_name: str) -> bool:
"""True, если в basicInfo.tus есть запись с указанными tuId и tuName."""
return any(tu.tuId == tu_id and tu.tuName == tu_name for tu in (tus or []))
def get_basic_info_admin(
http_client: StandHttpClient,
parser: WsMessageParser,
) -> GetBasicInfoAdminReply:
"""
Выполняет GetBasicInfoAdminRequest и парсит ответ.
"""
response = http_client.post_request(HttpConst.GET_BASIC_INFO_ADMIN_URL_PATH, {})
payload = t_utils.get_json_from_http_response(response)
return parser.parse_get_basic_info_admin_msg(payload)
def get_basic_info_admin_with_retry(
http_client: StandHttpClient,
parser: WsMessageParser,
retries: int = LdsCfgConst.GET_BASIC_INFO_ADMIN_RETRIES,
) -> GetBasicInfoAdminReply:
"""
Запрашивает GetBasicInfoAdminResponse с повторными попытками.
"""
last_error: Optional[BaseException] = None
for attempt in range(1, retries + 1):
with _step(f"Запрос списка ТУ в Администрировании - попытка {attempt} из {retries}"):
try:
return get_basic_info_admin(http_client, parser)
except (
asyncio.TimeoutError,
ConnectionError,
ConnectionResetError,
OSError,
RuntimeError,
KeyError,
Failed,
) as error:
last_error = error
logger.warning(
"[LDS_CONFIGURATOR] GetBasicInfoAdmin попытка %s/%s: %s: %r",
attempt,
retries,
type(error).__name__,
error,
)
if attempt < retries:
time.sleep(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_running_tus(
http_client: StandHttpClient,
parser: WsMessageParser,
tus: list[AdminTuInfo],
) -> None:
"""Последовательно останавливает включённые ТУ на стенде."""
if not tus:
return
logger.info("[SETUP] Остановка включённых ТУ на стенде: %s шт.", len(tus))
for tu in tus:
logger.info("[SETUP] Остановка ТУ на стенде: tuId=%s, tuName=%r", tu.tuId, tu.tuName)
await stop_tu_and_wait(http_client, parser, tu.tuId, tu.tuName)
async def restore_pre_run_tus(
http_client: StandHttpClient,
parser: WsMessageParser,
snapshot: list[dict[str, Any]],
exclude_tu_id: int,
) -> None:
"""
Включает обратно ТУ из снимка, кроме ТУ автотестов.
Ошибки по отдельным ТУ оформляются как ALERT без падения прогона.
"""
to_restore = [entry for entry in snapshot if entry.get("tuId") != exclude_tu_id]
if not to_restore:
return
logger.info(
"[TEARDOWN] Восстановление ТУ стенда: %s шт. (исключена ТУ автотестов tuId=%s)",
len(to_restore),
)
for entry in to_restore:
restore_tu_id = entry["tuId"]
restore_tu_name = entry.get("tuName", "")
logger.info(
"[TEARDOWN] Запуск ТУ из снимка: tuId=%s, tuName=%r",
restore_tu_id,
restore_tu_name,
)
if not await launch_tu_and_wait(http_client, parser, restore_tu_id):
attach_allure_alert(
f"Не удалось восстановить ТУ из снимка: tuId={restore_tu_id}, "
f"tuName={restore_tu_name!r}. Проверить вручную."
)
async def poll_basic_info_tu_presence(
http_client: StandHttpClient,
parser: WsMessageParser,
tu_id: int,
tu_name: str,
expect_present: bool,
total_wait_seconds: float = LdsCfgConst.POLL_TIMEOUT_SECONDS,
poll_interval_seconds: float = LdsCfgConst.POLL_INTERVAL_SECONDS,
) -> bool:
"""
Long-poll getBasicInfoRequest: ожидание появления или исчезновения ТУ в basicInfo.tus.
"""
action = "появления" if expect_present else "исчезновения"
with _step(
f"Ожидание {action} ТУ в BasicInfo " f"(tuId={tu_id}, tuName='{tu_name}', таймаут {int(total_wait_seconds)} с)"
):
deadline = asyncio.get_running_loop().time() + total_wait_seconds
while asyncio.get_running_loop().time() < deadline:
reply = get_basic_info(http_client, parser)
tus = reply.replyContent.basicInfo.tus if reply.replyContent else None
found = is_tu_in_basic_info(tus, tu_id, tu_name)
if expect_present and found:
return True
if not expect_present and not found:
return True
await asyncio.sleep(poll_interval_seconds)
if _configurator_flow_active:
logger.error(
"[LDS_CONFIGURATOR] Таймаут ожидания %s ТУ tuId=%s в BasicInfo за %s с",
action,
tu_id,
int(total_wait_seconds),
)
return False
async def poll_main_page_tu_presence(
ws_client: WebSocketClient,
tu_id: int,
expect_present: bool,
total_wait_seconds: float = LdsCfgConst.POLL_TIMEOUT_SECONDS,
poll_interval_seconds: float = LdsCfgConst.POLL_INTERVAL_SECONDS,
) -> bool:
"""
Long-poll MainPageInfoContent: ожидание появления или исчезновения ТУ в Состоянии МТ.
"""
action = "появления" if expect_present else "отсутствия"
with _step(f"Ожидание {action} ТУ в Состоянии МТ (tuId={tu_id}, таймаут {int(total_wait_seconds)} с)"):
ws_client.clear_queue()
await t_utils.connect(
ws_client,
LdsCfgConst.SUBSCRIBE_MAIN_PAGE_INFO_REQUEST,
{"tuIds": [tu_id], "additionalProperties": None},
)
deadline = asyncio.get_running_loop().time() + total_wait_seconds
while asyncio.get_running_loop().time() < deadline:
await asyncio.sleep(poll_interval_seconds)
batch = _drain_recv_queue(ws_client)
found = any(_tu_id_in_main_page_message(msg, tu_id) for msg in batch)
if expect_present and found:
return True
if not expect_present and not found:
return True
if not _configurator_flow_active:
t_utils.attach_ws_poll_failure(
[],
total_wait_seconds,
f"{LdsCfgConst.MAIN_PAGE_INFO_CONTENT} tuId={tu_id} present={expect_present}",
)
else:
logger.warning(
"[LDS_CONFIGURATOR] Таймаут ожидания %s ТУ tuId=%s за %s с",
action,
tu_id,
int(total_wait_seconds),
)
return False
def verify_get_tus_info(http_client: StandHttpClient, parser: WsMessageParser, tu_id: int) -> None:
"""
Проверяет, наличие ТУ и launchedAt в GetTusInformation.
"""
with _step(f"Запрос GetTusInformation для tuId={tu_id}"):
response = http_client.post_request(HttpConst.GET_TUS_INFORMATION_URL_PATH, {"tuIds": [tu_id]})
payload = t_utils.get_json_from_http_response(response)
reply: GetTusInformationReply = parser.parse_get_tus_information_msg(payload)
tus_info = reply.replyContent.tusInfo if reply.replyContent else []
tu_info = next((item for item in tus_info if item.tuId == tu_id), None)
with _step("Проверка: в ответе есть информация о запуске ТУ"):
if tu_info is None:
fail(f"GetTusInformationResponse не содержит tuId={tu_id}", pytrace=False)
launched_at = parser.timestamp_to_datetime(tu_info.launchedAt)
with _step(f"Проверка (GetTusInformation): Наличие времени запуска launchedAt: {launched_at} для tuId={tu_id}"):
if launched_at is None:
fail(f"GetTusInformationResponse: launchedAt отсутствует для tuId={tu_id}", pytrace=False)
def get_admin_tu_status(admin_reply: GetBasicInfoAdminReply, tu_id: int) -> Optional[SouAdminStatus]:
"""
Возвращает статус СОУ из GetBasicInfoAdmin для указанного tuId.
"""
tus = admin_reply.replyContent.basicInfo.tus if admin_reply.replyContent else []
tu = next((item for item in tus if item.tuId == tu_id), None)
if tu is None:
return None
try:
return SouAdminStatus(tu.status)
except ValueError:
return None
conftest
import asyncio
import glob
import os
import shutil
import threading
import time
import allure
import pytest
import pytest_asyncio
from clients.testops_client import AllureResultsUploader, logger
from constants.architecture_constants import ImitatorConstants as ImConst
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.pytest_auth import (
clear_suite_auth,
ensure_auth_for_fixture,
ensure_suite_auth,
init_http_stand_client,
init_ws_stand_client,
)
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,
"suite_setup_failure": None,
"stand_host": None,
"auth_token": None,
"x_user_id": None,
"auth_suite": None,
}
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.skip("[SETUP] [ERROR] Набор пропущен: инфраструктура не готова")
def _skip_current_suite_after_setup_failure(cfg: dict, message: str) -> None:
"""
Пропускает текущий набор после ошибки setup: cleanup частичной инфраструктуры и pytest.skip.
"""
logger.error(message)
try:
allure.attach(message, name="Ошибка setup набора", attachment_type=allure.attachment_type.TEXT)
except Exception:
logger.debug("Не удалось прикрепить ошибку setup к Allure", exc_info=True)
logger.info("[TEARDOWN] LDS Configurator очистка после ошибки setup набора")
_run_lds_configurator_teardown_if_needed(cfg)
if stand_manager := cfg.get("stand_manager"):
try:
stand_manager.stop_imitator_wrapper()
except Exception:
logger.exception("[SETUP] Ошибка остановки имитатора после неудачного setup набора")
cfg["suite_infra_ready"] = False
cfg["suite_setup_failure"] = message
pytest.skip("Набор пропущен: ошибка подготовки инфраструктуры")
@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
cfg["suite_setup_failure"] = None
clear_suite_auth(cfg)
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)
_skip_current_suite_after_setup_failure(cfg, msg)
try:
stand_manager.setup_stand_for_imitator_run()
except Exception as error:
_skip_current_suite_after_setup_failure(cfg, f"[SETUP] [ERROR] ошибка при подготовке стенда: {error}")
try:
ensure_suite_auth(cfg, current_test_suite)
except BaseException as error:
_skip_current_suite_after_setup_failure(cfg, f"[SETUP] [ERROR] не удалось инициализировать auth: {error}")
try:
_update_sensor_ids(stand_manager)
except Exception as error:
_skip_current_suite_after_setup_failure(
cfg,
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:
_skip_current_suite_after_setup_failure(cfg, 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:
_skip_current_suite_after_setup_failure(cfg, 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:
_skip_current_suite_after_setup_failure(cfg, 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:
_skip_current_suite_after_setup_failure(
cfg,
f"[SETUP] [ERROR] Набор '{suite_config.suite_name}': admin_tu обязателен "
"при use_lds_configurator=True",
)
try:
_run_lds_verify_after_core(suite_config, cfg)
except BaseException as error:
_skip_current_suite_after_setup_failure(
cfg, f"[SETUP] [ERROR] LDS Configurator проверка после запуска ядра: {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(group_state)
http_client.suppress_recv_logging = True
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, group_state: dict) -> None:
"""
WS-проверка готовности стенда после запуска lds-core.
"""
async def _verify() -> None:
websocket_client = init_ws_stand_client(group_state)
http_client = init_http_stand_client(group_state)
http_client.suppress_recv_logging = True
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:
http_client = init_http_stand_client(cfg)
http_client.suppress_recv_logging = True
await lds_configurator_scenarios.lds_configurator_teardown(
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"]:
if cfg.get("suite_infra_ready"):
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() фона имитатора")
@pytest_asyncio.fixture
async def ws_client(request):
"""
Фикстура для работы с websocket клиентом.
request - даёт доступ к request.config.group_state (общий кэш auth на dataset: stand_host, auth_token, x_user_id).
:return: Объект wss соединения
"""
cfg = request.config.group_state
ensure_auth_for_fixture(cfg)
ws_client = init_ws_stand_client(cfg)
async with ws_client as client:
yield client
@pytest.fixture
def http_client(request):
"""
Фикстура для работы с http клиентом.
request - даёт доступ к request.config.group_state (общий кэш auth на dataset: stand_host, auth_token).
:return: экземпляр класса для выполнения http запросов к стенду
"""
cfg = request.config.group_state
ensure_auth_for_fixture(cfg)
return init_http_stand_client(cfg)
@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)