Загрузка данных
саб проц
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 не задан в переменных окружения")