Загрузка данных
лдс конф утил
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