---

конфтест
from utils.helpers import lds_configurator_utils as lds_cfg_utils
from utils.helpers.ws_message_parser import ws_message_parser as lds_ws_parser





def _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)




















вс мс парс
from __future__ import annotations

import logging








from models.upload_exported_file_model import DownloadExportedDataReply

logger = logging.getLogger(__name__)





            message = from_dict(
                data_class=data_class, data=data, config=config or self._dacite_config  # type: ignore[arg-type]
            )
            if not self.suppress_recv_logging:
                try:
                    attach(
                        str(message) + f" {datetime.now(ZoneInfo(WebSocketClientConstants.ZONE_INFO))}",
                        name=data_class_name,
                        attachment_type=attachment_type.TEXT,
                    )
                except (KeyError, RuntimeError) as error:
                    logger.debug("Allure attach пропущен: %s", error)

























лдс конфиг утилс
import allure
from _pytest.outcomes import Failed





def is_configurator_flow_active() -> bool:
    """True во время setup/teardown/verify СОУ через Администрирование (lds-configurator) вне теста."""
    return _configurator_flow_active


def _fail_or_raise(message: str) -> None:
    """В configurator flow — RuntimeError для retry/teardown; иначе pytest.fail."""
    if _configurator_flow_active:
        raise RuntimeError(message)
    fail(message, pytrace=False)






async def get_basic_info_admin_with_retry(
    ws_client: WebSocketClient,
    parser: WsMessageParser,
    retries: int = LdsCfgConst.GET_BASIC_INFO_ADMIN_RETRIES,
) -> GetBasicInfoAdminReply:
    """
    Запрашивает GetBasicInfoAdminResponse с повторными попытками.
    """
    last_error: Optional[BaseException] = None
    for attempt in range(1, retries + 1):
        with _step(f"Запрос списка ТУ в Администрировании - попытка {attempt} из {retries}"):
            try:
                return await get_basic_info_admin(ws_client, parser)
            except (
                asyncio.TimeoutError,
                ConnectionError,
                ConnectionResetError,
                OSError,
                RuntimeError,
                KeyError,
                Failed,
            ) as error:
                last_error = error
                logger.warning(
                    "[LDS_CONFIGURATOR] GetBasicInfoAdmin попытка %s/%s: %s: %r",
                    attempt,
                    retries,
                    type(error).__name__,
                    error,
                )
                if attempt < retries:
                    await asyncio.sleep(1)

    with _step("Проверка: GetBasicInfoAdminResponse получен"):
        _fail_or_raise(
            f"Не удалось получить GetBasicInfoAdminResponse за {retries} попыток: {last_error}",
        )



















вс тест утилс
                await ws_client.invoke(ws_invoke_type, ws_invoke_params)
    except (asyncio.TimeoutError, ConnectionError, ConnectionResetError, OSError) as error:
        if _is_configurator_flow_active():
            raise










                f"tuId={tu_id}, adminTuName={admin_tu_name!r}. Проверить вручную."
            )
    except BaseException as error:
        logger.warning(
            "[TEARDOWN] [ALERT] LDS Configurator teardown: %s: %r. tuId=%s, adminTuName=%r",
            type(error).__name__,
            error,
            tu_id,
            admin_tu_name,
        )
        lds_utils.attach_allure_alert(
            f"Ошибка LDS Configurator teardown: {type(error).__name__}: {error!r}. "
            f"tuId={tu_id}, adminTuName={admin_tu_name!r}"
        )