feat: invoke with exception

Author: codematrixer

hmdriver2/_client.py

@@ -6,18 +6,18 @@
 import os
 import typing
 import subprocess
-import atexit
 import hashlib
 from datetime import datetime
 from functools import cached_property
 
 from . import logger
 from .hdc import HdcWrapper
 from .proto import HypiumResponse, DriverData
+from .exception import InvokeHypiumError
 
 
-UITEST_SERICE_PORT = 8012
-SOCKET_TIMEOUT = 60
+UITEST_SERVICE_PORT = 8012
+SOCKET_TIMEOUT = 30
 
 
 class HMClient:
@@ -35,11 +35,11 @@ def local_port(self):
         fports = self.hdc.list_fport()
         logger.debug(fports) if fports else None
 
-        return self.hdc.forward_port(UITEST_SERICE_PORT)
+        return self.hdc.forward_port(UITEST_SERVICE_PORT)
 
     def _rm_local_port(self):
         logger.debug("rm fport local port")
-        self.hdc.rm_forward(self.local_port, UITEST_SERICE_PORT)
+        self.hdc.rm_forward(self.local_port, UITEST_SERVICE_PORT)
 
     def _connect_sock(self):
         """Create socket and connect to the uiTEST server."""
@@ -81,6 +81,20 @@ def _recv_msg(self, buff_size: int = 1024, decode=False) -> typing.Union[bytearr
             return bytearray()
 
     def invoke(self, api: str, this: str = "Driver#0", args: typing.List = []) -> HypiumResponse:
+        """
+        Invokes a given API method with the specified arguments and handles exceptions.
+
+        Args:
+        api (str): The name of the API method to invoke.
+        args (List, optional): A list of arguments to pass to the API method. Default is an empty list.
+
+        Returns:
+        HypiumResponse: The response from the API call.
+
+        Raises:
+        InvokeHypiumError: If the API call returns an exception in the response.
+        """
+
         msg = {
             "module": "com.ohos.devicetest.hypiumApiHelper",
             "method": "callHypiumApi",
@@ -93,11 +107,11 @@ def invoke(self, api: str, this: str = "Driver#0", args: typing.List = []) -> Hy
             "request_id": datetime.now().strftime("%Y%m%d%H%M%S%f")
         }
         self._send_msg(msg)
-        result = self._recv_msg(decode=True)
-        # TODO handle exception
-        # {"exception":{"code":401,"message":"(PreProcessing: APiCallInfoChecker)Illegal argument count"}}
-
-        return HypiumResponse(**(json.loads(result)))
+        raw_data = self._recv_msg(decode=True)
+        data = HypiumResponse(**(json.loads(raw_data)))
+        if data.exception:
+            raise InvokeHypiumError(data.exception)
+        return data
 
     def start(self):
         logger.info("Start client connection")

hmdriver2/_toast.py

@@ -4,8 +4,8 @@
 
 
 class ToastWatcher:
-    def __init__(self, session: "Driver"):  # type: ignore
-        self.session = session
+    def __init__(self, driver: "Driver"):  # type: ignore
+        self.driver = driver
 
     def start(self) -> bool:
         """
@@ -15,10 +15,10 @@ def start(self) -> bool:
             bool: True if the observer starts successfully, else False.
         """
         api = "Driver.uiEventObserverOnce"
-        resp: HypiumResponse = self.session._invoke(api, args=["toastShow"])
+        resp: HypiumResponse = self.driver._invoke(api, args=["toastShow"])
         return resp.result
 
-    def get(self, timeout: int = 5) -> str:
+    def get(self, timeout: int = 3) -> str:
         """
         Read the latest toast message content from the recent period.
 
@@ -29,7 +29,7 @@ def get(self, timeout: int = 5) -> str:
             str: The content of the latest toast message.
         """
         api = "Driver.getRecentUiEvent"
-        resp: HypiumResponse = self.session._invoke(api, args=[timeout])
+        resp: HypiumResponse = self.driver._invoke(api, args=[timeout])
         if resp.result:
             return resp.result.get("text")
         return None

hmdriver2/driver.py

@@ -10,11 +10,11 @@
 except ImportError:
     from cached_property import cached_property
 
-from .exception import DeviceNotFoundError
 from ._client import HMClient
 from ._uiobject import UiObject
 from .hdc import list_devices
 from ._toast import ToastWatcher
+from .exception import DeviceNotFoundError
 from .proto import HypiumResponse, KeyCode, Point, DisplayRotation, DeviceInfo
 
 
@@ -50,6 +50,9 @@ def _is_device_online(self):
         _serials = list_devices()
         return True if self.serial in _serials else False
 
+    def _invoke(self, api: str, args: List = []) -> HypiumResponse:
+        return self._client.invoke(api, this=self._this_driver, args=args)
+
     def start_app(self, package_name: str, page_name: str = "MainAbility"):
         self.unlock()
         self.hdc.start_app(package_name, page_name)
@@ -103,9 +106,6 @@ def unlock(self):
         self.hdc.swipe(0.5 * w, 0.8 * h, 0.5 * w, 0.2 * h)
         time.sleep(.5)
 
-    def _invoke(self, api: str, args: List = []) -> HypiumResponse:
-        return self._client.invoke(api, this=self._this_driver, args=args)
-
     @cached_property
     def display_size(self) -> Tuple[int, int]:
         api = "Driver.getDisplaySize"

hmdriver2/exception.py

@@ -8,17 +8,17 @@ class ElementFoundTimeout(Exception):
     pass
 
 
-class HDriverError(Exception):
+class HmDriverError(Exception):
     pass
 
 
 class DeviceNotFoundError(Exception):
     pass
 
 
-class HDCException(Exception):
+class HdcError(Exception):
     pass
 
 
-class AttributeException(Exception):
+class InvokeHypiumError(Exception):
     pass

hmdriver2/hdc.py

@@ -76,7 +76,7 @@ def forward_port(self, rport: int) -> int:
     def rm_forward(self, lport: int, rport: int) -> int:
         result = _execute_command(f"hdc -t {self.serial} fport rm tcp:{lport} tcp:{rport}")
         if result.exit_code != 0:
-            raise RuntimeError("HDC forward port error", result.output)
+            raise RuntimeError("HDC rm forward error", result.output)
         return lport
 
     def list_fport(self) -> List:

hmdriver2/proto.py

@@ -41,7 +41,15 @@ class DeviceInfo:
 
 @dataclass
 class HypiumResponse:
-    result: Union[List, bool, str, None]  # {"result":"Driver#0"}
+    """
+    Example:
+    {"result":"On#1"}
+    {"result":null}
+    {"result":null,"exception":"Can not connect to AAMS, RET_ERR_CONNECTION_EXIST"}
+    {"exception":{"code":401,"message":"(PreProcessing: APiCallInfoChecker)Illegal argument count"}}
+    """
+    result: Union[List, bool, str, None] = None
+    exception: Union[List, bool, str, None] = None
 
 
 @dataclass