https://pastein.ru/t/gLa
Загрузка данных
return response
clients/http_client.py
def ping(self, correlation_id: str) -> Optional[requests.Response]:
"""
Периодический REST Ping к api-gateway (как у фронта после negotiate).
Заголовок x-tn-correlation-id позволяет gateway связать пинг с конкретным обращением.
Логирование и attach подавлены: пинг выполняется раз в 15 секунд и не должен наполнять
отчёт одинаковыми вложениями.
"""
full_url = self.generate_full_url(self._stand_url, Http_const.PING_URL_PATH)
headers = {Http_const.X_TN_CORRELATION_ID_KEY: correlation_id}
previous_suppression = self.suppress_recv_logging
self.suppress_recv_logging = True
try:
return self.make_session_request(
Http_const.POST_METHOD, full_url, data=json.dumps({}), headers=headers
)
finally:
self.suppress_recv_logging = previous_suppression
clients/persistent_ws_client.py
"""
Фасад постоянного соединения для тестов.
Сигнатуры методов совпадают с WebSocketClient, поэтому сценарии и хелперы не меняются.
Все операции выполняются в event loop фонового соединения через run_coroutine_threadsafe:
очереди принадлежат тому loop, а прямое обращение к asyncio.Queue из чужого потока
небезопасно (get_nowait будит future чужого loop).
"""
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING, Any, Callable, Coroutine, Dict, List, Optional
from constants.architecture_constants import PersistentConnectionConstants as PersistConst
from constants.architecture_constants import WebSocketClientConstants as WS_Const
if TYPE_CHECKING:
from clients.ws_connection_manager import WsConnectionManager
from clients.ws_message_dispatcher import SubscriptionScope
logger = logging.getLogger(__name__)
class PersistentWsClient:
"""
Клиент постоянного соединения, привязанный к области видимости одного теста.
Область видимости (scope) гарантирует, что в проверки теста попадают только сообщения,
пришедшие после его собственных подписок: контент подписок предыдущих тестов отсекается
по watermark, а не по invocationId (контент подписок приходит без invocationId).
"""
def __init__(self, manager: "WsConnectionManager", scope: "SubscriptionScope") -> None:
"""
:param manager: владелец постоянного соединения
:param scope: область видимости текущего теста
"""
self._manager = manager
self._scope = scope
# Флаг совместимости со сценариями: на постоянном соединении фоновый приём сообщений
# всегда логируется подавленно (иначе log-cli тонет в сотнях строк в минуту),
# поэтому переключение флага тестом на приём не влияет.
self.suppress_recv_logging: bool = False
@property
def scope_id(self) -> str:
"""Идентификатор области видимости (nodeid теста)."""
return self._scope.scope_id
@property
def invocation_id(self) -> Optional[str]:
"""invocationId последнего вызова: счётчик общий на соединение, тесты идут последовательно."""
return self._manager.last_invocation_id
@property
def connection_id(self) -> Optional[str]:
"""connectionId постоянного соединения из negotiate."""
return self._manager.connection_id
async def invoke(self, target: str, args: Any) -> None:
"""Отправляет подписку/вызов; watermark скоупа сдвигается до отправки."""
await self._call(self._manager.invoke, self._scope, target, args)
async def invoke_stream(self, target: str, args: Any) -> None:
"""Отправляет streaming-вызов (StreamInvocation)."""
await self._call(self._manager.invoke_stream, self._scope, target, args)
async def receive_by_type(self, message_type: str, timeout: float = WS_Const.FILTERING_TIMEOUT) -> List[Any]:
"""Ждёт сообщение скоупа с нужным типом контента."""
return await self._call(self._manager.dispatcher.receive_by_type, self._scope, message_type, timeout)
async def receive_by_invocation_id(
self, invocation_id: str, timeout: float = WS_Const.FILTERING_TIMEOUT
) -> List[Any]:
"""Ищет сообщение скоупа по invocationId."""
return await self._call(
self._manager.dispatcher.receive_by_invocation_id, self._scope, invocation_id, timeout
)
async def reconnect(self) -> None:
"""Инициирует переподключение постоянного соединения с восстановлением подписок."""
await self._call(self._manager.reconnect_now)
def clear_queue(self) -> None:
"""Очищает очереди скоупа. Синхронный, как у разового клиента: сценарии вызывают без await."""
self._call_sync(self._manager.dispatcher.clear, self._scope)
def drain_messages(self) -> List[Any]:
"""Вычитывает все очереди скоупа без блокирующего ожидания, сохраняя порядок сообщений."""
return self._call_sync(self._manager.dispatcher.drain, self._scope)
def queue_sizes(self) -> Dict[str, int]:
"""Размеры непустых очередей скоупа (мониторинг)."""
return self._manager.dispatcher.queue_sizes(self._scope)
async def _call(self, coroutine_factory: Callable[..., Coroutine], *args: Any) -> Any:
"""
Выполняет корутину в loop фонового соединения и ждёт результат в текущем loop.
Исключения (включая asyncio.TimeoutError) пробрасываются в поток теста как есть,
поэтому существующая логика ретраев и реконнектов в сценариях продолжает работать.
"""
future = asyncio.run_coroutine_threadsafe(coroutine_factory(*args), self._manager.loop)
return await asyncio.wrap_future(future)
def _call_sync(self, coroutine_factory: Callable[..., Coroutine], *args: Any) -> Any:
"""Синхронный вариант _call для методов, которые сценарии вызывают без await."""
future = asyncio.run_coroutine_threadsafe(coroutine_factory(*args), self._manager.loop)
return future.result(timeout=PersistConst.CALL_TIMEOUT_S)
clients/websocket_client.py
return self._connection_token
@property
def is_connected(self) -> bool:
"""True, если wss-соединение установлено и приём сообщений активен."""
if self._ws is None:
return False
return self._recv_task is not None and not self._recv_task.done()
logger.info(f"Очередь сообщений очищена: {message_empty}")
break
def drain_messages(self) -> List[Any]:
"""
Забирает все сообщения из очереди без блокирующего ожидания.
Единый API с фасадом постоянного соединения: хелперы сценариев вычитывают поток
одним способом независимо от того, разовое соединение или постоянное.
"""
messages: List[Any] = []
while not self.recv_queue.empty():
try:
messages.append(self.recv_queue.get_nowait())
except asyncio.QueueEmpty:
break
return messages
async def __aenter__(self):
await self._connect_loop()
await self.connect()
return self
async def __aexit__(self, exc_type, exc, tb):
await self.close()
async def connect(self) -> None:
"""
Устанавливает соединение: negotiate -> wss по connectionToken -> handshake -> фоновые задачи.
Публичный метод нужен владельцу постоянного соединения: он управляет жизненным циклом
соединения сам и не использует контекстный менеджер.
"""
await self._connect_loop()
async def close(self) -> None:
"""Закрывает соединение, фоновые задачи и http-сессию negotiate."""
self._log_connection_summary()
status_info = ""
pause = self._reconnect_interval
if isinstance(exc, InvalidStatus):
response = getattr(exc, "response", None)
if response is not None:
status_info = f", HTTP {response.status_code}"
if response.status_code == WS_Const.NOT_FOUND_STATUS_CODE:
# 404 означает, что балансировщик увёл handshake не на тот узел
# api-gateway, где negotiate зарегистрировал connectionToken.
# Следующая попытка снова делает negotiate и получает новый токен,
# поэтому долгая пауза только теряет время прогона.
pause = WS_Const.LB_MISS_RETRY_INTERVAL_S
logger.warning(
"WSS подключение не установлено (попытка %s%s): %s. " "Повтор через %s с, осталось %.0f с",
attempt,
status_info,
exc,
self._reconnect_interval,
pause,
remaining,
)
await asyncio.sleep(self._reconnect_interval)
await asyncio.sleep(pause)
async def _recv_loop(self) -> None:
clients/ws_connection_manager.py
"""
Владелец постоянного ws-соединения с хабом на время набора данных.
Почему соединение живёт в отдельном потоке со своим event loop: фикстура offset_wait ждёт
свой offset через time.sleep в главном потоке (до 55 минут), и loop pytest-asyncio в это
время не работает. Если бы соединение жило в loop pytest, пинги вставали бы на всё время
ожидания, сервер SignalR считал бы клиента мёртвым, а api-gateway переставал бы вычитывать
сообщения - то есть воспроизвёлся бы исходный баг с накоплением слоёв.
Менеджер повторяет поведение фронта: negotiate -> wss по connectionToken -> handshake ->
bundle подписок -> пинги ws ([6]) и rest (/apigateway/Ping) раз в 15 секунд.
Ошибки пингов и подключения прогон не роняют: они логируются и повторяются.
"""
from __future__ import annotations
import asyncio
import logging
import threading
import time
import uuid
from typing import Any, Callable, Coroutine, Dict, List, Optional, Tuple
from allure import attach, attachment_type
from clients.http_client import StandHttpClient
from clients.persistent_ws_client import PersistentWsClient
from clients.websocket_client import WebSocketClient
from clients.ws_message_dispatcher import SubscriptionScope, WsMessageDispatcher
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.architecture_constants import PersistentConnectionConstants as PersistConst
logger = logging.getLogger(__name__)
TOKEN_EXPIRED_HINT_CODES: Tuple[str, ...] = ("401", "403")
class WsConnectionManager:
"""
Постоянное соединение набора данных: фоновый loop, bundle подписок, пинги, реконнект.
"""
def __init__(self, host: str, access_token: str, x_user_id: str, bundle: List[Tuple[str, Any]]) -> None:
"""
:param host: хост api-gateway
:param access_token: JWT access token набора данных
:param x_user_id: идентификатор пользователя, полученный из /Ping
:param bundle: подписки, которые соединение держит фоном (набор фронта)
"""
self._host = host
self._access_token = access_token
self._x_user_id = x_user_id
self._bundle = bundle
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._thread: Optional[threading.Thread] = None
self._client: Optional[WebSocketClient] = None
self._http_client = StandHttpClient(host, access_token)
self._stopped = threading.Event()
self._tasks: List[asyncio.Task] = []
self._started_at: Optional[float] = None
self.dispatcher = WsMessageDispatcher()
self.last_invocation_id: Optional[str] = None
self.connections_established: int = 0
self.connect_failures: int = 0
self.rest_pings_sent: int = 0
self.rest_pings_failed: int = 0
@property
def loop(self) -> asyncio.AbstractEventLoop:
"""Event loop фонового соединения: через него проксируются вызовы из потока теста."""
if self._loop is None:
raise RuntimeError("Постоянное соединение не запущено: event loop отсутствует")
return self._loop
@property
def connection_id(self) -> Optional[str]:
"""connectionId текущего соединения из negotiate."""
return self._client.connection_id if self._client else None
def start(self) -> None:
"""
Запускает фоновый поток и постоянное соединение.
Ошибка подключения не пробрасывается: прогон продолжается, а соединение поднимает
supervisor-задача с экспоненциальным backoff.
"""
if self._thread is not None:
logger.warning("[WS_MANAGER] Соединение уже запущено, повторный start пропущен")
return
self._thread = threading.Thread(target=self._run_loop, name="ws-connection-manager", daemon=True)
self._thread.start()
self._wait_for_loop()
self._started_at = time.monotonic()
self._call_sync(self._start_async, timeout=PersistConst.START_TIMEOUT_S)
# Ждём фактического подключения: тесты стартуют сразу после setup набора, а соединение
# поднимается асинхронно (negotiate + несколько попыток wss). Без ожидания первый тест
# попал бы на ещё не готовое соединение.
is_connected = self._call_sync(self._wait_for_connection, PersistConst.START_TIMEOUT_S)
if is_connected:
logger.info(f"[WS_MANAGER] Постоянное соединение запущено для хоста {self._host}")
else:
logger.warning(
"[WS_MANAGER] [ALERT] Постоянное соединение не поднялось за отведённое время, "
"supervisor продолжит попытки в фоне"
)
def stop(self) -> None:
"""Останавливает задачи, соединение и поток; публикует итоги в лог и Allure."""
if self._thread is None:
return
try:
self._call_sync(self._stop_async, timeout=PersistConst.STOP_TIMEOUT_S)
except Exception as error:
logger.warning(f"[WS_MANAGER] Ошибка при остановке соединения: {error}")
finally:
self._stopped.set()
if self._loop is not None:
self._loop.call_soon_threadsafe(self._loop.stop)
if self._thread is not None:
self._thread.join(timeout=PersistConst.STOP_TIMEOUT_S)
self._log_health()
self._attach_health_summary("Итоги постоянного ws-соединения")
self._http_client.session.close()
self._thread = None
def _wait_for_loop(self) -> None:
"""Ждёт создания event loop в фоновом потоке."""
deadline = time.monotonic() + PersistConst.LOOP_START_TIMEOUT_S
while self._loop is None and time.monotonic() < deadline:
time.sleep(0.05)
if self._loop is None:
raise RuntimeError("Не удалось запустить event loop постоянного соединения")
async def _start_async(self) -> None:
"""Создаёт клиента и фоновые задачи: маршрутизация, REST Ping, мониторинг, supervisor."""
self._client = WebSocketClient(self._host, self._access_token, self._x_user_id)
# Приём сообщений на постоянном соединении непрерывный: без подавления log-cli
# утонет в сотнях строк в минуту. Содержимое видно в attach парсера внутри теста.
self._client.suppress_recv_logging = True
self._client.set_disconnected_handler(self._on_disconnected)
self._tasks = [
asyncio.create_task(self._dispatch_loop(), name="ws-dispatch"),
asyncio.create_task(self._rest_ping_loop(), name="ws-rest-ping"),
asyncio.create_task(self._health_log_loop(), name="ws-health"),
asyncio.create_task(self._supervisor_loop(), name="ws-supervisor"),
]
def _on_disconnected(self) -> None:
"""
Реакция на разрыв соединения из приёмного или пингового цикла.
Реконнект здесь не запускается намеренно: этим занимается supervisor-задача,
иначе при обрыве обоих циклов возникли бы два параллельных переподключения.
"""
logger.warning("[WS_MANAGER] Соединение разорвано, реконнект выполнит supervisor")
async def _supervisor_loop(self) -> None:
"""
Следит за соединением и поднимает его заново.
Соединение должно жить всё время набора, включая паузы между тестами и offset-ожидания:
именно живое соединение с пингами держит консьюмеры api-gateway, которые вычитывают
сообщения core и layer builder.
"""
while not self._stopped.is_set():
if not self._is_connected():
await self._connect_with_backoff()
await asyncio.sleep(PersistConst.CONNECTION_CHECK_INTERVAL_S)
def _is_connected(self) -> bool:
"""True, если клиент создан и его соединение активно."""
return bool(self._client and self._client.is_connected)
async def _wait_for_connection(self, timeout: float) -> bool:
"""Ждёт появления активного соединения; возвращает False по истечении таймаута."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline and not self._stopped.is_set():
if self._is_connected():
return True
await asyncio.sleep(PersistConst.RECEIVE_POLL_INTERVAL_S)
return self._is_connected()
async def _connect_with_backoff(self) -> None:
"""
Устанавливает соединение и восстанавливает подписки, при ошибке повторяет.
Прогон не прерывается: ошибки логируются, backoff растёт до RECONNECT_MAX_BACKOFF_S.
"""
backoff = PersistConst.RECONNECT_BACKOFF_S
while not self._stopped.is_set():
try:
await self._client.connect()
self.connections_established += 1
logger.info(
f"[WS_MANAGER] [OK] Постоянное соединение установлено "
f"(connectionId={self._client.connection_id})"
)
await self._subscribe_bundle()
await self._restore_scope_subscriptions()
return
except Exception as error:
self.connect_failures += 1
logger.warning(f"[WS_MANAGER] Не удалось установить соединение: {error}. Повтор через {backoff:.0f} с")
if any(code in str(error) for code in TOKEN_EXPIRED_HINT_CODES):
logger.warning("[WS_MANAGER] Похоже на истёкший JWT: требуется новый access token")
await asyncio.sleep(backoff)
backoff = min(backoff * 2, PersistConst.RECONNECT_MAX_BACKOFF_S)
async def _subscribe_bundle(self) -> None:
"""Отправляет bundle подписок фронта: они и держат консьюмеры gateway активными."""
for target, args in self._bundle:
try:
await self._client.invoke(target, args)
logger.info(f"[WS_MANAGER] Bundle-подписка отправлена: {target}")
except Exception as error:
logger.warning(f"[WS_MANAGER] Не удалось отправить bundle-подписку {target}: {error}")
async def _restore_scope_subscriptions(self) -> None:
"""После реконнекта возвращает подписки активных тестов (bundle отправлен отдельно)."""
for target, args in self.dispatcher.active_subscriptions():
try:
await self._client.invoke(target, args)
logger.info(f"[WS_MANAGER] Подписка теста восстановлена: {target}")
except Exception as error:
logger.warning(f"[WS_MANAGER] Не удалось восстановить подписку {target}: {error}")
async def _dispatch_loop(self) -> None:
"""Перекладывает сообщения из очереди клиента в диспетчер областей видимости."""
while not self._stopped.is_set():
client = self._client
if client is None:
await asyncio.sleep(PersistConst.DISPATCH_IDLE_TIMEOUT_S)
continue
try:
message = await asyncio.wait_for(
client.recv_queue.get(), timeout=PersistConst.DISPATCH_IDLE_TIMEOUT_S
)
except asyncio.TimeoutError:
continue
except Exception as error:
logger.warning(f"[WS_MANAGER] Ошибка приёма сообщения из очереди клиента: {error}")
await asyncio.sleep(PersistConst.DISPATCH_IDLE_TIMEOUT_S)
continue
try:
self.dispatcher.route(message)
except Exception as error:
logger.warning(f"[WS_MANAGER] Ошибка маршрутизации сообщения: {error}")
async def _rest_ping_loop(self) -> None:
"""Шлёт REST Ping раз в 15 с: так же делает фронт, и gateway видит живого клиента."""
while not self._stopped.is_set():
await asyncio.sleep(HttpConst.REST_PING_INTERVAL_S)
if self._stopped.is_set():
return
try:
await asyncio.to_thread(self._send_rest_ping)
self.rest_pings_sent += 1
except Exception as error:
# Прогон не роняем: пинг повторится на следующей итерации
self.rest_pings_failed += 1
logger.warning(f"[WS_MANAGER] REST Ping не выполнен: {error}")
def _send_rest_ping(self) -> None:
"""Выполняет POST /apigateway/Ping с x-tn-correlation-id (requests, в отдельном потоке)."""
self._http_client.ping(str(uuid.uuid4()))
async def _health_log_loop(self) -> None:
"""Периодически логирует состояние соединения: пинги, счётчики, очереди."""
while not self._stopped.is_set():
await asyncio.sleep(PersistConst.HEALTH_LOG_INTERVAL_S)
if self._stopped.is_set():
return
self._log_health()
def health_snapshot(self) -> Dict[str, Any]:
"""Сводка состояния соединения для логов и Allure."""
client = self._client
uptime_seconds = time.monotonic() - self._started_at if self._started_at else 0.0
return {
"connectionId": self.connection_id,
"isConnected": self._is_connected(),
"uptimeSeconds": round(uptime_seconds),
"connectionsEstablished": self.connections_established,
"connectFailures": self.connect_failures,
"wsPingsSent": client.ws_pings_sent if client else 0,
"wsPingsFailed": client.ws_pings_failed if client else 0,
"restPingsSent": self.rest_pings_sent,
"restPingsFailed": self.rest_pings_failed,
"messagesReceived": client.messages_received if client else 0,
"messagesRouted": self.dispatcher.messages_routed,
"messagesDropped": self.dispatcher.messages_dropped,
"clientReconnects": client.reconnect_count if client else 0,
}
def _log_health(self) -> None:
"""Выводит сводку соединения в лог: рост счётчиков означает, что клиент подписан и жив."""
logger.info(f"[WS_MANAGER] [WS_HEALTH] {self.health_snapshot()}")
def _attach_health_summary(self, name: str) -> None:
"""Публикует итоги соединения в Allure."""
try:
body = "\n".join(f"{key}: {value}" for key, value in self.health_snapshot().items())
attach(body, name=name, attachment_type=attachment_type.TEXT)
except (KeyError, RuntimeError) as error:
logger.debug(f"Allure attach итогов соединения пропущен: {error}")
def open_scope(self, scope_id: str) -> SubscriptionScope:
"""Открывает область видимости теста на постоянном соединении."""
return self._call_sync(self._open_scope_async, scope_id)
def close_scope(self, scope: SubscriptionScope) -> None:
"""Закрывает область видимости теста и уничтожает его очереди."""
self._call_sync(self._close_scope_async, scope.scope_id)
def client_for(self, scope: SubscriptionScope) -> PersistentWsClient:
"""Фасад соединения для конкретного теста."""
return PersistentWsClient(self, scope)
async def _open_scope_async(self, scope_id: str) -> SubscriptionScope:
"""Создаёт область видимости внутри loop соединения."""
return self.dispatcher.open_scope(scope_id)
async def _close_scope_async(self, scope_id: str) -> None:
"""Закрывает область видимости внутри loop соединения."""
self.dispatcher.close_scope(scope_id)
async def invoke(self, scope: SubscriptionScope, target: str, args: Any) -> None:
"""
Отправляет подписку теста.
watermark сдвигается до отправки: контент, прилетевший от подписок прошлых тестов,
в проверки текущего теста попасть не может. Подписка запоминается, чтобы восстановить
её после реконнекта.
"""
await self._ensure_connected_for_call()
self.dispatcher.register_subscription(scope, target, args)
self.dispatcher.advance_watermark(scope)
await self._client.invoke(target, args)
self.last_invocation_id = self._client.invocation_id
async def invoke_stream(self, scope: SubscriptionScope, target: str, args: Any) -> None:
"""Отправляет streaming-вызов; в реестр не добавляется, потому что он одноразовый."""
await self._ensure_connected_for_call()
self.dispatcher.advance_watermark(scope)
await self._client.invoke_stream(target, args)
self.last_invocation_id = self._client.invocation_id
async def reconnect_now(self) -> None:
"""Принудительное переподключение с восстановлением bundle и подписок тестов."""
self._ensure_client()
await self._client.reconnect()
await self._subscribe_bundle()
await self._restore_scope_subscriptions()
def _ensure_client(self) -> None:
"""Проверяет, что клиент соединения создан."""
if self._client is None:
raise ConnectionError("Постоянное соединение не инициализировано")
async def _ensure_connected_for_call(self) -> None:
"""
Ждёт восстановления соединения перед вызовом теста.
Ожидание нужно, потому что supervisor поднимает соединение асинхронно: без него обрыв
соединения в момент подписки ронял бы тест вместо прозрачного восстановления.
Если соединение так и не появилось, ошибка содержит диагностику вместо обезличенного
таймаута ожидания сообщения.
"""
if self._is_connected():
return
self._ensure_client()
if await self._wait_for_connection(PersistConst.CALL_CONNECTION_WAIT_S):
return
snapshot = self.health_snapshot()
raise ConnectionError(
f"Постоянное соединение не активно: isConnected={snapshot['isConnected']}, "
f"connectionsEstablished={snapshot['connectionsEstablished']}, "
f"connectFailures={snapshot['connectFailures']}, wsPingsSent={snapshot['wsPingsSent']}"
)
async def _stop_async(self) -> None:
"""Останавливает фоновые задачи и закрывает соединение."""
for task in self._tasks:
task.cancel()
for task in self._tasks:
try:
await task
except asyncio.CancelledError:
pass
except Exception as error:
logger.warning(f"[WS_MANAGER] Фоновая задача завершилась с ошибкой: {error}")
self._tasks = []
if self._client is not None:
await self._client.close()
def _run_loop(self) -> None:
"""Точка входа фонового потока: собственный event loop на всё время набора."""
loop = asyncio.new_event_loop()
self._loop = loop
asyncio.set_event_loop(loop)
try:
loop.run_forever()
finally:
try:
loop.run_until_complete(loop.shutdown_asyncgens())
except Exception as error:
logger.debug(f"[WS_MANAGER] shutdown_asyncgens завершился с ошибкой: {error}")
loop.close()
logger.info("[WS_MANAGER] Event loop постоянного соединения закрыт")
def _call_sync(
self,
coroutine_factory: Callable[..., Coroutine],
*args: Any,
timeout: float = PersistConst.CALL_TIMEOUT_S,
) -> Any:
"""Выполняет корутину в loop соединения из основного потока и ждёт результат."""
future = asyncio.run_coroutine_threadsafe(coroutine_factory(*args), self.loop)
return future.result(timeout=timeout)
clients/ws_message_dispatcher.py
"""
Маршрутизация входящих ws-сообщений по областям видимости тестов (subscription scope).
Причина появления модуля: ответ на подписку приходит с invocationId (replyStatus 200),
а контент подписки - без invocationId, только с типом контента. На постоянном соединении
подписки прошлых тестов продолжают присылать контент, поэтому сообщения нужно разделять
не по invocationId, а по принадлежности тесту и моменту подписки.
Механизм:
- каждое входящее сообщение получает монотонный номер seq;
- у области видимости теста есть watermark - номер seq на момент последней подписки;
- сообщения с seq <= watermark скоупу не отдаются: контент, прилетевший до подписки
текущего теста (в том числе от подписок предыдущих тестов), в проверки попасть не может;
- после закрытия скоупа его очереди уничтожаются.
"""
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Tuple
from constants.architecture_constants import PersistentConnectionConstants as PersistConst
from constants.architecture_constants import WebSocketClientConstants as WS_Const
from utils.msgpack_utils.message_filters import is_desired_invocation_id, is_desired_type
logger = logging.getLogger(__name__)
UNKNOWN_MESSAGE_QUEUE_PREFIX: str = "unknown"
SIGNALR_MESSAGE_QUEUE_PREFIX: str = "signalr"
SeqMessage = Tuple[int, Any]
def get_message_queue_key(message: Any) -> str:
"""
Возвращает ключ очереди для сообщения.
Invocation (тип 1) несёт тип контента в msg[EVENT_TYPE_INDEX] - именно по нему сценарии
ищут сообщения. StreamItem и Completion типа контента не имеют, поэтому раскладываются
по служебным очередям с номером типа SignalR: так streaming-ответы не смешиваются
с контентом подписок и не теряются при вычитывании по типу.
"""
if not isinstance(message, list) or not message:
return f"{UNKNOWN_MESSAGE_QUEUE_PREFIX}:{type(message).__name__}"
message_type = message[WS_Const.SIGNALR_MESSAGE_TYPE_INDEX]
is_invocation = message_type == WS_Const.DEFAULT_SIGNALR_MESSAGE_TYPE
if is_invocation and len(message) > WS_Const.EVENT_TYPE_INDEX:
content_type = message[WS_Const.EVENT_TYPE_INDEX]
if isinstance(content_type, str):
return content_type
return f"{SIGNALR_MESSAGE_QUEUE_PREFIX}:{message_type}"
@dataclass
class SubscriptionScope:
"""
Область видимости одного теста на постоянном соединении.
watermark сдвигается на каждую подписку теста: всё, что пришло раньше, тесту не видно.
subscriptions хранятся, чтобы после реконнекта отправить подписки текущего теста заново.
"""
scope_id: str
start_seq: int
watermark: int = 0
queues: Dict[str, asyncio.Queue] = field(default_factory=dict)
subscriptions: List[Tuple[str, Any]] = field(default_factory=list)
closed: bool = False
def __post_init__(self) -> None:
self.watermark = self.start_seq
class WsMessageDispatcher:
"""
Раскладывает входящие сообщения по очередям областей видимости.
Все методы вызываются внутри event loop фонового соединения: очереди принадлежат ему,
а обращение к asyncio.Queue из чужого потока небезопасно.
"""
def __init__(self) -> None:
self._seq: int = 0
self._scopes: Dict[str, SubscriptionScope] = {}
self.messages_routed: int = 0
self.messages_dropped: int = 0
@property
def current_seq(self) -> int:
"""Текущий номер seq: используется как watermark в момент подписки."""
return self._seq
def open_scope(self, scope_id: str) -> SubscriptionScope:
"""Создаёт область видимости теста; повторный вызов возвращает существующую."""
existing = self._scopes.get(scope_id)
if existing is not None:
return existing
scope = SubscriptionScope(scope_id=scope_id, start_seq=self._seq)
self._scopes[scope_id] = scope
logger.info(f"[DISPATCH] Открыт scope {scope_id} (start_seq={self._seq})")
return scope
def close_scope(self, scope_id: str) -> None:
"""Закрывает область видимости и уничтожает её очереди."""
scope = self._scopes.pop(scope_id, None)
if scope is None:
return
scope.closed = True
scope.queues.clear()
logger.info(f"[DISPATCH] Закрыт scope {scope_id}")
def advance_watermark(self, scope: SubscriptionScope) -> None:
"""
Сдвигает watermark на текущий seq.
Вызывается до отправки подписки: сообщения, уже лежащие в потоке от прошлых подписок,
становятся невидимыми, а ответ и контент новой подписки - видимыми.
"""
scope.watermark = self._seq
def register_subscription(self, scope: SubscriptionScope, target: str, args: Any) -> None:
"""Запоминает подписку скоупа для повторной отправки после реконнекта."""
scope.subscriptions.append((target, args))
def active_subscriptions(self) -> List[Tuple[str, Any]]:
"""Подписки всех активных скоупов (для восстановления после реконнекта)."""
subscriptions: List[Tuple[str, Any]] = []
for scope in self._scopes.values():
if not scope.closed:
subscriptions.extend(scope.subscriptions)
return subscriptions
def route(self, message: Any) -> None:
"""Раскладывает сообщение по очередям активных скоупов с учётом watermark."""
self._seq += 1
seq = self._seq
key = get_message_queue_key(message)
is_routed = False
for scope in list(self._scopes.values()):
if scope.closed or seq <= scope.watermark:
continue
self._put_message(scope, key, seq, message)
is_routed = True
if is_routed:
self.messages_routed += 1
else:
# Скоупов нет (пауза между тестами) либо сообщения старше watermark - не копим
self.messages_dropped += 1
def _put_message(self, scope: SubscriptionScope, key: str, seq: int, message: Any) -> None:
"""Кладёт сообщение в очередь скоупа, вытесняя самое старое при переполнении."""
queue = scope.queues.setdefault(key, asyncio.Queue())
if queue.qsize() >= PersistConst.SCOPE_QUEUE_MAXSIZE:
# Тест не вычитывает этот тип: вытесняем старое сообщение, чтобы не расти по памяти
try:
queue.get_nowait()
except asyncio.QueueEmpty:
logger.debug(f"[DISPATCH] Очередь {key} оказалась пустой при вытеснении")
queue.put_nowait((seq, message))
async def receive_by_type(self, scope: SubscriptionScope, content_type: str, timeout: float) -> List[Any]:
"""Ждёт сообщение нужного типа контента в очереди скоупа."""
queue = scope.queues.setdefault(content_type, asyncio.Queue())
deadline = time.monotonic() + timeout
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise asyncio.TimeoutError(f"Timeout при фильтрации сообщений {timeout:.1f} секунд")
_, message = await asyncio.wait_for(queue.get(), timeout=remaining)
if isinstance(message, list) and is_desired_type(message, content_type):
return message
async def receive_by_invocation_id(self, scope: SubscriptionScope, invocation_id: str, timeout: float) -> List[Any]:
"""
Ищет сообщение с нужным invocationId среди всех очередей скоупа.
Ненайденные сообщения возвращаются в свои очереди, поэтому поиск не теряет данные
(в отличие от чтения из общей очереди, где чужие сообщения отбрасывались).
"""
deadline = time.monotonic() + timeout
while True:
drained = self._drain_with_seq(scope)
for _, message in drained:
if isinstance(message, list) and is_desired_invocation_id(message, invocation_id):
self._requeue(scope, [(seq, item) for seq, item in drained if item is not message])
return message
remaining = deadline - time.monotonic()
if remaining <= 0:
raise asyncio.TimeoutError(f"Timeout при фильтрации сообщений {timeout:.1f} секунд")
await asyncio.sleep(PersistConst.RECEIVE_POLL_INTERVAL_S)
async def drain(self, scope: SubscriptionScope) -> List[Any]:
"""Вычитывает все очереди скоупа, сохраняя порядок поступления (сортировка по seq)."""
return [message for _, message in self._drain_with_seq(scope)]
async def clear(self, scope: SubscriptionScope) -> None:
"""Очищает все очереди скоупа (аналог clear_queue разового клиента)."""
for queue in scope.queues.values():
while not queue.empty():
try:
queue.get_nowait()
except asyncio.QueueEmpty:
break
def queue_sizes(self, scope: SubscriptionScope) -> Dict[str, int]:
"""Размеры непустых очередей скоупа для мониторинга."""
return {key: queue.qsize() for key, queue in scope.queues.items() if queue.qsize()}
def _drain_with_seq(self, scope: SubscriptionScope) -> List[SeqMessage]:
"""Вычитывает все очереди скоупа и возвращает пары (seq, сообщение) по порядку."""
drained: List[SeqMessage] = []
for queue in scope.queues.values():
while not queue.empty():
try:
drained.append(queue.get_nowait())
except asyncio.QueueEmpty:
break
drained.sort(key=lambda item: item[0])
return drained
def _requeue(self, scope: SubscriptionScope, items: List[SeqMessage]) -> None:
"""Возвращает сообщения обратно в очереди скоупа с сохранением их seq."""
for seq, message in items:
self._put_message(scope, get_message_queue_key(message), seq, message)
арх конст
SIGNALR_PING_INTERVAL_S: float = 15.0
# 404 на wss-handshake означает, что балансировщик увёл подключение не на тот узел
# api-gateway, где negotiate зарегистрировал connectionToken. Пауза короче обычной,
# потому что повторный negotiate всё равно выдаёт новый токен.
NOT_FOUND_STATUS_CODE: int = 404
LB_MISS_RETRY_INTERVAL_S: float = 1.0
FILTERING_TIMEOUT: int | float = 30.0
class PersistentConnectionConstants:
"""
Константы постоянного (фонового) соединения с хабом.
Постоянное соединение повторяет поведение фронта: после negotiate держит bundle подписок
и шлёт пинги (ws и rest раз в 15 с). Благодаря этому api-gateway непрерывно вычитывает
сообщения core и layer builder, а не копит их до подключения очередного теста.
"""
# ===== Bundle подписок, который фронт отправляет сразу после handshake =====
SUBSCRIBE_PLAYER_STATUS_INFO_REQUEST: str = "SubscribePlayerStatusInfoRequest"
SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST: str = "SubscribeReportsDataExportedRequest"
SUBSCRIBE_SELF_DIAGNOSTIC_INFO_REQUEST: str = "SubscribeSelfDiagnosticInfoRequest"
SUBSCRIBE_MAIN_PAGE_INFO_REQUEST: str = "subscribeMainPageInfoRequest"
SUBSCRIBE_MAIN_PAGE_SIGNALS_INFO_REQUEST: str = "subscribeMainPageSignalsInfoRequest"
EMPTY_SUBSCRIBE_ARGS: list = []
TU_IDS_KEY: str = "tuIds"
ADDITIONAL_PROPERTIES_KEY: str = "additionalProperties"
# ===== Тайминги и лимиты фонового соединения =====
LOOP_START_TIMEOUT_S: float = 10.0
START_TIMEOUT_S: float = 180.0
STOP_TIMEOUT_S: float = 30.0
CALL_TIMEOUT_S: float = 300.0
RECONNECT_BACKOFF_S: float = 5.0
RECONNECT_MAX_BACKOFF_S: float = 60.0
CONNECTION_CHECK_INTERVAL_S: float = 5.0
CALL_CONNECTION_WAIT_S: float = 60.0
HEALTH_LOG_INTERVAL_S: float = 60.0
DISPATCH_IDLE_TIMEOUT_S: float = 1.0
RECEIVE_POLL_INTERVAL_S: float = 0.2
SCOPE_QUEUE_MAXSIZE: int = 500
utils/helpers/lds_configurator_utils.py
def _drain_recv_queue(ws_client: WebSocketClient) -> list[Any]:
"""Забирает все сообщения из очереди WS без блокирующего ожидания."""
return ws_client.drain_messages()
utils/helpers/pytest_auth.py
from clients.keycloak_clients import KeycloakAuthError, KeycloakClient
from clients.testops_client import logger
from clients.websocket_client import WebSocketClient
from clients.ws_connection_manager import WsConnectionManager
from constants.architecture_constants import EnvKeyConstants as EnvConst
from constants.architecture_constants import HTTPClientConstants as HttpConst
from constants.architecture_constants import PersistentConnectionConstants as PersistConst
from constants.architecture_constants import WebSocketClientConstants as WSCliConst
# Один KeycloakClient на pytest-сессию; TTL токена проверяется внутри клиента.
_keycloak_client: KeycloakClient | None = None
# Ключ постоянного ws-соединения в group_state (жизненный цикл - набор данных).
WS_MANAGER_STATE_KEY: str = "ws_manager"
def get_instance() -> str:
group_state["auth_token"],
group_state["x_user_id"],
)
def build_subscription_bundle(tu_id: int) -> list[tuple[str, Any]]:
"""
Bundle подписок постоянного соединения - повторяет набор фронта после авторизации.
Эти подписки вместе с negotiate дают api-gateway понять, что есть подписанный клиент,
и держать консьюмеры для чтения core и layer builder на всё время набора данных.
"""
main_page_args = {
PersistConst.TU_IDS_KEY: [tu_id],
PersistConst.ADDITIONAL_PROPERTIES_KEY: None,
}
return [
(PersistConst.SUBSCRIBE_PLAYER_STATUS_INFO_REQUEST, PersistConst.EMPTY_SUBSCRIBE_ARGS),
(PersistConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST, PersistConst.EMPTY_SUBSCRIBE_ARGS),
(PersistConst.SUBSCRIBE_SELF_DIAGNOSTIC_INFO_REQUEST, PersistConst.EMPTY_SUBSCRIBE_ARGS),
(PersistConst.SUBSCRIBE_MAIN_PAGE_INFO_REQUEST, dict(main_page_args)),
(PersistConst.SUBSCRIBE_MAIN_PAGE_SIGNALS_INFO_REQUEST, dict(main_page_args)),
]
def start_ws_connection_manager(group_state: dict, tu_id: int) -> WsConnectionManager:
"""
Создаёт и запускает постоянное соединение набора данных (один раз на dataset).
Ошибка запуска не пробрасывается в pytest: прогон должен продолжаться, а соединение
поднимет supervisor внутри менеджера с экспоненциальным backoff.
"""
_require_auth(group_state, require_x_user_id=True)
existing = group_state.get(WS_MANAGER_STATE_KEY)
if existing is not None:
return existing
manager = WsConnectionManager(
host=group_state["stand_host"],
access_token=group_state["auth_token"],
x_user_id=group_state["x_user_id"],
bundle=build_subscription_bundle(tu_id),
)
group_state[WS_MANAGER_STATE_KEY] = manager
try:
manager.start()
logger.info("[WS_MANAGER] Постоянное соединение набора запущено (tuId=%s)", tu_id)
except BaseException as error:
logger.warning("[WS_MANAGER] [ALERT] Не удалось запустить постоянное соединение: %s", error)
return manager
def get_ws_connection_manager(group_state: dict) -> Optional[WsConnectionManager]:
"""Возвращает запущенное постоянное соединение набора либо None."""
return group_state.get(WS_MANAGER_STATE_KEY)
def close_ws_connection_manager(group_state: dict) -> None:
"""Останавливает постоянное соединение набора. Идемпотентно: повторный вызов безопасен."""
manager = group_state.pop(WS_MANAGER_STATE_KEY, None)
if manager is None:
return
try:
manager.stop()
logger.info("[WS_MANAGER] Постоянное соединение набора остановлено")
except BaseException as error:
logger.warning("[WS_MANAGER] [ALERT] Ошибка при остановке постоянного соединения: %s", error)
utils/helpers/ws_test_utils.py
def _drain_recv_queue(ws_client: WebSocketClient) -> List[Any]:
"""
Забирает все сообщения из очереди ws без блокирующего receive_by_...
Вычитывание делегируется клиенту: у постоянного соединения очереди принадлежат event loop
фонового потока, и прямое обращение к asyncio.Queue из потока теста небезопасно.
"""
return ws_client.drain_messages()
тут тоже с удалением
latest_msg = None
for msg in ws_client.drain_messages():
if isinstance(msg, list) and is_desired_type(msg, "BalanceAlgorithmResultsContent"):
latest_msg = msg
конфтест
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 EnvKeyConstants
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,
close_ws_connection_manager,
ensure_auth_for_fixture,
ensure_suite_auth,
get_ws_connection_manager,
init_http_stand_client,
init_ws_stand_client,
start_ws_connection_manager,
)
from utils.helpers.pytest_skip_utils import (
decorate_skipped_test_in_allure,
get_skip_reason_marker,
is_marked_as_skipped,
resolve_skip_reason,
)
from utils.helpers.vault_pytest_utils import configure_vault_for_suite, reset_vault_process_empty_values_rejection
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,
"ws_manager": None, # постоянное ws-соединение набора данных (WsConnectionManager)
"imitator_start_time": None, # datetime объект времени старта имитатора для расчёта интервалов утечек
"is_target_tu_active": False,
"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,
"vault_rejection_enabled": False,
}
def _update_configuration_data(stand_manager: StandSetupManager) -> None:
"""
Обновляет данные для тестов из файла конфигурации стенда.
"""
sensor_ids_by_address, segments = stand_manager.get_data_from_configuration()
BaseTN3Constants.SENSOR_IDS_BY_ADDRESS.update(sensor_ids_by_address)
BaseTN3Constants.CONTROLLED_SITE_SEGMENTS.update(segments)
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_stationary_status_common_scheme': 'stationary_status_common_scheme_test',
'test_stationary_status_main_page_info': 'stationary_status_main_page_info_test',
'test_stationary_status_in_output_signals': 'stationary_status_in_output_signals_test',
'test_stationary_status_journal': 'stationary_status_journal_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_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_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_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_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
# ===== ТЕСТЫ бики ДУ2. ЭФ Журнал =====
'test_degradation_temperature_du_2_in_journal': 'degradation_temperature_du_2_in_journal_test',
'test_degradation_density_du_2_in_journal': 'degradation_density_du_2_in_journal_test',
'test_degradation_viscosity_du_2_in_journal': 'degradation_viscosity_du_2_in_journal_test',
# ===== ТЕСТЫ бики ДУ3. ЭФ Журнал =====
'test_degradation_temperature_du_3_in_journal': 'degradation_temperature_du_3_in_journal_test',
'test_degradation_density_du_3_in_journal': 'degradation_density_du_3_in_journal_test',
'test_degradation_viscosity_du_3_in_journal': 'degradation_viscosity_du_3_in_journal_test',
# ===== ТЕСТЫ течение. ЭФ Журнал =====
'test_lds_status_degradation_gravity_section_pumping_in_stopping_in_journal': 'deg_gravity_section_pumping_in_stopping_in_journal_test', # noqa: E501
'test_between_si_pressure_more_50_km_in_journal': 'between_si_pressure_more_50_km_in_journal_test',
'test_deg_gravity_section_pumping_in_journal': 'deg_gravity_section_pumping_in_journal_test',
'test_deg_absence_min_pressure_sensors_in_journal': 'deg_absence_min_pressure_sensors_in_journal_test',
'test_deg_starting_pumping_out_pumps_in_journal': 'deg_starting_pumping_out_pumps_in_journal_test',
'test_faulty_absence_min_flow_meters_in_journal': 'faulty_absence_min_flow_meters_in_journal_test',
'test_faulty_absence_min_flow_meters_continuous_in_journal': 'faulty_absence_min_flow_meters_continuous_in_journal_test', # noqa: E501
'test_deg_exceeding_distance_between_flow_meters_in_journal': 'deg_exceeding_distance_between_flow_meters_in_journal_test', # noqa: E501
'test_deg_pig_sensor_passage_in_journal': 'deg_pig_sensor_passage_in_journal_test',
'test_deg_additive_injectors_operation_in_journal': 'deg_additive_injectors_operation_in_journal_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_common_scheme_cold_start': 'common_scheme_cold_start_test',
'test_journal_cold_start': 'journal_cold_start_test',
'test_main_page_info_cold_start': 'main_page_info_cold_start_test',
'test_output_signals_cold_start': 'output_signals_cold_start_test',
'test_common_scheme_stationary_after_cold': 'common_scheme_stationary_after_cold_test',
'test_journal_stationary_after_cold': 'journal_stationary_after_cold_test',
'test_main_page_info_stationary_after_cold': 'main_page_info_stationary_after_cold_test',
'test_output_signals_stationary_after_cold': 'output_signals_stationary_after_cold_test',
'test_common_scheme_switch_on': 'common_scheme_switch_on_test',
'test_journal_switch_on': 'journal_switch_on_test',
'test_main_page_info_switch_on': 'main_page_info_switch_on_test',
'test_output_signals_switch_on': 'output_signals_switch_on_test',
'test_common_scheme_stationary_after_switch_on': 'common_scheme_stationary_after_switch_on_test',
'test_journal_stationary_after_switch_on': 'journal_stationary_after_switch_on_test',
'test_main_page_info_stationary_after_switch_on': 'main_page_info_stationary_after_switch_on_test',
'test_output_signals_stationary_after_switch_on': 'output_signals_stationary_after_switch_on_test',
'test_common_scheme_switch_off': 'common_scheme_switch_off_test',
'test_journal_switch_off': 'journal_switch_off_test',
'test_main_page_info_switch_off': 'main_page_info_switch_off_test',
'test_output_signals_switch_off': 'output_signals_switch_off_test',
'test_common_scheme_stationary_after_switch_off': 'common_scheme_stationary_after_switch_off_test',
'test_journal_stationary_after_switch_off': 'journal_stationary_after_switch_off_test',
'test_main_page_info_stationary_after_switch_off': 'main_page_info_stationary_after_switch_off_test',
'test_output_signals_stationary_after_switch_off': 'output_signals_stationary_after_switch_off_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 тесты имитации выходных сигналов (маркеры из OutputSignalImitationConfig - параметр config)
OUTPUT_SIGNAL_IMITATION_SUITE_LEVEL_MAPPING = {
'test_imitate_leak': 'imitate_leak_test',
'test_imitate_leak_coordinate': 'imitate_leak_coordinate_test',
'test_imitate_leak_volume': 'imitate_leak_volume_test',
'test_imitate_leak_time': 'imitate_leak_time_test',
'test_imitate_acknowledge': 'imitate_acknowledge_test',
'test_imitate_mask': 'imitate_mask_test',
'test_imitate_mask_reason': 'imitate_mask_reason_test',
'test_imitate_pumping_status': 'imitate_pumping_status_test',
'test_imitate_lds_status': 'imitate_lds_status_test',
'test_imitate_free_flow': 'imitate_free_flow_test',
}
# Мержим все вместе чтобы не переписывать логику коллектора айтемов (тестов)
SUITE_LEVEL_TEST_MAPPING = {
**SMOKE_SUITE_LEVEL_MAPPING,
**LDS_STATUS_SUITE_LEVEL_MAPPING,
**STATIONARY_STATUS_SUITE_LEVEL_MAPPING,
**IS_REJECTED_SUITE_LEVEL_MAPPING,
**OUTPUT_SIGNAL_IMITATION_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)
if test_name in OUTPUT_SIGNAL_IMITATION_SUITE_LEVEL_MAPPING:
suite_config = params['config']
attr_name = OUTPUT_SIGNAL_IMITATION_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 для группировки по наборам данных
5. Помечает маркером skip_reason тесты, у которых в конфигурации теста задан skip_reason (CaseMarkers.skip_reason)
"""
# Получаем список выбранных наборов из --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
# Скип по причине из конфигурации теста: тест остаётся в прогоне, но будет пропущен
# фикстурой skip_reason_guard и попадёт в отчёт как skipped с причиной
if skip_reason := resolve_skip_reason(test_config):
item.add_marker(pytest.mark.skip_reason(reason=skip_reason))
logger.info("[SKIP] %s: %s", item.nodeid, skip_reason)
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 skip_reason_guard(request, allure_suite_hierarchy, allure_tms_link):
"""
Скипает тест, если для него задан skip_reason в конфигурации теста (CaseMarkers.skip_reason).
Порядок фикстур важен, поэтому зависимости запрошены явно:
- allure_suite_hierarchy и allure_tms_link должны отработать ДО скипа, иначе в отчёте
потеряются группировка по набору данных и линк на тест-кейс TestOps;
- require_suite_infra и offset_wait зависят от этой фикстуры, чтобы скипнутый тест
не ждал свой offset и не подменял причину скипа сообщением об инфраструктуре.
"""
if skip_reason := get_skip_reason_marker(request.node):
decorate_skipped_test_in_allure(request.node, skip_reason)
pytest.skip(skip_reason)
# skip_reason_guard в аргументах require_suite_infra/offset_wait нужен только для порядка фикстур:
# скип теста должен сработать до проверки готовности инфраструктуры
@pytest.fixture(autouse=True)
def require_suite_infra(request, skip_reason_guard):
"""
Пропускает тесты набора, если 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] Набор пропущен: инфраструктура не готова")
# skip_reason_guard в аргументах offset_wait нужен только для порядка фикстур:
# скипнутый тест не должен ждать свой offset
@pytest.fixture(autouse=True)
def offset_wait(request, skip_reason_guard):
"""
Offset‑ожидание перед каждым тестом относительно фактического старта core
"""
cfg = request.config.group_state
if cfg.get("current_suite") and not cfg.get("suite_infra_ready"):
return
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(...) (в минутах)
- Offset скипнутых тестов (маркер skip_reason) не учитывается: они не выполняются
- Если 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 не нужно
if is_marked_as_skipped(suite_item):
continue
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()"
)
def _start_persistent_ws_connection(cfg: dict, marker_tu_id: int) -> None:
"""
Поднимает постоянное соединение набора до старта имитатора.
Соединение должно существовать с нулевой минуты прогона: именно negotiate и живые пинги
заставляют api-gateway держать консьюмеры core и layer builder, иначе сообщения копятся
и приходят в тесты с опозданием. Ошибка не прерывает прогон - соединение поднимет
supervisor менеджера, а в отчёте остаётся ALERT.
"""
tu_id = cfg.get("resolved_tu_id") or marker_tu_id
try:
start_ws_connection_manager(cfg, int(tu_id))
except BaseException as error:
message = f"[WS_MANAGER] [ALERT] Не удалось поднять постоянное соединение: {error}"
logger.warning(message)
try:
allure.attach(message, name="ALERT", attachment_type=allure.attachment_type.TEXT)
except Exception:
logger.debug("Не удалось прикрепить ALERT постоянного соединения к Allure")
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 набора")
stand_manager.server_test_data_remover()
cfg["suite_infra_ready"] = False
cfg["suite_setup_failure"] = message
cfg["suite_start_time"] = None
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
# Постоянное соединение закрывается до сброса auth: ему нужны текущие token и x-user-id
close_ws_connection_manager(cfg)
if stand_manager := cfg["stand_manager"]:
try:
stand_manager.stop_imitator_wrapper()
try:
stand_manager.restore_signal_unit_conversion_rules()
except Exception:
logger.exception(
"[ERROR] [SETUP] Ошибка при восстановлении signal_unit_conversion_rules.json "
"перед запуском нового набора"
)
stand_manager.server_test_data_remover()
finally:
cfg["stand_manager"] = None
# 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]
ost_name = item.get_closest_marker("ost_name").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"
)
try:
vault_rejection_enabled, vault_setup_error = configure_vault_for_suite(suite_config)
if vault_setup_error:
_skip_current_suite_after_setup_failure(cfg, vault_setup_error)
else:
cfg["vault_rejection_enabled"] = vault_rejection_enabled
except Exception as error:
_skip_current_suite_after_setup_failure(cfg, f"[SETUP] [ERROR] Vault: {error})")
stand_manager = StandSetupManager(
duration_m=imitator_duration,
test_data_id=data_id,
test_data_name=test_data_name,
tu_id=tu_id,
ost_name=ost_name,
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}"
)
_skip_current_suite_after_setup_failure(cfg, msg)
try:
stand_manager.start_opc_ssh_tunnel()
except Exception as error:
_skip_current_suite_after_setup_failure(
cfg, f"[SETUP] [ERROR] не удалось поднять SSH-туннель " f"для подключения по OPC: {error}"
)
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_configuration_data(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}")
_start_persistent_ws_connection(cfg, tu_id)
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, cfg, 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["is_target_tu_active"] = 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"]:
close_ws_connection_manager(cfg)
if cfg.get("vault_rejection_enabled"):
reset_vault_process_empty_values_rejection()
cfg["vault_rejection_enabled"] = False
# Обёртка teardown в try-except для предотвращения влияния ошибок на статус тестов
try:
if stand_manager := cfg["stand_manager"]:
try:
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"
)
stand_manager.server_test_data_remover()
finally:
cfg["stand_manager"] = None
cfg["current_suite"] = None
cfg["suite_start_time"] = None
cfg["imitator_start_time"] = None
cfg["suite_infra_ready"] = False
# опционально дождаться завершения потока - безопасный 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() фона имитатора")
except Exception as teardown_error:
# Логирование ошибки teardown, но не прерывание pytest
# Ошибки teardown не должны влиять на статус тестов
logger.error(
"[TEARDOWN] [ERROR] Ошибка при teardown набора '%s': %s",
cfg.get("current_suite"),
teardown_error,
exc_info=True,
)
try:
allure.attach(
f"[TEARDOWN] [ERROR] Остановка инфраструктуры завершилась с ошибкой: {teardown_error}",
name="Ошибка teardown",
attachment_type=allure.attachment_type.TEXT,
)
except Exception:
logger.debug("Не удалось прикрепить ошибку teardown к Allure", exc_info=True)
@pytest.fixture
def ws_client(request):
"""
Фикстура постоянного ws-соединения набора данных.
Соединение одно на набор и живёт в фоновом потоке менеджера, поэтому пинги и вычитывание
сообщений не прерываются ни между тестами, ни во время offset-ожидания. Тест получает фасад,
привязанный к своей области видимости (scope): watermark scope сдвигается на каждую подписку,
поэтому контент подписок прошлых тестов в проверки текущего попасть не может, а по
завершении теста очереди scope уничтожаются.
request - даёт доступ к request.config.group_state (общий кэш auth на dataset).
:return: фасад постоянного wss соединения для текущего теста
"""
cfg = request.config.group_state
ensure_auth_for_fixture(cfg)
manager = get_ws_connection_manager(cfg)
if manager is None:
manager = start_ws_connection_manager(cfg, _resolve_bundle_tu_id(request, cfg))
scope = manager.open_scope(request.node.nodeid)
try:
yield manager.client_for(scope)
finally:
manager.close_scope(scope)
def _resolve_bundle_tu_id(request, cfg: dict) -> int:
"""tuId для bundle-подписок: из Администрирования, иначе из маркера набора данных."""
resolved_tu_id = cfg.get("resolved_tu_id")
if resolved_tu_id:
return int(resolved_tu_id)
tu_id_marker = request.node.get_closest_marker("tu_id")
if tu_id_marker is None:
pytest.fail("Не удалось определить tuId для bundle-подписок постоянного соединения")
return int(tu_id_marker.args[0])
@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
@pytest_asyncio.fixture
async def opc_client(request):
"""
Фикстура OPC UA клиента для чтения выходных сигналов СОУ.
Подключается к серверу, указанному в OPC_URL (opc.tcp://host:port).
"""
from clients.opc_ua_client import OpcUaClient
stand_manager = request.config.group_state.get("stand_manager")
opc_url = stand_manager.opc_tunnel_url if stand_manager else None
if not opc_url:
opc_url = os.environ.get(EnvKeyConstants.OPC_URL)
if not opc_url:
pytest.fail("OPC_URL не задан в переменных окружения")
client = OpcUaClient(opc_url)
try:
await client.connect()
except Exception as error:
pytest.fail(f"Не удалось подключиться к OPC UA серверу ({opc_url}): {error}")
try:
yield client
finally:
try:
await client.disconnect()
except Exception as error:
logger.warning("[OPC] [WARNING] Ошибка при отключении OPC UA клиента: %s", error)
def pytest_sessionfinish(session, exitstatus):
"""
В завершении сессии — отправляем единый Allure‑отчёт в TestOps.
"""
# 1) teardown стенда: LDS Configurator + остановка имитатора
try:
group_state = getattr(session.config, "group_state", {})
close_ws_connection_manager(group_state)
if group_state.get("vault_rejection_enabled"):
reset_vault_process_empty_values_rejection()
group_state["vault_rejection_enabled"] = False
_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)
