Add pluggable clipboard system and ADB app update method

airtest/core/android/adb.py

@@ -14,7 +14,7 @@
 from six.moves import reduce
 
 from airtest.core.android.constant import (DEFAULT_ADB_PATH, IP_PATTERN,
-                                           SDK_VERISON_ANDROID7)
+                                           SDK_VERSION_ANDROID7)
 from airtest.core.error import (AdbError, AdbShellError, AirtestError,
                                 DeviceConnectionError)
 from airtest.utils.compat import decode_path, raisefrom, proc_communicate_timeout, SUBPROCESS_FLAG
@@ -393,7 +393,7 @@ def shell(self, cmd):
             command output
 
         """
-        if self.sdk_version < SDK_VERISON_ANDROID7:
+        if self.sdk_version < SDK_VERSION_ANDROID7:
             # for sdk_version < 25, adb shell do not raise error
             # https://stackoverflow.com/questions/9379400/adb-error-codes
             cmd = split_cmd(cmd) + [";", "echo", "---$?---"]
@@ -831,6 +831,18 @@ def pm_install(self, filepath, replace=False, install_options=None):
         finally:
             # delete apk file
             self.cmd(["shell", "rm", device_path], timeout=30)
+
+    def pm_update_app(self, filepath, package):
+        apk_version = int(APK(filepath).androidversion_code)
+        installed_version = self.get_package_version(package)
+        if installed_version is None or apk_version > int(installed_version):
+            LOGGING.info(
+                "local version code is {}, installed version code is {}".format(apk_version, installed_version))
+            try:
+                self.pm_install(filepath, replace=True, install_options=["-t"])
+            except Exception as e:
+                LOGGING.error(f"Failed to install {package}: {e}")
+
 
     def uninstall_app(self, package):
         """
@@ -1019,7 +1031,7 @@ def line_breaker(self):
 
         """
         if not self._line_breaker:
-            if self.sdk_version >= SDK_VERISON_ANDROID7:
+            if self.sdk_version >= SDK_VERSION_ANDROID7:
                 line_breaker = os.linesep
             else:
                 line_breaker = '\r' + os.linesep

airtest/core/android/android.py

@@ -6,11 +6,11 @@
 import warnings
 from copy import copy
 from airtest import aircv
+from airtest.core.android.clipboard import CLIPBOARD_MAP, ClipperClipboard
 from airtest.core.device import Device
 from airtest.core.android.ime import YosemiteIme
-from airtest.core.android.yosemite_ext import YosemiteExt
-from airtest.core.android.constant import CAP_METHOD, TOUCH_METHOD, IME_METHOD, ORI_METHOD, \
-    SDK_VERISON_ANDROID10
+from airtest.core.android.constant import CAP_METHOD, CLIPBOARD_METHOD, TOUCH_METHOD, IME_METHOD, ORI_METHOD, \
+   SDK_VERSION_ANDROID10 
 from airtest.core.android.adb import ADB
 
 from airtest.core.android.rotation import RotationWatcher, XYTransformer
@@ -41,6 +41,7 @@ def __init__(self, serialno=None, host=None,
                  touch_method=TOUCH_METHOD.MINITOUCH,
                  ime_method=IME_METHOD.YOSEMITEIME,
                  ori_method=ORI_METHOD.MINICAP,
+                 clipboard_method=CLIPBOARD_METHOD.YOSEMITE,
                  display_id=None,
                  input_event=None,
                  adb_path=None,
@@ -52,21 +53,22 @@ def __init__(self, serialno=None, host=None,
         self._touch_method = touch_method.upper()
         self.ime_method = ime_method.upper()
         self.ori_method = ori_method.upper()
+        self.clipboard_method = clipboard_method.upper()
         self.display_id = display_id
         self.input_event = input_event
         # init adb
         self.adb = ADB(self.serialno, adb_path=adb_path, server_addr=host, display_id=self.display_id, input_event=self.input_event)
         self.adb.wait_for_device()
         self.sdk_version = self.adb.sdk_version
-        if self.sdk_version >= SDK_VERISON_ANDROID10 and self._touch_method == TOUCH_METHOD.MINITOUCH:
+        if self.sdk_version >= SDK_VERSION_ANDROID10 and self._touch_method == TOUCH_METHOD.MINITOUCH:
             self._touch_method = TOUCH_METHOD.MAXTOUCH
         self._display_info = {}
         self._current_orientation = None
         # init components
         self.rotation_watcher = RotationWatcher(self.adb, self.ori_method)
         self.yosemite_ime = YosemiteIme(self.adb)
         self.yosemite_recorder = Recorder(self.adb)
-        self.yosemite_ext = YosemiteExt(self.adb)
+        self.clipboard = CLIPBOARD_MAP[self.clipboard_method](self.adb)
         self._register_rotation_watcher()
 
         self._touch_proxy = None
@@ -1018,7 +1020,7 @@ def get_clipboard(self):
             >>> dev.paste()  # paste the clipboard content
 
         """
