Merge pull request #1221 from AirtestProject/v1.3.5

v.1.3.5: Add iOS file api

airtest/aircv/screen_recorder.py

@@ -9,6 +9,7 @@
 import time
 import numpy as np
 import subprocess
+import traceback
 
 
 RECORDER_ORI = {
@@ -102,9 +103,16 @@ def write(self, frame):
         self.writer.write(frame.astype(np.uint8))
 
     def close(self):
-        self.writer.close()
-        self.process.wait()
-        self.process.terminate()
+        try:
+            self.writer.close()
+            self.process.wait(timeout=5)
+        except Exception as e:
+            print(f"Error closing ffmpeg process: {e}")
+        finally:
+            try:
+                self.process.terminate()
+            except Exception as e:
+                print(f"Error terminating ffmpeg process: {e}")
 
 
 class ScreenRecorder:
@@ -160,6 +168,7 @@ def stop(self):
         self._stop_flag = True
         self.t_write.join()
         self.t_stream.join()
+        self.writer.close()  # Ensure writer is closed
 
     def get_frame_loop(self):
         # 单独一个线程持续截图
@@ -177,15 +186,18 @@ def get_frame_loop(self):
             raise
 
     def write_frame_loop(self):
-        # 按帧率间隔获取图像写入视频
         try:
             duration = 1.0/self.writer.fps
             last_time = time.time()
             self._stop_flag = False
             while True:
                 if time.time()-last_time >= duration:
                     last_time += duration
-                    self.writer.write(self.tmp_frame)
+                    try:
+                        self.writer.write(self.tmp_frame)
+                    except Exception as e:
+                        print(f"Error writing frame: {e}")
+                        break
                 if self.is_stop():
                     break
                 time.sleep(0.0001)
@@ -194,4 +206,5 @@ def write_frame_loop(self):
         except Exception as e:
             print("write thread error", e)
             self._stop_flag = True
+            self.writer.close()  # Ensure the writer is closed on error
             raise

airtest/core/android/adb.py

@@ -25,6 +25,7 @@
 from airtest.utils.snippet import get_std_encoding, reg_cleanup, split_cmd, make_file_executable
 
 LOGGING = get_logger(__name__)
+TMP_PATH = "/data/local/tmp"  # Android's temporary file directory
 
 
 class ADB(object):
@@ -470,16 +471,10 @@ def sdk_version(self):
 
     def push(self, local, remote):
         """
-        Perform `adb push` command
-
-        Note:
-            If there is a space (or special symbol) in the file name, it will be forced to add escape characters,
-            and the new file name will be added with quotation marks and returned as the return value
-
-            注意：文件名中如果带有空格（或特殊符号），将会被强制增加转义符，并将新的文件名添加引号，作为返回值返回
+        Push file or folder to the specified directory to the device
 
         Args:
-            local: local file to be copied to the device
+            local: local file or folder to be copied to the device
             remote: destination on the device where the file will be copied
 
         Returns:
@@ -495,18 +490,65 @@ def push(self, local, remote):
             "/data/local/tmp/test\ space.txt"
             >>> adb.shell("rm " + new_name)
 
+            >>> adb.push("test_dir", "/sdcard/Android/data/com.test.package/files")
+            >>> adb.push("test_dir", "/sdcard/Android/data/com.test.package/files/test_dir")
+
         """
-        local = decode_path(local)  # py2
-        if os.path.isfile(local) and os.path.splitext(local)[-1] != os.path.splitext(remote)[-1]:
-            # If remote is a folder, add the filename and escape
-            filename = os.path.basename(local)
-            # Add escape characters for spaces, parentheses, etc. in filenames
-            filename = re.sub(r"[ \(\)\&]", lambda m: "\\" + m.group(0), filename)
-            remote = '%s/%s' % (remote, filename)
-        self.cmd(["push", local, remote], ensure_unicode=False)
-        return '\"%s\"' % remote
+        _, ext = os.path.splitext(remote)
+        if ext:
+            # The target path is a file
+            dst_parent = os.path.dirname(remote)
+        else:
+            dst_parent = remote
 
-    def pull(self, remote, local):
+        # If the target file already exists, delete it first to avoid overwrite failure
+        src_filename = os.path.basename(local)
+        _, src_ext = os.path.splitext(local)
+        if src_ext:
+            dst_path = f"{dst_parent}/{src_filename}"
+        else:
+            if src_filename == os.path.basename(remote):
+                dst_path = remote
+            else:
+                dst_path = f"{dst_parent}/{src_filename}"
+        try:
+            self.shell(f"rm -r {dst_path}")
+        except:
+            pass
+
+        # If the target folder has multiple levels that have never been created, try to create them
+        try:
+            self.shell(f"mkdir -p {dst_parent}")
+        except:
+            pass
+
+        # Push the file to the tmp directory to avoid permission issues
+        tmp_path = f"{TMP_PATH}/{src_filename}"
+        try:
+            self.cmd(["push", local, tmp_path])
+        except:
+            self.cmd(["push", local, dst_parent])
+        else:
+            try:
+                if src_ext:
+                    try:
+                        self.shell(f'mv "{tmp_path}" "{remote}"')
+                    except:
+                        self.shell(f'mv "{tmp_path}" "{remote}"')
+                else:
+                    try:
+                        self.shell(f'cp -frp "{tmp_path}/*" "{remote}"')
+                    except:
+                        self.shell(f'mv "{tmp_path}" "{remote}"')
+            finally:
+                try:
+                    if TMP_PATH != dst_parent:
+                        self.shell(f'rm -r "{tmp_path}"')
+                except:
+                    pass
+        return dst_path
+
+    def pull(self, remote, local=""):
         """
         Perform `adb pull` command
 
@@ -521,6 +563,8 @@ def pull(self, remote, local):
         Returns:
             None
         """
+        if not local:
+            local = os.path.basename(remote)
         local = decode_path(local)  # py2
         if PY3:
             # If python3, use Path to force / convert to \
@@ -921,8 +965,8 @@ def exists_file(self, filepath):
 
         """
         try:
-            out = self.shell(["ls", filepath])
-        except AdbShellError:
+            out = self.shell("ls \"%s\"" % filepath)
+        except (AdbShellError, AdbError):
             return False
         else:
             return not ("No such file or directory" in out)

airtest/core/android/android.py

@@ -1030,6 +1030,41 @@ def set_clipboard(self, text):
         """
         self.yosemite_ext.set_clipboard(text)
 
+    def push(self, local, remote):
+        """
+        Push file to the device
+
+        Args:
+            local: local file or folder to be copied to the device
+            remote: destination on the device where the file will be copied
+
+        Returns:
+            The file path saved in the phone may be enclosed in quotation marks, eg. '"test\ file.txt"'
+
+        Examples:
+            >>> dev = connect_device("android:///")
+            >>> dev.push("test.txt", "/sdcard/test.txt")
+
+        """
+        return self.adb.push(local, remote)
+
+    def pull(self, remote, local=""):
+        """
+        Pull file from the device
+
+        Args:
+            remote: remote file to be downloaded from the device
+            local: local destination where the file will be downloaded from the device, if not specified, the current directory is used
+
+        Returns:
+            None
+
+        Examples:
+            >>> dev = connect_device("android:///")
+            >>> dev.pull("/sdcard/test.txt", "rename.txt")
+        """
+        return self.adb.pull(remote, local=local)
+
     def _register_rotation_watcher(self):
         """
         Register callbacks for Android and minicap when rotation of screen has changed

airtest/core/api.py

@@ -724,6 +724,58 @@ def paste(*args, **kwargs):
     G.DEVICE.paste(*args, **kwargs)
 
 
+@logwrap
+def push(local, remote, *args, **kwargs):
+    """
+    Push file from local to remote
+
+    :param local: local file path
+    :param remote: remote file path
+    :return: filename of the pushed file
+    :platforms: Android, iOS
+    :Example:
+
+        Android::
+
+            >>> connect_device("android:///")
+            >>> push(r"D:\demo\test.text", "/data/local/tmp/test.text")
+
+
+        iOS::
+
+            >>> connect_device("iOS:///http+usbmux://udid")
+            >>> push("test.png", "/DCIM/")  # push to the DCIM directory
+            >>> push(r"D:\demo\test.text", "/Documents", bundle_id="com.apple.Keynote")  # push to the Documents directory of the Keynote app
+
+    """
+    return G.DEVICE.push(local, remote, *args, **kwargs)
+
+
+@logwrap
+def pull(remote, local, *args, **kwargs):
+    """
+    Pull file from remote to local
+
+    :param remote: remote file path
+    :param local: local file path
+    :return: filename of the pulled file
+    :platforms: Android, iOS
+    :Example:
+
+        Android::
+
+            >>> connect_device("android:///")
+            >>> pull("/data/local/tmp/test.txt", r"D:\demo\test.txt")
+
+        iOS::
+
+            >>> connect_device("iOS:///http+usbmux://udid")
+            >>> pull("/DCIM/test.png", r"D:\demo\test.png")
+            >>> pull("/Documents/test.key", r"D:\demo\test.key", bundle_id="com.apple.Keynote")
+
+    """
+    return G.DEVICE.pull(remote, local, *args, **kwargs)
+
 """
 Assertions: see airtest/core/assertions.py
 """