Загрузка данных


лдс конф утил
                fail(f"Неизвестный статус СОУ для ТУ '{tu.tuName}': {tu.status}", pytrace=False)


def extract_running_tus(admin_reply: GetBasicInfoAdminReply) -> list[AdminTuInfo]:
    """Возвращает все ТУ со статусом RUNNING из GetBasicInfoAdminResponse."""
    tus = admin_reply.replyContent.basicInfo.tus if admin_reply.replyContent else []
    return [tu for tu in tus if tu.status == SouAdminStatus.RUNNING.value]


def running_tus_to_snapshot(tus: list[AdminTuInfo]) -> list[dict[str, Any]]:
    """Сериализует список ТУ для хранения в group_state."""
    return [{"tuId": tu.tuId, "tuName": tu.tuName} for tu in tus]








                return True
            await asyncio.sleep(poll_interval_seconds)
        return False


async def stop_tu_and_wait(
    ws_client: WebSocketClient,
    parser: WsMessageParser,
    tu_id: int,
    tu_name: Optional[str] = None,
) -> None:
    """
    Останавливает СОУ и ждёт статус STOPPED в Администрировании.
    При таймауте - pytest.fail (блокирующая ошибка setup).
    """
    label = f"tuId={tu_id}, tuName={tu_name!r}" if tu_name else f"tuId={tu_id}"
    with _step(f"Остановка СОУ ({label})"):
        await invoke_lds_command(ws_client, parser, LdsCfgConst.STOP_LDS_REQUEST, tu_id)
        if not await poll_admin_tu_status(ws_client, parser, tu_id, SouAdminStatus.STOPPED):
            fail(
                f"СОУ не выключилась за {int(LdsCfgConst.POLL_TIMEOUT_SECONDS)} с: {label}",
                pytrace=False,
            )


async def launch_tu_and_wait(
    ws_client: WebSocketClient,
    parser: WsMessageParser,
    tu_id: int,
) -> bool:
    """
    Запускает СОУ и ждёт статус RUNNING в Администрировании.
    Возвращает True при успехе (для soft teardown).
    """
    with _step(f"Запуск СОУ (tuId={tu_id})"):
        await invoke_lds_command(ws_client, parser, LdsCfgConst.LAUNCH_LDS_REQUEST, tu_id)
        return await poll_admin_tu_status(ws_client, parser, tu_id, SouAdminStatus.RUNNING)


async def stop_all_running_tus(
    ws_client: WebSocketClient,
    parser: WsMessageParser,
    tus: list[AdminTuInfo],
) -> None:
    """Последовательно останавливает все включённые ТУ на стенде."""
    if not tus:
        return
    logger.info("[SETUP] Остановка всех включённых ТУ на стенде: %s шт.", len(tus))
    for tu in tus:
        logger.info("[SETUP] Остановка ТУ на стенде: tuId=%s, tuName=%r", tu.tuId, tu.tuName)
        await stop_tu_and_wait(ws_client, parser, tu.tuId, tu.tuName)


async def restore_pre_run_tus(
    ws_client: WebSocketClient,
    parser: WsMessageParser,
    snapshot: list[dict[str, Any]],
    exclude_tu_id: int,
) -> None:
    """
    Включает обратно ТУ из снимка, кроме ТУ автотестов.
    Ошибки по отдельным ТУ оформляются как ALERT без падения прогона.
    """
    to_restore = [entry for entry in snapshot if entry.get("tuId") != exclude_tu_id]
    if not to_restore:
        return
    logger.info(
        "[TEARDOWN] Восстановление ТУ стенда: %s шт. (исключена ТУ автотестов tuId=%s)",
        len(to_restore),
        exclude_tu_id,
    )
    for entry in to_restore:
        restore_tu_id = entry["tuId"]
        restore_tu_name = entry.get("tuName", "")
        logger.info(
            "[TEARDOWN] Запуск ТУ из снимка: tuId=%s, tuName=%r",
            restore_tu_id,
            restore_tu_name,
        )
        if not await launch_tu_and_wait(ws_client, parser, restore_tu_id):
            attach_allure_alert(
                f"Не удалось восстановить ТУ из снимка: tuId={restore_tu_id}, "
                f"tuName={restore_tu_name!r}. Проверить вручную."
            )



