-        return self.yosemite_ext.get_clipboard()
+        return self.clipboard.get_clipboard()
 
     def set_clipboard(self, text):
         """
@@ -1031,7 +1033,7 @@ def set_clipboard(self, text):
             None
 
         """
-        self.yosemite_ext.set_clipboard(text)
+        self.clipboard.set_clipboard(text)
 
     def push(self, local, remote):
         """

airtest/core/android/cap_methods/adbcap.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 import warnings
 from airtest.core.android.cap_methods.base_cap import BaseCap
-from airtest.core.android.constant import SDK_VERISON_ANDROID7
+from airtest.core.android.constant import SDK_VERSION_ANDROID7
 from airtest import aircv
 
 
@@ -12,6 +12,6 @@ def get_frame_from_stream(self):
 
     def snapshot(self, ensure_orientation=True):
         screen = super(AdbCap, self).snapshot()
-        if ensure_orientation and self.adb.sdk_version <= SDK_VERISON_ANDROID7:
+        if ensure_orientation and self.adb.sdk_version <= SDK_VERSION_ANDROID7:
             screen = aircv.rotate(screen, self.adb.display_info["orientation"] * 90, clockwise=False)
         return screen

airtest/core/android/clipboard.py

@@ -0,0 +1,102 @@
+import re
+from time import sleep
+from airtest.core.error import AirtestError
+from airtest.utils.snippet import escape_special_char
+from airtest.core.android.constant import CLIPPER_APK, CLIPPER_PACKAGE, YOSEMITE_APK, YOSEMITE_PACKAGE,SDK_VERSION_ANDROID10, CLIPBOARD_METHOD
+from airtest.utils.logger import get_logger
+
+LOGGING = get_logger(__name__)
+
+class Clipboard(object):
+    """
+    Base class for clipboard implementations.
+    Subclass this and implement get_clipboard/set_clipboard for each backend.
+    """
+    def __init__(self, adb):
+        self.adb = adb
+
+    def get_clipboard(self):
+        raise NotImplementedError
+
+    def set_clipboard(self, text):
+        raise NotImplementedError
+
+
+# Clipper: android clipboard access via broadcast intent
+# https://github.com/amirshams8/clipper
+class ClipperClipboard(Clipboard):
+    def __init__(self, adb):
+        super(ClipperClipboard, self).__init__(adb)
+        self.adb.pm_update_app(CLIPPER_APK, CLIPPER_PACKAGE)
+
+        # background execution is sufficient for Android 9 and below.
+        if self.is_supported_background():
+            self.shell("am start -n ca.zgrs.clipper/.Main")
+            sleep(0.5)
+            self.close_clipper()
+
+    def get_clipboard(self):
+        if not self.is_supported_background():
+            # clipboard access requires foreground app 
+            self.shell("am start -n ca.zgrs.clipper/.Main")
+            sleep(0.5)
+
+        #text type : str 
+        #text example: Broadcasting: Intent { act=clipper.get flg=0x400000 cmp=ca.zgrs.clipper/.ClipperReceiver }Broadcast completed: result=-1, data="hello"
+        clipboard_data = self.shell(f"am broadcast -a clipper.get -n ca.zgrs.clipper/.ClipperReceiver") 
+        if not self.is_supported_background():
+            self.close_clipper()
+
+        text = re.search(r'data="(.*?)"',clipboard_data)
+        if text:
+            return text.group(1)
+        return ""
+
+    def set_clipboard(self, text):
+        text = escape_special_char(text)
+        self.shell(f"am broadcast -a clipper.set -e text \"{text}\" -n ca.zgrs.clipper/.ClipperReceiver")
+
+    def close_clipper(self):
+        #focused type : str
+        #focused example : mFocusedApp=ActivityRecord{55b302a u0 ca.zgrs.clipper/.Main t2016}
+        focused = self.shell("dumpsys window | grep -E 'mFocusedApp'")
+        if CLIPPER_PACKAGE in focused:
+            self.shell("input keyevent KEYCODE_BACK")
+
+    def is_supported_background(self):
+        supported_background = self.adb.sdk_version < SDK_VERSION_ANDROID10
+        return supported_background
+
+    def shell(self, cmd):
+        return self.adb.shell(cmd)
+
+class YosemiteClipboard(Clipboard):
+    def __init__(self, adb):
+        super(YosemiteClipboard, self).__init__(adb)
+        self.adb.pm_update_app(YOSEMITE_APK, YOSEMITE_PACKAGE)
+
+    def get_clipboard(self):
+        #text type : str
+        #text example : hello
+        text = self.adb.shell(f"app_process -Djava.class.path={self.adb.path_app(YOSEMITE_PACKAGE)} / com.netease.nie.yosemite.control.Control --DEVICE_OP clipboard_get")
+        if text:
+            return text.strip()
+        return ""
+
+    def set_clipboard(self, text):
+        text = escape_special_char(text)
+        try:
+            ret = self.adb.shell(f"app_process -Djava.class.path={self.adb.path_app(YOSEMITE_PACKAGE)} / com.netease.nie.yosemite.control.Control --DEVICE_OP clipboard --TEXT \"{text}\"")
+        except Exception as e:
+            raise AirtestError("set clipboard failed, %s" % repr(e))
+        else:
+            if ret and "Exception" in ret:
+                raise AirtestError("set clipboard failed: %s" % ret)
+
+
+# maps ime_method to clipboard implementation
+CLIPBOARD_MAP = {
+    CLIPBOARD_METHOD.CLIPPER: ClipperClipboard,
+    CLIPBOARD_METHOD.YOSEMITE: YosemiteClipboard,
+}
+

airtest/core/android/constant.py

@@ -15,16 +15,18 @@
     "Linux-armv7l": os.path.join(STATICPATH, "adb", "linux_arm", "adb"),
 }
 DEFAULT_ADB_SERVER = ('127.0.0.1', 5037)
-SDK_VERISON_ANDROID7 = 24
+SDK_VERSION_ANDROID7 = 24
 # Android 10 SDK version
-SDK_VERISON_ANDROID10 = 29
+SDK_VERSION_ANDROID10 = 29
 DEBUG = True
 STFLIB = os.path.join(STATICPATH, "stf_libs")
 ROTATIONWATCHER_APK = os.path.join(STATICPATH, "apks", "RotationWatcher.apk")
 ROTATIONWATCHER_PACKAGE = "jp.co.cyberagent.stf.rotationwatcher"
 YOSEMITE_APK = os.path.join(STATICPATH, "apks", "Yosemite.apk")
 YOSEMITE_PACKAGE = 'com.netease.nie.yosemite'
 YOSEMITE_IME_SERVICE = 'com.netease.nie.yosemite/.ime.ImeService'
+CLIPPER_APK = os.path.join(STATICPATH, "apks", "Clipper.apk")
+CLIPPER_PACKAGE = "ca.zgrs.clipper"
 MAXTOUCH_JAR = os.path.join(STATICPATH, "apks", "maxpresent.jar")
 ROTATIONWATCHER_JAR = os.path.join(STATICPATH, "apks", "rotationwatcher.jar")
 IP_PATTERN = re.compile(r'(\d+\.){3}\d+')
@@ -50,3 +52,7 @@ class IME_METHOD(object):
 class ORI_METHOD(object):
     ADB = "ADBORI"
     MINICAP = "MINICAPORI"
+
+class CLIPBOARD_METHOD(object):
+    CLIPPER = "CLIPPER"
+    YOSEMITE = "YOSEMITE"

airtest/core/android/yosemite.py

@@ -21,33 +21,7 @@ def install_or_upgrade(self):
             None
 
         """
