From eaeac67742615be36161098ad79455710ed8a9b6 Mon Sep 17 00:00:00 2001 From: Tony Coder <407243179@qq.com> Date: Thu, 20 Aug 2026 10:06:40 +0000 Subject: [PATCH 1/2] fix: release hdc fport on driver close and process exit Signed-off-by: Tony Coder <407243179@qq.com> --- hmdriver2/_client.py | 74 +++++++++++-- hmdriver2/driver.py | 46 ++++++++- hmdriver2/hdc.py | 23 ++++- tests/test_hdc_port_lifecycle.py | 172 +++++++++++++++++++++++++++++++ 4 files changed, 301 insertions(+), 14 deletions(-) create mode 100644 tests/test_hdc_port_lifecycle.py diff --git a/hmdriver2/_client.py b/hmdriver2/_client.py index 6dfe01a..71a5bb2 100644 --- a/hmdriver2/_client.py +++ b/hmdriver2/_client.py @@ -1,13 +1,14 @@ # -*- coding: utf-8 -*- +import atexit import socket import json +import sys import time import os import hashlib import typing from typing import Optional from datetime import datetime -from functools import cached_property from . import logger from .hdc import HdcWrapper @@ -24,17 +25,40 @@ class HmClient: def __init__(self, serial: str): self.hdc = HdcWrapper(serial) self.sock = None + self._local_port = None + self._released = False + self._atexit_registered = False - @cached_property + @property def local_port(self): - fports = self.hdc.list_fport() - logger.debug(fports) if fports else None - - return self.hdc.forward_port(UITEST_SERVICE_PORT) + if self._local_port is None: + fports = self.hdc.list_fport() + logger.debug(fports) if fports else None + self._local_port = self.hdc.forward_port(UITEST_SERVICE_PORT) + self._register_atexit() + return self._local_port + + def _register_atexit(self): + if self._atexit_registered: + return + atexit.register(self.release) + self._atexit_registered = True def _rm_local_port(self): + """Remove the HDC forward opened by this client. + + Must not read :attr:`local_port` when it was never allocated: that + property opens a new ``hdc fport`` and would leak another mapping. + """ + port = self._local_port + if not port: + logger.debug("rm fport skipped: no local port allocated") + return logger.debug("rm fport local port") - self.hdc.rm_forward(self.local_port, UITEST_SERVICE_PORT) + try: + self.hdc.rm_forward(port, UITEST_SERVICE_PORT) + finally: + self._local_port = None def _connect_sock(self): """Create socket and connect to the uiTEST server.""" @@ -140,23 +164,57 @@ def invoke_captures(self, api: str, args: typing.List = []) -> HypiumResponse: def start(self): logger.info("Start HmClient connection") + self._released = False _UITestService(self.hdc).init() self._connect_sock() self._create_hdriver() + def __enter__(self): + self.start() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.release() + def release(self): + """Close the socket and remove the local HDC port forward. + + Idempotent. Safe from ``Driver.close``, ``__exit__``, ``atexit``, and + ``__del__``. During interpreter finalization, ``hdc fport rm`` is + skipped (subprocess is unsafe then); atexit / explicit close run + earlier so the forward is still removed on a normal exit. + """ + if self._released: + return + self._released = True + logger.info(f"Release {self.__class__.__name__} connection") + try: if self.sock: + try: + self.sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass self.sock.close() self.sock = None - self._rm_local_port() + if getattr(sys, "is_finalizing", lambda: False)(): + logger.debug("release: interpreter finalizing, skipped fport rm") + return + self._rm_local_port() except Exception as e: logger.error(f"An error occurred: {e}") + finally: + if self._atexit_registered: + try: + atexit.unregister(self.release) + except Exception: + pass + self._atexit_registered = False def _create_hdriver(self) -> DriverData: logger.debug("Create uitest driver") diff --git a/hmdriver2/driver.py b/hmdriver2/driver.py index 5d588cc..30b3e4e 100644 --- a/hmdriver2/driver.py +++ b/hmdriver2/driver.py @@ -1,5 +1,6 @@ # -*- coding: utf-8 -*- +import atexit import json import uuid import re @@ -47,8 +48,10 @@ def __init__(self, serial: Optional[str] = None): self.serial = serial self._client = HmClient(self.serial) self.hdc = self._client.hdc + self._closed = False self._init_hmclient() self._initialized = True # Mark the instance as initialized + atexit.register(self.close) del self._serial_for_init # Clean up temporary attribute @classmethod @@ -71,10 +74,47 @@ def __call__(self, **kwargs) -> UiObject: return UiObject(self._client, **kwargs) - def __del__(self): - Driver._instance.clear() - if hasattr(self, '_client') and self._client: + def _unregister_singleton_if_self(self) -> None: + """Drop this serial from the singleton map only when the slot is us.""" + serial = getattr(self, "serial", None) + if serial is None: + return + if Driver._instance.get(serial) is self: + Driver._instance.pop(serial, None) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + def close(self) -> None: + """Release the Hypium socket and ``hdc fport`` mapping. + + Idempotent. Call this (or use ``with Driver()`` / ``stop()``) so the + forwarded local port is freed before the next run. ``__del__`` alone + is not reliable; ``atexit`` also invokes this on process exit. + """ + if getattr(self, "_closed", False): + return + self._closed = True + try: + atexit.unregister(self.close) + except Exception: + pass + if hasattr(self, "_client") and self._client: self._client.release() + self._unregister_singleton_if_self() + + def stop(self) -> None: + """Alias for :meth:`close`.""" + self.close() + + def __del__(self): + try: + self.close() + except Exception: + pass def _init_hmclient(self): self._client.start() diff --git a/hmdriver2/hdc.py b/hmdriver2/hdc.py index e1ff09c..bc3c44b 100644 --- a/hmdriver2/hdc.py +++ b/hmdriver2/hdc.py @@ -14,7 +14,13 @@ from .exception import HdcError, DeviceNotFoundError -def _execute_command(cmdargs: Union[str, List[str]]) -> CommandResult: +# hdc fport rm can block indefinitely (issues #47, #54). Bound teardown so +# driver close / atexit can finish and the local port is eligible for reuse. +FPORT_RM_TIMEOUT = 5 + + +def _execute_command(cmdargs: Union[str, List[str]], + timeout: Optional[float] = None) -> CommandResult: if isinstance(cmdargs, (list, tuple)): cmdline: str = ' '.join(list(map(shlex.quote, cmdargs))) elif isinstance(cmdargs, str): @@ -24,7 +30,16 @@ def _execute_command(cmdargs: Union[str, List[str]]) -> CommandResult: try: process = subprocess.Popen(cmdline, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) - output, error = process.communicate() + try: + output, error = process.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + process.kill() + try: + process.communicate() + except Exception: + pass + return CommandResult("", "timeout expired", -1) + output = output.decode('utf-8') error = error.decode('utf-8') exit_code = process.returncode @@ -86,7 +101,9 @@ def forward_port(self, rport: int) -> int: return lport def rm_forward(self, lport: int, rport: int) -> int: - result = _execute_command(f"{self.hdc_prefix} -t {self.serial} fport rm tcp:{lport} tcp:{rport}") + result = _execute_command( + f"{self.hdc_prefix} -t {self.serial} fport rm tcp:{lport} tcp:{rport}", + timeout=FPORT_RM_TIMEOUT) if result.exit_code != 0: raise HdcError("HDC rm forward error", result.error) return lport diff --git a/tests/test_hdc_port_lifecycle.py b/tests/test_hdc_port_lifecycle.py new file mode 100644 index 0000000..d45547b --- /dev/null +++ b/tests/test_hdc_port_lifecycle.py @@ -0,0 +1,172 @@ +# -*- coding: utf-8 -*- +"""HDC port-forward lifecycle: no Harmony device required.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from hmdriver2._client import UITEST_SERVICE_PORT, HmClient +from hmdriver2.driver import Driver +from hmdriver2.hdc import FPORT_RM_TIMEOUT, HdcWrapper, _execute_command +from hmdriver2.proto import CommandResult + + +@pytest.fixture(autouse=True) +def _clean_driver_registry(): + Driver._instance.clear() + yield + Driver._instance.clear() + + +@patch("hmdriver2.hdc.list_devices", return_value=["SERIAL"]) +@patch("hmdriver2.hdc._execute_command") +def test_rm_forward_runs_fport_rm_with_timeout(mock_ex, _mock_list): + mock_ex.return_value = CommandResult("", "", 0) + hdc = HdcWrapper("SERIAL") + hdc.rm_forward(10001, UITEST_SERVICE_PORT) + + args, kwargs = mock_ex.call_args + assert "fport rm tcp:10001 tcp:%s" % UITEST_SERVICE_PORT in args[0] + assert kwargs.get("timeout") == FPORT_RM_TIMEOUT + + +@patch("hmdriver2.hdc.list_devices", return_value=["SERIAL"]) +@patch("hmdriver2.hdc.FreePort") +@patch("hmdriver2.hdc._execute_command") +def test_forward_port_runs_hdc_fport(mock_ex, mock_free_port, _mock_list): + mock_ex.return_value = CommandResult("", "", 0) + mock_free_port.return_value.get.return_value = 12345 + hdc = HdcWrapper("SERIAL") + assert hdc.forward_port(UITEST_SERVICE_PORT) == 12345 + assert "fport tcp:12345 tcp:%s" % UITEST_SERVICE_PORT in mock_ex.call_args[0][0] + + +def test_execute_command_timeout_does_not_block(): + result = _execute_command("sleep 10", timeout=0.2) + assert result.exit_code != 0 + assert "timeout" in result.error.lower() + + +@patch("hmdriver2._client.HdcWrapper") +def test_client_release_removes_forward(mock_hdc_cls): + hdc = MagicMock() + hdc.forward_port.return_value = 10001 + hdc.list_fport.return_value = [] + mock_hdc_cls.return_value = hdc + + client = HmClient("SERIAL") + assert client.local_port == 10001 + hdc.forward_port.assert_called_once_with(UITEST_SERVICE_PORT) + + sock = MagicMock() + client.sock = sock + client.release() + + sock.close.assert_called_once() + hdc.rm_forward.assert_called_once_with(10001, UITEST_SERVICE_PORT) + assert client._local_port is None + + client.release() + assert hdc.rm_forward.call_count == 1 + + +@patch("hmdriver2._client.HdcWrapper") +def test_release_without_local_port_does_not_open_forward(mock_hdc_cls): + hdc = MagicMock() + mock_hdc_cls.return_value = hdc + + client = HmClient("SERIAL") + client.release() + + hdc.forward_port.assert_not_called() + hdc.rm_forward.assert_not_called() + + +@patch("hmdriver2._client.HdcWrapper") +@patch("hmdriver2._client.atexit.register") +def test_local_port_registers_atexit(mock_register, mock_hdc_cls): + hdc = MagicMock() + hdc.forward_port.return_value = 10001 + hdc.list_fport.return_value = [] + mock_hdc_cls.return_value = hdc + + client = HmClient("SERIAL") + _ = client.local_port + mock_register.assert_called() + assert mock_register.call_args[0][0] == client.release + + +@patch("hmdriver2._client.HdcWrapper") +def test_client_context_manager_releases_forward(mock_hdc_cls): + hdc = MagicMock() + hdc.forward_port.return_value = 10002 + hdc.list_fport.return_value = [] + mock_hdc_cls.return_value = hdc + + with patch.object(HmClient, "start", autospec=True) as mock_start: + def _start(self): + self.sock = MagicMock() + _ = self.local_port + + mock_start.side_effect = _start + with HmClient("SERIAL") as client: + assert client.local_port == 10002 + + hdc.rm_forward.assert_called_once_with(10002, UITEST_SERVICE_PORT) + + +@patch("hmdriver2.driver.HmClient") +@patch("hmdriver2.driver.list_devices", return_value=["SERIAL"]) +def test_driver_close_releases_client_and_frees_singleton(mock_list, mock_hm): + client = MagicMock() + mock_hm.return_value = client + + d = Driver("SERIAL") + d.close() + + client.release.assert_called() + assert "SERIAL" not in Driver._instance + + d.close() + assert client.release.call_count == 1 + + +@patch("hmdriver2.driver.HmClient") +@patch("hmdriver2.driver.list_devices", return_value=["SERIAL"]) +def test_driver_stop_and_context_manager_release(mock_list, mock_hm): + client = MagicMock() + mock_hm.return_value = client + + d = Driver("SERIAL") + d.stop() + client.release.assert_called() + assert "SERIAL" not in Driver._instance + + d2 = Driver("SERIAL") + with d2: + pass + assert client.release.call_count == 2 + assert "SERIAL" not in Driver._instance + + +@patch("hmdriver2.driver.HmClient") +@patch("hmdriver2.driver.list_devices", return_value=["AAA", "BBB"]) +def test_close_one_serial_keeps_other(mock_list, mock_hm): + mock_hm.side_effect = lambda serial: MagicMock() + + da = Driver("AAA") + db = Driver("BBB") + da.close() + + assert "AAA" not in Driver._instance + assert Driver._instance.get("BBB") is db + + +@patch("hmdriver2.driver.HmClient") +@patch("hmdriver2.driver.list_devices", return_value=["SERIAL"]) +@patch("hmdriver2.driver.atexit.register") +def test_driver_registers_atexit_close(mock_register, mock_list, mock_hm): + mock_hm.return_value = MagicMock() + d = Driver("SERIAL") + mock_register.assert_called() + assert mock_register.call_args[0][0] == d.close From b6b4472987487bc4005bdc3d65c5e074a4053a70 Mon Sep 17 00:00:00 2001 From: Tony Coder <407243179@qq.com> Date: Thu, 20 Aug 2026 10:08:21 +0000 Subject: [PATCH 2/2] fix: kill hung fport rm process group on timeout Signed-off-by: Tony Coder <407243179@qq.com> --- hmdriver2/hdc.py | 35 +++++++++++++++++++++++++++++++---- 1 file changed, 31 insertions(+), 4 deletions(-) diff --git a/hmdriver2/hdc.py b/hmdriver2/hdc.py index bc3c44b..6e86ce1 100644 --- a/hmdriver2/hdc.py +++ b/hmdriver2/hdc.py @@ -5,6 +5,7 @@ import shlex import re import os +import signal import subprocess from typing import Union, List, Dict, Tuple, Optional @@ -28,14 +29,27 @@ def _execute_command(cmdargs: Union[str, List[str]], logger.debug(cmdline) try: - process = subprocess.Popen(cmdline, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, shell=True) + popen_kwargs = dict( + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + shell=True, + ) + # Isolate timed commands so a hung hdc child (fport rm) can be killed + # without waiting for grandchildren after the shell exits. + if timeout is not None: + if os.name == "nt": + popen_kwargs["creationflags"] = getattr( + subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + else: + popen_kwargs["start_new_session"] = True + + process = subprocess.Popen(cmdline, **popen_kwargs) try: output, error = process.communicate(timeout=timeout) except subprocess.TimeoutExpired: - process.kill() + _kill_process_group(process) try: - process.communicate() + process.communicate(timeout=1) except Exception: pass return CommandResult("", "timeout expired", -1) @@ -52,6 +66,19 @@ def _execute_command(cmdargs: Union[str, List[str]], return CommandResult("", str(e), -1) +def _kill_process_group(process: subprocess.Popen) -> None: + try: + if os.name == "nt": + process.kill() + else: + os.killpg(process.pid, signal.SIGKILL) + except Exception: + try: + process.kill() + except Exception: + pass + + def _build_hdc_prefix() -> str: """ Construct the hdc command prefix based on environment variables.