From dfcd2fbb82fe1de3f91ef2c837957834f6aa3464 Mon Sep 17 00:00:00 2001 From: codematrixer Date: Fri, 6 Sep 2024 13:13:01 +0800 Subject: [PATCH] feat: invoke with exception --- hmdriver2/_client.py | 34 ++++++++++++++++++++++++---------- hmdriver2/_toast.py | 10 +++++----- hmdriver2/driver.py | 8 ++++---- hmdriver2/exception.py | 6 +++--- hmdriver2/hdc.py | 2 +- hmdriver2/proto.py | 10 +++++++++- 6 files changed, 46 insertions(+), 24 deletions(-) diff --git a/hmdriver2/_client.py b/hmdriver2/_client.py index 7fccbbb..5ec02cf 100644 --- a/hmdriver2/_client.py +++ b/hmdriver2/_client.py @@ -6,7 +6,6 @@ import os import typing import subprocess -import atexit import hashlib from datetime import datetime from functools import cached_property @@ -14,10 +13,11 @@ 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") diff --git a/hmdriver2/_toast.py b/hmdriver2/_toast.py index d2bdca7..7684490 100644 --- a/hmdriver2/_toast.py +++ b/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 \ No newline at end of file diff --git a/hmdriver2/driver.py b/hmdriver2/driver.py index 27ec2a2..fe9f0fb 100644 --- a/hmdriver2/driver.py +++ b/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" diff --git a/hmdriver2/exception.py b/hmdriver2/exception.py index ac148cb..3e0ae08 100644 --- a/hmdriver2/exception.py +++ b/hmdriver2/exception.py @@ -8,7 +8,7 @@ class ElementFoundTimeout(Exception): pass -class HDriverError(Exception): +class HmDriverError(Exception): pass @@ -16,9 +16,9 @@ class DeviceNotFoundError(Exception): pass -class HDCException(Exception): +class HdcError(Exception): pass -class AttributeException(Exception): +class InvokeHypiumError(Exception): pass diff --git a/hmdriver2/hdc.py b/hmdriver2/hdc.py index bcd7c47..5fe5fc6 100644 --- a/hmdriver2/hdc.py +++ b/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: diff --git a/hmdriver2/proto.py b/hmdriver2/proto.py index 7184556..518f735 100644 --- a/hmdriver2/proto.py +++ b/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