---

хттп кли
from constants.architecture_constants import HTTPClientConstants as Http_const
from utils.helpers.log_safety import mask_secrets



            if not self.suppress_recv_logging:
                logging.info(f"[HTTP_CLIENT] Выполняю запрос: METHOD: {method} URL: {mask_secrets(url)}")




43стр
        except requests.HTTPError as error:
            logger.error(
                f"[HTTP_CLIENT] [ERROR] При выполнении запроса. METHOD: {method} "
                f"URL: {mask_secrets(url)} ERROR_TEXT:{mask_secrets(error)}"
            )
            raise
        except requests.RequestException as error:
            # traceback печатает текст исключения requests, в который вложен URL, поэтому вместо
            # logger.exception логируем ошибку вручную с маскированием; тип исключения сохраняется
            logger.error(
                f"[HTTP_CLIENT] [ERROR] При выполнении запроса. METHOD: {method} "
                f"URL: {mask_secrets(url)} ERROR_TEXT:{type(error).__name__}: {mask_secrets(error)}"
            )
            raise







            if not self.suppress_recv_logging:
                logging.info(f"[HTTP_CLIENT] Выполняю запрос: METHOD: {method} URL: {mask_secrets(url)}")





        except requests.HTTPError as error:
            logger.error(
                f"[HTTP_CLIENT] [ERROR] При выполнении запроса. METHOD: {method} "
                f"URL: {mask_secrets(url)} ERROR_TEXT:{mask_secrets(error)}"
            )
            raise
        except requests.RequestException as error:
            logger.error(
                f"[HTTP_CLIENT] [ERROR] При выполнении запроса. METHOD: {method} "
                f"URL: {mask_secrets(url)} ERROR_TEXT:{type(error).__name__}: {mask_secrets(error)}"
            )
            raise




























персист вс кли
        return await self._call(self._manager.dispatcher.receive_by_invocation_id, self._scope, invocation_id, timeout)

    def health_snapshot(self) -> dict[str, Any]:
        """Снимок здоровья постоянного соединения: для диагностики падений тестов."""
        return self._manager.health_snapshot()
























сигнал р негот
from constants.architecture_constants import WebSocketClientConstants as WS_Const
from utils.helpers.log_safety import mask_secrets



                   last_error = error
                    logger.warning(
                        "[NEGOTIATE] Попытка %s из %s неуспешна: %s", attempt, max_retries, mask_secrets(error)
                    )
                    if attempt < max_retries:
                        time.sleep(backoff)
        finally:
            self._http_client.suppress_recv_logging = previous_suppression

        raise NegotiationError(f"negotiate не выполнен после {max_retries} попыток: {mask_secrets(last_error)}")



















вс кли 
import time
from collections import Counter




    async def connect(self) -> None:
после текста все тело
        await self._cancel_task(self._ping_task)
        self._ping_task = None
        await self._cancel_task(self._recv_task)
        self._recv_task = None
        if self._ws is not None:
            await self._ws.close()
            self._ws = None
        await self._connect_loop()









        transient_errors = (ConnectionError, OSError, asyncio.TimeoutError, InvalidStatus)
        attempt_reasons: Counter[str] = Counter()




                remaining = deadline - time.monotonic()
                if isinstance(exc, InvalidStatus):
                    stage = "wss-handshake"
                elif isinstance(exc, asyncio.TimeoutError):
                    stage = "handshake-timeout"
                else:
                    stage = type(exc).__name__  # например NegotiationError
                attempt_reasons[stage] += 1
                if remaining <= 0:
                    raise TimeoutError(
                        f"Подключение к hub не удалось: этап={stage}, бюджет={timeout:.0f} с, "
                        f"попыток={attempt}, причины={dict(attempt_reasons)}, "
                        f"последняя ошибка={exc}"
                    ) from exc




                            pause = WS_Const.LB_MISS_RETRY_INTERVAL_S
                logger.warning(
                    "WSS подключение не установлено (попытка %s, этап=%s%s): %s. " "Повтор через %s с, осталось %.0f с",
                    attempt,
                    stage,
                    status_info,
                    exc,
                    pause,
                    remaining,
                )







для 
    def _log_connection_summary(self) -> None:


новое тело 

        logger.info(
            "[WS_HEALTH] connectionId=%s pings_sent=%s pings_failed=%s messages_received=%s " "queue_size=%s",
            self._connection_id,
            self.ws_pings_sent,
            self.ws_pings_failed,
            self.messages_received,
            self.recv_queue.qsize(),
        )