-        self._install_apk_upgrade(YOSEMITE_APK, YOSEMITE_PACKAGE)
-
-    def _install_apk_upgrade(self, apk_path, package):
-        """
-        Install or update the `.apk` file on the device
-
-        Args:
-            apk_path: full path `.apk` file
-            package: package name
-
-        Returns:
-            None
-
-        """
-        apk_version = int(APK(apk_path).androidversion_code)
-        installed_version = self.adb.get_package_version(package)
-        if installed_version is None or apk_version > int(installed_version):
-            LOGGING.info(
-                "local version code is {}, installed version code is {}".format(apk_version, installed_version))
-            try:
-                self.adb.pm_install(apk_path, replace=True, install_options=["-t"])
-            except:
-                if installed_version is None:
-                    raise
-                # If the installation fails, but the phone has an old version, do not force the installation
-                print(traceback.format_exc())
-                warnings.warn("Yosemite.apk update failed, please try to reinstall manually(airtest/core/android/static/apks/Yosemite.apk).")
+        self.adb.pm_update_app(YOSEMITE_APK, YOSEMITE_PACKAGE)
 
     @on_method_ready('install_or_upgrade')
     def get_ready(self):

airtest/core/android/yosemite_ext.py

@@ -1,9 +1,6 @@
-import re
-
-from .constant import YOSEMITE_APK, YOSEMITE_PACKAGE
+from .constant import YOSEMITE_PACKAGE
 from airtest.core.android.yosemite import Yosemite