лдс конф сцен
    group_state["resolved_tu_id"] = tu_id
    group_state["admin_tu_name"] = cfg.admin_tu_name


def _save_pre_run_running_tus(
    group_state: Optional[Dict[str, Any]],
    snapshot: list[Dict[str, Any]],
) -> None:
    """Сохраняет список включённых ТУ стенда до запуска автотестов."""
    if group_state is None:
        return
    group_state["pre_run_running_tus"] = snapshot








удалить
    2. При необходимости остановить уже запущенную СОУ.
на
    2. Снять слепок всех RUNNING ТУ и остановить их на стенде.




удалить
sou_status: SouAdminStatus





удалить
    sou_status = SouAdminStatus(admin_tu.status)







        admin_tu.tuName,
        SouAdminStatus(admin_tu.status),










заменить
 if sou_status == SouAdminStatus.RUNNING:
        logger.info("[SETUP] Остановка СОУ перед холодным запуском (уже была включена)")
        await lds_utils.invoke_lds_command(ws_client, parser, LdsCfgConst.STOP_LDS_REQUEST, tu_id)
        if not await lds_utils.poll_admin_tu_status(ws_client, parser, tu_id, SouAdminStatus.STOPPED):
            fail(
                "Не удалось перезапустить СОУ: статус в Администрировании не стал 'выключена' за 2 минуты",
                pytrace=False,
            )
на 
    running_tus = lds_utils.extract_running_tus(admin_reply)
    pre_run_snapshot = lds_utils.running_tus_to_snapshot(running_tus)
    _save_pre_run_running_tus(group_state, pre_run_snapshot)
    logger.info(
        "[SETUP] Спсиок включённых ТУ на стенде: %s шт. %s",
        len(pre_run_snapshot),
        pre_run_snapshot,
    )
    await lds_utils.stop_all_running_tus(ws_client, parser, running_tus)









async def lds_configurator_teardown(
    ws_client: WebSocketClient,
    tu_id: int,
    admin_tu_name: str,
    pre_run_running_tus: Optional[list[Dict[str, Any]]] = None,
) -> None:
    """
    Teardown набора: остановка СОУ автотестов и восстановление ТУ стенда из снимка.
    Некритичные отклонения логируются без падения прогона.
    """
    snapshot = pre_run_running_tus or []
    try:
        logger.info("[TEARDOWN] Проверка статуса СОУ (tuId=%s, «%s»)", tu_id, admin_tu_name)
        ws_client.clear_queue()
        admin_reply = await lds_utils.get_basic_info_admin_with_retry(ws_client, parser)

        sou_status = lds_utils.get_admin_tu_status(admin_reply, tu_id)
        if sou_status == SouAdminStatus.RUNNING:
            logger.info("[TEARDOWN] Остановка СОУ (StopLdsRequest) для tuId=%s", tu_id)
            await lds_utils.invoke_lds_command(ws_client, parser, LdsCfgConst.STOP_LDS_REQUEST, tu_id)

            logger.info("[TEARDOWN] Ожидание выключения СОУ в Администрировании")
            if not await lds_utils.poll_admin_tu_status(ws_client, parser, tu_id, SouAdminStatus.STOPPED):
                lds_utils.attach_allure_alert(
                    f"СОУ не выключилась за 2 минуты после StopLdsRequest. "
                    f"tuId={tu_id}, adminTuName={admin_tu_name!r}. Проверить вручную."
                )
        else:
            lds_utils.attach_allure_alert(
                f"СОУ не в статусе 'включена' при teardown (status={sou_status}), остановка пропущена. "
                f"tuId={tu_id}, adminTuName='{admin_tu_name}'"
            )

        await lds_utils.restore_pre_run_tus(ws_client, parser, snapshot, exclude_tu_id=tu_id)
    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}"
        )


























конфтест
        "admin_tu_name": None,
        "pre_run_running_tus": None,





admin_tu_name = cfg.get("admin_tu_name") or ""
    pre_run_running_tus = cfg.get("pre_run_running_tus") or []






заменить
            await lds_configurator_scenarios.lds_configurator_teardown(client, tu_id, admin_tu_name)
на 
            await lds_configurator_scenarios.lds_configurator_teardown(
                client, tu_id, admin_tu_name, pre_run_running_tus
            )








        cfg["admin_tu_name"] = None
        cfg["pre_run_running_tus"] = None