УДАЛИТЬ
async def reconnect(self):
        """
        Переподключение websocket соединения
        """
        try:
            await self._cancel_task(self._ping_task)
            self._ping_task = None
            await self._cancel_task(self._recv_task)
            self._recv_task = None

            if self._ws:
                await self._ws.close()
                self._ws = None
            await self._connect_loop(WS_Const.WS_RECONNECT_TIMEOUT_SECONDS)
            self.reconnect_count += 1
            return self
        except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
            raise RuntimeError(f"Ошибка при попытке переподключения {error}")























вс конн мен
from constants.architecture_constants import PersistentConnectionConstants as PersistConst
from constants.architecture_constants import WebSocketClientConstants as WS_Const
from constants.test_constants import BaseTN3Constants as TestConst
from utils.helpers.log_safety import mask_secrets





        self._tasks: List[asyncio.Task] = []
        self._connect_lock: Optional[asyncio.Lock] = None






        self._client = WebSocketClient(self._host, self._access_token, self._x_user_id)
        self._connect_lock = asyncio.Lock()






        return bool(self._client and self._client.is_connected)

    def _get_connect_lock(self) -> Lock | None:
        """
        Лок установки соединения с ленивой инициализацией.

        asyncio.Lock создаётся в том же loop, который будет им пользоваться, 
        первый вызов происходит внутри корутин менеджера.
        """
        if self._connect_lock is None:
            self._connect_lock = asyncio.Lock()
        return self._connect_lock






        while not self._stopped.is_set():
            try:
                async with self._get_connect_lock():
                    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







                self.connect_failures += 1
                logger.warning(
                    f"[WS_MANAGER] Не удалось установить соединение: {mask_secrets(error)}. "
                    f"Повтор через {backoff:.0f} с"
                )











                self.rest_pings_failed += 1
                logger.warning(f"[WS_MANAGER] REST Ping не выполнен: {mask_secrets(error)}")









удалить
            "clientReconnects": client.reconnect_count if client else 0,






удалить
    async def reconnect_now(self) -> None:
        """Принудительное переподключение с восстановлением bundle и подписок тестов."""
        self._ensure_client()
        await self._client.reconnect()
        await self._subscribe_bundle()
        await self._restore_scope_subscriptions()























вс месс диспатчер
    def register_subscription(self, scope: SubscriptionScope, target: str, args: Any) -> None:
        """
        Запоминает подписку скоупа для повторной отправки после реконнекта.
        """
        key = (target, repr(args))
        known = {(subscription[0], repr(subscription[1])) for subscription in scope.subscriptions}
        if key not in known:
            scope.subscriptions.append((target, args))





























арх конст
удалить
    WS_RECONNECT_TIMEOUT_SECONDS: float = 20.0





    SIGNAL_STATE_POLL_ATTEMPTS = 12  # Число опросов состояния датчика
    SUBSCRIBE_REPLY_TIMEOUT_SECONDS = 60.0






























вс тест утился
    fail(
        f"При {purpose} за {total_wait_seconds} с не получен StreamItem с fileChunk "
        f"({request_name}, invocation_id={invocation_id}). Смотреть вложения received ws message"
    )


def describe_connection_health(ws_client: Any) -> str:
    """
    Однострочное состояние ws-соединения для текста падения теста.
    """
    try:
        if hasattr(ws_client, "health_snapshot"):
            snapshot = ws_client.health_snapshot()
            return (
                f"Состояние ws-соединения: isConnected={snapshot.get('isConnected')}, "
                f"connectionId={snapshot.get('connectionId')}, "
                f"установлено соединений={snapshot.get('connectionsEstablished')}, "
                f"ошибок подключения={snapshot.get('connectFailures')}, "
                f"ws-пинги={snapshot.get('wsPingsSent')}/{snapshot.get('wsPingsFailed')}, "
                f"rest-пинги={snapshot.get('restPingsSent')}/{snapshot.get('restPingsFailed')}, "
                f"принято сообщений={snapshot.get('messagesReceived')}, "
                f"доставлено в тесты={snapshot.get('messagesRouted')}"
            )
        # Разовый клиент (flow LDS Configurator): его счётчики копятся на одном соединении
        return (
            f"Состояние ws-соединения: isConnected={ws_client.is_connected}, "
            f"connectionId={ws_client.connection_id}, "
            f"ws-пинги={ws_client.ws_pings_sent}/{ws_client.ws_pings_failed}, "
            f"принято сообщений={ws_client.messages_received}"
        )
    except Exception as error:  # диагностика не должна маскировать исходное падение
        return f"Состояние ws-соединения: недоступно ({type(error).__name__}: {error})"