-from airtest.core.error import AirtestError
-from airtest.utils.snippet import on_method_ready, escape_special_char
+from airtest.utils.snippet import on_method_ready
 from airtest.utils.logger import get_logger
 LOGGING = get_logger(__name__)
 
@@ -35,40 +32,6 @@ def device_op(self, op_name, op_args=""):
         """
         return self.adb.shell(f"app_process -Djava.class.path={self.path} / com.netease.nie.yosemite.control.Control --DEVICE_OP {op_name} {op_args}")
 
-    def get_clipboard(self):
-        """
-        Get clipboard content
-
-        Returns:
-            clipboard content
-
-        """
-        text = self.device_op("clipboard_get")
-        if text:
-            return text.strip()
-        return ""
-
-    def set_clipboard(self, text):
-        """
-        Set clipboard content
-
-        Args:
-            text: text to be set
-
-        Returns:
-            None
-
-        """
-        text = escape_special_char(text)
-
-        try:
-            ret = self.device_op("clipboard", f'--TEXT {text}')
-        except Exception as e:
-            raise AirtestError("set clipboard failed, %s" % repr(e))
-        else:
-            if ret and "Exception" in ret:
-                raise AirtestError("set clipboard failed: %s" % ret)
-
     def change_lang(self, lang):
         """
         Change language

tests/test_clipboard.py

@@ -0,0 +1,39 @@
+# encoding=utf-8
+import unittest
+from airtest.core.android.android import ADB
+from airtest.core.android.clipboard import ClipperClipboard, YosemiteClipboard
+
+class TestClipboard(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.adb = ADB()
+        devices = cls.adb.devices()
+        if not devices:
+            raise RuntimeError("At lease one adb device required")
+        cls.adb.serialno = devices[0][0]
+        cls.yosemite_clipboard = YosemiteClipboard(cls.adb)
+        cls.clipper_clipboard = ClipperClipboard(cls.adb)
+
+    def test_clipboard(self):
+        text1 = "test clipboard"
+        text2 = "test clipboard with $pecial char #@!#%$#^&*()'"
+
+        # test clipper clipboard
+        self.clipper_clipboard.set_clipboard(text1)
+        value = self.clipper_clipboard.get_clipboard()
+        self.assertEqual(value, text1)
+
+        self.clipper_clipboard.set_clipboard(text2)
+        value = self.clipper_clipboard.get_clipboard()
+        self.assertEqual(value, text2)
+
+
+        # test yosemite clipboard
+        self.yosemite_clipboard.set_clipboard(text1)
+        value = self.yosemite_clipboard.get_clipboard()
+        self.assertEqual(value, text1)
+
+        self.yosemite_clipboard.set_clipboard(text2)
+        value = self.yosemite_clipboard.get_clipboard()
+        self.assertEqual(value, text2)