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


саб проц
            logging.exception(f"[POPEN] [ERROR] Ошибка выполнения команды {cmd}")

    def open_ssh_tunnel(self, local_port: int, remote_host: str, remote_port: int) -> Optional[subprocess.Popen]:
        """
        Открывает SSH-туннель (локальный проброс порта) до remote_host:remote_port.
        После этого remote_host:remote_port доступен на раннере как localhost:local_port.
        """
        if os.name == Im_const.OS_NAME_WIN:
            ssh_prefix = f"ssh -i {self._ssh_key_name}"
        else:
            ssh_prefix = "ssh"
        tunnel_cmd = f"{ssh_prefix} -N -L {local_port}:{remote_host}:{remote_port} {self._username}@{self._host}"
        return self.exec_popen(tunnel_cmd, use_ssh=False)











арх конст

    DISCOVERY_TIMEOUT_S: float = 5.0
    # Пауза после поднятия SSH-туннеля до OPC перед началом работы.
    TUNNEL_START_DELAY_S: float = 2.0


























стенд сет

        self._remote_data_uploaded = False
        self._opc_tunnel_process = None
        self._opc_tunnel_port: Optional[int] = None





        В teardown может вызываться даже если имитатор не запущен
        """
        try:
            self.stop_opc_ssh_tunnel()
        except Exception:
            logger.exception("[TEARDOWN] [ERROR] Ошибка при закрытии OPC SSH-туннеля")
        try:





        logger.info(f"[SETUP] [OK] OPC сервер {host}:{port} доступен")

    def start_opc_ssh_tunnel(self) -> None:
        """
        Поднимает SSH-туннель до OPC сервера: пробрасывает порт OPC на runner.
        После этого OPC доступен на runner'е как opc.tcp://localhost:<port>.
        """
        from constants.architecture_constants import OpcUaConstants as OpcConst

        opc_host, opc_port = self._parse_opc_target()
        self._opc_tunnel_port = opc_port
        process = self._stand_client.open_ssh_tunnel(opc_port, opc_host, opc_port)
        if process is None:
            raise RuntimeError("[SETUP] [ERROR] Не удалось запустить OPC SSH-туннель")
        self._opc_tunnel_process = process
        time.sleep(OpcConst.TUNNEL_START_DELAY_S)
        logger.info("[SETUP] [OK] OPC SSH-туннель поднят: localhost:%s -> %s:%s", opc_port, opc_host, opc_port)

    def stop_opc_ssh_tunnel(self) -> None:
        """Закрывает SSH-туннель до OPC сервера."""
        if self._opc_tunnel_process is None:
            return
        SubprocessClient.terminate_process(self._opc_tunnel_process, timeout=5.0)
        self._opc_tunnel_process = None
        self._opc_tunnel_port = None
        logger.info("[TEARDOWN] [OK] OPC SSH-туннель закрыт")

    @property
    def opc_tunnel_url(self) -> Optional[str]:
        """URL OPC через SSH-туннель (localhost) либо None, если туннель не поднят."""
        if self._opc_tunnel_port:
            return f"opc.tcp://localhost:{self._opc_tunnel_port}"
        return None

    def _get_server_ip(self) -> str:














конф
            _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()







    from clients.opc_ua_client import OpcUaClient

    stand_manager = request.config.group_state.get("stand_manager")
    opc_url = stand_manager.opc_tunnel_url if stand_manager else None
    if not opc_url:
        opc_url = os.environ.get(EnvKeyConstants.OPC_URL)
    if not opc_url:
        pytest.fail("OPC_URL не задан в переменных окружения")