функц целиком
async def _receive_subscribed_messages(
    ws_client: WebSocketClient,
    ws_message_type: str,
    ws_invoke_type: str,
    ws_invoke_params: Any,
    *,
    attempts: int,
    per_attempt_timeout: float,
) -> AsyncIterator[list]:
    """
    Генератор сообщений подписки: подписывается один раз и последовательно отдаёт
    сообщения нужного типа до исчерпания запаса времени.
    """
    await connect(ws_client, ws_invoke_type, ws_invoke_params)
    for attempt in range(1, attempts + 1):
        with allure.step(
            f"Получение сообщения с контентом типа: {ws_message_type} - попытка {attempt} из {attempts} "
            f"(до {per_attempt_timeout:.0f} с)"
        ):
            try:
                message = await ws_client.receive_by_type(ws_message_type, timeout=per_attempt_timeout)
            except asyncio.TimeoutError:
                continue
        yield message





    Подписывается и ищет allLeaksInfo сообщение для конкретного ТУ.
    Читает сообщения нужного типа с ретраями, фильтруя по tuId.




        if parsed_payload.replyContent.tuId == tu_id:
            return parsed_payload

    fail(
        f"Не удалось получить сообщение {ws_message_type} для ТУ {tu_id} за {attempts} попыток "
        f"(лимит ~{attempts * TestConst.POLL_BY_TIMEOUT_SECONDS:.0f} с). "
        f"{describe_connection_health(ws_client)}"
    )









вся функция
async def connect_and_poll_subscribed_msg(
    ws_client: WebSocketClient,
    ws_message_type: str,
    ws_invoke_type: str,
    ws_invoke_params: Any = None,
) -> list:
    """
    Подписывается и ждёт нужное сообщение одним чтением до SUBSCRIBE_REPLY_TIMEOUT_SECONDS.
    """
    ws_client.suppress_recv_logging = True
    try:
        async for payload in _receive_subscribed_messages(
            ws_client,
            ws_message_type,
            ws_invoke_type,
            ws_invoke_params,
            attempts=1,
            per_attempt_timeout=TestConst.SUBSCRIBE_REPLY_TIMEOUT_SECONDS,
        ):
            return payload
        fail(
            f"Бэкенд не прислал {ws_message_type} за {TestConst.SUBSCRIBE_REPLY_TIMEOUT_SECONDS:.0f} с "
            f"после подписки {ws_invoke_type} {ws_invoke_params}. {describe_connection_health(ws_client)}"
        )
    finally:
        ws_client.suppress_recv_logging = False









    fail(
        f"Сигналы id={sensor_ids} не перешли в состояние {flag_field}={expected_value} "
        f"за {attempts} попыток (лимит ~{attempts * per_attempt_timeout:.0f} с). "
        f"{describe_connection_health(ws_client)}"
    )






) -> tuple[ParsedPayloadType, InputSignalType]:
    """
    Подписывается и читает до max_messages сообщений из потока, пока не найдёт сигнал
    с нужным id.
    """







    fail(
        f"Информация по датчику {sensor_description} (id={sensor_id}) не пришла в ответе {ws_message_type} "
        f"после {max_messages} попыток чтения из потока подписки "
        f"(лимит ~{max_messages * timeout:.0f} с). Полученные id: {received_ids_text}. "
        f"{describe_connection_health(ws_client)}"
    )


























utils/helpers/log_safety.py
"""
Маскирование секретов в текстах логов и сообщений об ошибках.
"""

from __future__ import annotations

import re
from typing import Any

# token=<значение> в query-строке: значение идёт до разделителя параметра, пробела или кавычки
_TOKEN_QUERY_PATTERN = re.compile(r"(?i)([?&]token=)[^&\s'\"]+")
# standalone JWT (header.payload.signature), например внутри текста исключения
_JWT_PATTERN = re.compile(r"eyJ[\w-]+\.[\w-]+\.[\w-]+")

MASK: str = "***"


def mask_secrets(text: Any) -> str:
    """
    Возвращает текст с замаскированными секретами.
    """
    if text is None:
        return ""
    masked = _TOKEN_QUERY_PATTERN.sub(rf"\g<1>{MASK}", str(text))
    return _JWT_PATTERN.sub(MASK, masked)