feat: add dump_hierarchy and ToastWatcher

Author: codematrixer

.flake8

@@ -5,4 +5,5 @@ ignore =
     F401
     E402
     W292
-    F403
+    F403
+    F821

.github/workflows/release.yml

@@ -1,8 +1,12 @@
 name: Release
 
+# on:
+#   release:
+#     types: [created]
 on:
-  release:
-    types: [created]
+  push:
+    tags:
+      - '*.*.*'
 
 jobs:
   build-n-publish:
@@ -21,6 +25,15 @@ jobs:
         pip install setuptools wheel twine poetry
         poetry install
 
+    - name: Get the version from the tag
+      id: get_version
+      run: echo "::set-output name=VERSION::${GITHUB_REF/refs\/tags\//}"
+
+    - name: Update version in pyproject.toml
+      run: |
+        version=${{ steps.get_version.outputs.VERSION }}
+        poetry version $version
+
     - name: Build a binary wheel and a source tarball
       run: poetry build
 

README.md

@@ -63,7 +63,7 @@ d.screenshot(lpath)
 
 # Device touch
 d.click(500, 1000)
-d.click(0.5, 0.4)
+d.click(0.5, 0.4)  # "If of type float, it represents percentage coordinates."
 d.double_click(500, 1000)
 d.double_click(0.5, 0.4)
 d.long_click(500, 1000)
@@ -100,21 +100,28 @@ d(text="showToast").info
 # }
 
 d(id="swiper").exists()
-d(type="Button", text="showToast").exists()
-d(text="showToast", isAfter=True).exists()
-d(text="showToast").click_if_exists()
+d(type="Button", text="tab_recrod").exists()
+d(text="tab_recrod", isAfter=True).exists()
+d(text="tab_recrod").click_if_exists()
 d(type="Button", index=3).click()
-d(text="showToast").double_click()
-d(text="showToast").long_click()
+d(text="tab_recrod").double_click()
+d(text="tab_recrod").long_click()
 
 component: ComponentData = d(type="ListItem", index=1).find_component()
 d(type="ListItem").drag_to(component)
 
-d(text="showToast").input_text("abc")
-d(text="showToast").clear_text()
-d(text="showToast").pinch_in()
-d(text="showToast").pinch_out()
+d(text="tab_recrod").input_text("abc")
+d(text="tab_recrod").clear_text()
+d(text="tab_recrod").pinch_in()
+d(text="tab_recrod").pinch_out()
 
+# Dump hierarchy
+d.dump_hierarchy()
+
+# Toast Watcher
+d.toast_watcher.start()
+d(type="Button", text="tab_recrod").click()  # 触发toast的操作
+toast = d.toast_watcher.get()
 
 ```
 

docs/DEVELOP.md

@@ -168,6 +168,27 @@
 {"result":"UiWindow#10"}
 ```
 
+### uiEventObserverOnce
+**send**
+```
+{"module":"com.ohos.devicetest.hypiumApiHelper","method":"callHypiumApi","params":{"api":"Driver.uiEventObserverOnce","this":"Driver#0","args":["toastShow"],"message_type":"hypium"},"request_id":"20240905144543056211","client":"127.0.0.1"}
+```
+**recv**
+```
+{"result":true}
+```
+
+
+### getRecentUiEvent
+**send**
+```
+{"module":"com.ohos.devicetest.hypiumApiHelper","method":"callHypiumApi","params":{"api":"Driver.getRecentUiEvent","this":"Driver#0","args":[3000],"message_type":"hypium"},"request_id":"20240905143857794307","client":"127.0.0.1"}
+```
+**recv**
+```
+{"result":{"bundleName":"com.samples.test.uitest","text":"testMessage","type":"Toast"}}
+```
+
 
 ## Component
 ### Component.getId

hmdriver2/_client.py

@@ -64,15 +64,15 @@ def _send_msg(self, msg: typing.Dict):
             }
         """
         msg = json.dumps(msg, ensure_ascii=False, separators=(',', ':'))
-        logger.debug(f"send msg: {msg}")
+        logger.debug(f"sendMsg: {msg}")
         self.sock.sendall(msg.encode('utf-8') + b'\n')
 
     def _recv_msg(self, buff_size: int = 1024, decode=False) -> typing.Union[bytearray, str]:
         try:
             relay = self.sock.recv(buff_size)
             if decode:
                 relay = relay.decode()
-            logger.debug(f"recv msg: {relay}")
+            logger.debug(f"recvMsg: {relay}")
             return relay
         except (socket.timeout, UnicodeDecodeError) as e:
             logger.warning(e)
@@ -100,26 +100,13 @@ def invoke(self, api: str, this: str = "Driver#0", args: typing.List = []) -> Hy
         return HypiumResponse(**(json.loads(result)))
 
     def start(self):
-        """
-
-        Args:
-            file_path (str): Path where the recorded video will be saved.
-
-        Raises:
-            RuntimeError: If the screen capture fails to start.
-        """
         logger.info("Start client connection")
         self._init_so_resource()
         self._restart_uitest_service()
 
         self._connect_sock()
 
     def release(self):
-        """
-
-        Raises:
-            RuntimeError: If the screen capture fails to start.
-        """
         logger.info("Release client connection")
         try:
             if self.sock:
@@ -176,6 +163,7 @@ def __get_local_md5sum(f: str) -> str:
     def _restart_uitest_service(self):
         """
         Restart the UITest daemon.
+
         Note: 'hdc shell aa test' will also start a uitest daemon.
         $ hdc shell ps -ef |grep uitest
         shell        44306     1 25 11:03:37 ?    00:00:16 uitest start-daemon singleness

hmdriver2/_toast.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+
+from .proto import HypiumResponse
+
+
+class ToastWatcher:
+    def __init__(self, session: "Driver"):  # type: ignore
+        self.session = session
+
+    def start(self) -> bool:
+        """
+        Initiates the observer to listen for a UI toast event
+
+        Returns:
+            bool: True if the observer starts successfully, else False.
+        """
+        api = "Driver.uiEventObserverOnce"
+        resp: HypiumResponse = self.session._invoke(api, args=["toastShow"])
+        return resp.result
+
+    def get(self, timeout: int = 5) -> str:
+        """
+        Read the latest toast message content from the recent period.
+
+        Args:
+            timeout (int): The maximum time to wait for a toast to appear if there are no matching toasts within the given time.
+
+        Returns:
+            str: The content of the latest toast message.
+        """
+        api = "Driver.getRecentUiEvent"
+        resp: HypiumResponse = self.session._invoke(api, args=[timeout])
+        if resp.result:
+            return resp.result.get("text")
+        return None