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