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


@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
        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}")

        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 продолжит выполнение теста