From 2532ddd28877573f9c4737813f20c9898c6e7430 Mon Sep 17 00:00:00 2001 From: codeskyblue Date: Thu, 4 Apr 2024 18:00:08 +0800 Subject: [PATCH] add xpath tests, remove logzero, add enable_pretty_logging --- docs/2to3.md | 23 +++- examples/com.netease.cloudmusic/xpath_test.py | 56 -------- examples/runyaml/run.py | 6 +- pyproject.toml | 5 +- tests/unittests/test_logger.py | 25 ++++ tests/unittests/test_xpath.py | 83 ++++++++++++ uiautomator2/__init__.py | 70 +++++----- uiautomator2/__main__.py | 11 +- uiautomator2/image.py | 9 +- uiautomator2/init.py | 59 ++++----- uiautomator2/watcher.py | 32 ++--- uiautomator2/webview.py | 5 +- uiautomator2/widget.py | 7 +- uiautomator2/xpath.py | 125 +++--------------- 14 files changed, 232 insertions(+), 284 deletions(-) delete mode 100644 examples/com.netease.cloudmusic/xpath_test.py create mode 100644 tests/unittests/test_logger.py create mode 100644 tests/unittests/test_xpath.py diff --git a/docs/2to3.md b/docs/2to3.md index e773bfff..ef1d8622 100644 --- a/docs/2to3.md +++ b/docs/2to3.md @@ -4,10 +4,27 @@ - 版本管理从pbr改为poetry，同时降低依赖库的数量 - Python依赖调整为最低3.8 -- 增加更多的单测代码，让项目变的更稳定 -- 移除一些陈年老代码 +- 增加更多的单测 - minicap,minitouch不再默认安装 +# 增加 +开启库的详细日志的方法 + +```python +from uiautomator2 import enable_pretty_logging +enable_pretty_logging() +``` + ## 移除的功能 - current_app函数移除 -- `d.xpath("...").wait()`从原有的返回 XMLElement|None，改为返回 bool \ No newline at end of file +- `d.xpath("...").wait()`从原有的返回 XMLElement|None，改为返回 bool +- 日志库清理 + - 移除 logzero依赖 + - 移除 property u2.logger + - 移除 property u2.xpath.XPath.logger + - 移除 property d.watcher.debug + - 移除 d.settings["xpath_debug"] + - 移除 function XPath.apply_watch_from_yaml +- connect_usb函数中的init参数移除，这个参数目前没什么用了 + +TODO: atx-agent不在依赖外网下载，直接打包到内部 diff --git a/examples/com.netease.cloudmusic/xpath_test.py b/examples/com.netease.cloudmusic/xpath_test.py deleted file mode 100644 index 1d2bfb88..00000000 --- a/examples/com.netease.cloudmusic/xpath_test.py +++ /dev/null @@ -1,56 +0,0 @@ -# coding: utf-8 -# - -import unittest - -from logzero import logger - -import uiautomator2 as u2 - -# d = u2.connect_usb("3578298f") -d = u2.connect_usb() - -class MusicTestCase(unittest.TestCase): - def setUp(self): - self.package_name = "com.netease.cloudmusic" - d.ext_xpath.global_set({"timeout": 10}) - logger.info("setUp unlock-screen") - # unlock screen - # d.shell("input keyevent WAKEUP") - d.screen_on() - d.shell("input keyevent HOME") - d.swipe(0.1, 0.9, 0.9, 0.1) # swipe to unlock - - def runTest(self): - logger.info("runTest") - d.app_clear(self.package_name) - s = d.session(self.package_name) - s.set_fastinput_ime(True) - - xp = d.ext_xpath - xp._d = s - - # 处理弹窗 - xp.when("跳过").click() - xp.when("允许").click() # 系统弹窗 - # xp.when("@com.tencent.ibg.joox:id/btn_dismiss").click() - - xp("立即体验").click() - logger.info("Search") - xp("搜索").click() - s.send_keys("周杰伦") - s.send_action("search") - self.assertTrue(xp("布拉格广场").wait()) - # xp("@com.tencent.ibg.joox:id/search_area").click() - # xp("@com.tencent.ibg.joox:id/searchItem").click() - # s.send_keys("One Call Away") - # s.send_action("search") - - def tearDown(self): - d.set_fastinput_ime(False) - d.app_stop(self.package_name) - d.screen_off() - -if __name__ == "__main__": - unittest.main() - \ No newline at end of file diff --git a/examples/runyaml/run.py b/examples/runyaml/run.py index 9eba868b..7380ed77 100755 --- a/examples/runyaml/run.py +++ b/examples/runyaml/run.py @@ -3,6 +3,7 @@ # import argparse +import logging import os import re import time @@ -13,6 +14,7 @@ import uiautomator2 as u2 + CLICK = "click" # swipe SWIPE_UP = "swipe_up" @@ -52,7 +54,9 @@ def read_file_content(path: str, mode:str = "r") -> str: def run_step(cf: bunch.Bunch, app: u2.Session, step: str): logger.info("Step: %s", step) oper, body = split_step(step) - logger.debug("parse as: %s %s", oper, body) + logger + + logger = logging.getLogger(__name__).debug("parse as: %s %s", oper, body) if oper == CLICK: app.xpath(body).click() diff --git a/pyproject.toml b/pyproject.toml index e1d7d547..7ebe1fbc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,16 +12,15 @@ include = ["*/assets/*.apk"] python = "^3.8" requests = "*" lxml = "*" -adbutils = ">=2.2" +adbutils = "^2.2.3" retry = ">=0,<1" packaging = ">=20.3" +pillow = "*" # TODO: remove later Deprecated = "*" -logzero = "*" filelock = ">=3,<4" progress = "^1.6" -pillow = "*" [tool.poetry.group.dev.dependencies] pytest = "^8.1.1" diff --git a/tests/unittests/test_logger.py b/tests/unittests/test_logger.py new file mode 100644 index 00000000..f03bff24 --- /dev/null +++ b/tests/unittests/test_logger.py @@ -0,0 +1,25 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""Created on Thu Apr 04 2024 16:57:34 by codeskyblue +""" + +import logging +import pytest +from uiautomator2 import enable_pretty_logging + + +def test_enable_pretty_logging(caplog: pytest.LogCaptureFixture): + logger = logging.getLogger("uiautomator2") + + logger.info("should not be printed") + enable_pretty_logging() + logger.info("hello") + enable_pretty_logging(logging.INFO) + logger.info("world") + logger.debug("should not be printed") + + # Use caplog.text to check the entire log output as a single string + assert "hello" in caplog.text + assert "world" in caplog.text + assert "should not be printed" not in caplog.text \ No newline at end of file diff --git a/tests/unittests/test_xpath.py b/tests/unittests/test_xpath.py new file mode 100644 index 00000000..58583e60 --- /dev/null +++ b/tests/unittests/test_xpath.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""Created on Thu Apr 04 2024 16:41:25 by codeskyblue +""" + +import pytest +from unittest.mock import Mock +from PIL import Image +from uiautomator2.xpath import XMLElement, XPathSelector, XPath, XPathElementNotFoundError + + +mock = Mock() +mock.screenshot.return_value = Image.new("RGB", (1080, 1920), "white") +mock.dump_hierarchy.return_value = """ + + + + + + + +""" + +x = XPath(mock) + +def test_xpath_click(): + x("n1").click() + assert mock.click.called + assert mock.click.call_args[0] == (540, 50) + + mock.click.reset_mock() + assert x("n1").click_exists() == True + assert mock.click.call_args[0] == (540, 50) + + mock.click.reset_mock() + assert x("n3").click_exists(timeout=.1) == False + assert not mock.click.called + + +def test_xpath_exists(): + assert x("n1").exists + assert not x("n3").exists + + +def test_xpath_wait_and_wait_gone(): + assert x("n1").wait() is True + assert x("n3").wait(timeout=.1) is False + + assert x("n3").wait_gone(timeout=.1) is True + assert x("n1").wait_gone(timeout=.1) is False + + +def test_xpath_get(): + assert x("n1").get().text == "n1" + assert x("n2").get().text == "n2" + + with pytest.raises(XPathElementNotFoundError): + x("n3").get(timeout=.1) + + +def test_xpath_all(): + assert len(x("//TextView").all()) == 2 + assert len(x("n3").all()) == 0 + + assert len(x("n1").all()) == 1 + el = x("n1").all()[0] + assert isinstance(el, XMLElement) + assert el.text == "n1" + + +def test_xpath_element(): + el = x("n1").get(timeout=0) + assert el.text == "n1" + assert el.center() == (540, 50) + assert el.screenshot().size == (1080, 100) + assert el.bounds == (0, 0, 1080, 100) + assert el.get_xpath() == "/hierarchy/FrameLayout/TextView[1]" + + mock.click.reset_mock() + el.click() + assert mock.click.called + assert mock.click.call_args[0] == (540, 50) diff --git a/uiautomator2/__init__.py b/uiautomator2/__init__.py index db975d49..adda3a01 100644 --- a/uiautomator2/__init__.py +++ b/uiautomator2/__init__.py @@ -39,14 +39,11 @@ # import progress.bar import adbutils import filelock -import logzero import requests from deprecated import deprecated -from logzero import setup_logger from packaging import version as packaging_version from PIL import Image from retry import retry -from urllib3.util.retry import Retry from . import xpath from ._proto import HTTP_TIMEOUT, SCROLL_STEPS, Direction @@ -54,24 +51,30 @@ from .exceptions import BaseError, ConnectError, GatewayError, JSONRPCError, NullObjectExceptionError, \ NullPointerExceptionError, RetryError, ServerError, SessionBrokenError, StaleObjectExceptionError, \ UiAutomationNotConnectedError, UiObjectNotFoundError -from .init import Initer # from .session import Session # noqa: F401 from .settings import Settings from .swipe import SwipeExt -from .utils import list2cmdline, process_safe_wrapper, thread_safe_wrapper +from .utils import list2cmdline from .version import __apk_version__, __atx_agent_version__ from .watcher import WatchContext, Watcher DEBUG = False WAIT_FOR_DEVICE_TIMEOUT = int(os.getenv("WAIT_FOR_DEVICE_TIMEOUT", 20)) - -log_format = '%(color)s[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d]%(end_color)s [pid:%(process)d] %(message)s' -formatter = logzero.LogFormatter(fmt=log_format) -logger = setup_logger("uiautomator2", level=logging.DEBUG, formatter=formatter) _mswindows = (os.name == "nt") +logger = logging.getLogger(__name__) + +def enable_pretty_logging(level=logging.DEBUG): + if not logger.handlers: + # Configure handler + handler = logging.StreamHandler() + formatter = logging.Formatter('[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d pid:%(process)d] %(message)s') + handler.setFormatter(formatter) + logger.addHandler(handler) + logger.setLevel(level) + class AppInstaller(abc.ABC): """ 解决手机安装apk弹窗问题 """ @abc.abstractmethod @@ -207,7 +210,7 @@ def request(self, method, url, **kwargs): raise OSError( "http-request to atx-agent error, can only recover from USB") - logger.warning("atx-agent has something wrong, auto recovering") + logger.info("atx-agent has something wrong, auto recovering") # ReadTimeout: sometime means atx-agent is running but not responsing # one reason is futex_wait_queue: https://stackoverflow.com/questions/9801256/app-hangs-on-futex-wait-queue-me-every-a-couple-of-minutes @@ -244,11 +247,6 @@ def __init__(self, serial_or_url: Optional[str] = None): # USB 连接 self._serial = serial_or_url self._atx_agent_url = None - - # setup logger - log_format = f'%(color)s[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d]%(end_color)s [pid:%(process)d] [{self._serial}] %(message)s' - formatter = logzero.LogFormatter(fmt=log_format) - self._logger = setup_logger(name="uiautomator2.client", level=logging.DEBUG, formatter=formatter) filelock_path = os.path.expanduser("~/.uiautomator2/filelocks/") + base64.urlsafe_b64encode(self._serial.encode('utf-8')).decode('utf-8') + ".lock" os.makedirs(os.path.dirname(filelock_path), exist_ok=True) @@ -256,10 +254,6 @@ def __init__(self, serial_or_url: Optional[str] = None): self._app_installer: AppInstaller = None - @property - def logger(self) -> logging.Logger: - return self._logger - def set_app_installer(self, app_installer: AppInstaller): """ set uiautomator.apk installer """ assert isinstance(app_installer, AppInstaller) @@ -277,7 +271,7 @@ def _get_atx_agent_url(self) -> str: except adbutils.AdbError as e: if not _is_tmq_production() and self._atx_agent_url: # when device offline, use atx-agent-url - self.logger.info( + logger.info( "USB disconnected, fallback to WiFi, ATX_AGENT_URL=%s", self._atx_agent_url) return self._atx_agent_url @@ -306,7 +300,7 @@ def _prepare_atx_agent(self): _d = self._wait_for_device() if not _d: raise RuntimeError("USB device %s is offline" % self._serial) - self.logger.debug("device %s is online", self._serial) + logger.debug("device %s is online", self._serial) version_url = self.path2url("/version") try: version = requests.get(version_url, timeout=3).text @@ -346,13 +340,13 @@ def _wait_for_device(self, timeout=None) -> adbutils.AdbDevice: deadline = time.time() + timeout while time.time() < deadline: title = "device reconnecting" if _is_remote else "wait-for-device" - self.logger.info("%s, time left(%.1fs)", title, deadline - time.time()) + logger.info("%s, time left(%.1fs)", title, deadline - time.time()) if _is_remote: try: adb.disconnect(self._serial) adb.connect(self._serial, timeout=1) except (adbutils.AdbError, adbutils.AdbTimeout) as e: - self.logger.debug("adb reconnect error: %s", str(e)) + logger.debug("adb reconnect error: %s", str(e)) time.sleep(1.0) continue try: @@ -380,7 +374,7 @@ def _setup_uiautomator(self): if not os.path.exists(apk_path) and os.path.exists(cwd_apk_path): apk_path = cwd_apk_path target_path = "/data/local/tmp/" + name - self.logger.debug("Install %s", name) + logger.debug("Install %s", name) self.push(apk_path, target_path) self.shell(['pm', 'install', '-r', '-t', target_path]) @@ -481,7 +475,7 @@ def _jsonrpc_retry_call(self, *args, **kwargs): except (NullObjectExceptionError, NullPointerExceptionError, StaleObjectExceptionError) as e: - self.logger.warning("jsonrpc call got: %s", str(e)) + logger.warning("jsonrpc call got: %s", str(e)) self.reset_uiautomator(str(e)) # added to fix strange fatal jsonrpc NullPointerException return self._jsonrpc_call(*args, **kwargs) @@ -618,7 +612,7 @@ def reset_uiautomator(self, reason="unknown", depth=0): ) if depth > 0: - self.logger.info("restart-uiautomator since \"%s\"", reason) + logger.info("restart-uiautomator since \"%s\"", reason) # Note: # atx-agent check has moved to _AgentRequestSession @@ -635,7 +629,7 @@ def reset_uiautomator(self, reason="unknown", depth=0): ok = self._force_reset_uiautomator_v2( launch_test_app=depth > 0) # uiautomator 2.0 if ok: - self.logger.info("uiautomator back to normal") + logger.info("uiautomator back to normal") return output = self._test_run_instrument() @@ -648,12 +642,11 @@ def reset_uiautomator(self, reason="unknown", depth=0): def _force_reset_uiautomator_v2(self, launch_test_app=False): brand = self.shell("getprop ro.product.brand").output.strip() - # self.logger.debug("Device: %s, %s", brand, self.serial) package_name = "com.github.uiautomator" self.uiautomator.stop() - self.logger.debug("kill process(ps): uiautomator") + logger.debug("kill process(ps): uiautomator") self._kill_process_by_name("uiautomator") ## Note: Here do not reinstall apks, since vivo and oppo reinstall need password @@ -673,7 +666,7 @@ def _force_reset_uiautomator_v2(self, launch_test_app=False): deadline = time.time() + 40.0 # in vivo-Y67, launch timeout 24s flow_window_showed = False while time.time() < deadline: - self.logger.debug("uiautomator-v2 is starting ... left: %.1fs", + logger.debug("uiautomator-v2 is starting ... left: %.1fs", deadline - time.time()) if not self.uiautomator.running(): @@ -687,7 +680,7 @@ def _force_reset_uiautomator_v2(self, launch_test_app=False): if not flow_window_showed: flow_window_showed = True self.show_float_window(True) - self.logger.debug("show float window") + logger.debug("show float window") time.sleep(1.0) continue return True @@ -735,7 +728,7 @@ def _package_version(self, package_name: str) -> Optional[packaging_version.Vers return None def _grant_app_permissions(self): - self.logger.debug("grant permissions") + logger.debug("grant permissions") for permission in [ "android.permission.SYSTEM_ALERT_WINDOW", "android.permission.ACCESS_FINE_LOCATION", @@ -752,7 +745,7 @@ def _test_run_instrument(self): def _kill_process_by_name(self, name, use_adb=False): for p in self._iter_process(use_adb=use_adb): if p.name == name and p.user == "shell": - self.logger.debug("kill %s", name) + logger.debug("kill %s", name) kill_cmd = ["kill", "-9", str(p.pid)] if use_adb: self._adb_device.shell(kill_cmd) @@ -876,7 +869,7 @@ def set_new_command_timeout(self, timeout: int): r = self.http.post("/newCommandTimeout", data=str(int(timeout))) data = r.json() assert data['success'], data['description'] - self.logger.info("%s", data['description']) + logger.info("%s", data['description']) @cached_property def device_info(self): @@ -1030,11 +1023,11 @@ def _operation_delay(self, operation_name: str = None): before, after = 0, 0 if before: - self.logger.debug(f"operation [{operation_name}] pre-delay {before}s") + logger.debug(f"operation [{operation_name}] pre-delay {before}s") time.sleep(before) yield if after: - self.logger.debug(f"operation [{operation_name}] post-delay {after}s") + logger.debug(f"operation [{operation_name}] post-delay {after}s") time.sleep(after) @property @@ -1953,7 +1946,7 @@ def connect_adb_wifi(addr) -> Device: return connect_usb(addr) -def connect_usb(serial: Optional[str] = None, init: bool = False) -> Device: +def connect_usb(serial: Optional[str] = None) -> Device: """ Args: serial (str): android device serial @@ -1964,9 +1957,6 @@ def connect_usb(serial: Optional[str] = None, init: bool = False) -> Device: Raises: ConnectError """ - if init: - logger.warning("connect_usb, args init=True is deprecated since 2.8.0") - if not serial: device = adbutils.adb.device() serial = device.serial diff --git a/uiautomator2/__main__.py b/uiautomator2/__main__.py index 31c171d1..76a267a0 100644 --- a/uiautomator2/__main__.py +++ b/uiautomator2/__main__.py @@ -4,17 +4,10 @@ from __future__ import absolute_import, print_function import argparse -import hashlib import json import logging -import os -import re import adbutils -import progress.bar -import requests -from logzero import logger -from retry import retry import uiautomator2 as u2 @@ -22,6 +15,8 @@ from .version import __version__ +logger = logging.getLogger(__name__) + def cmd_init(args): serial = args.serial or args.serial_optional if serial: @@ -284,6 +279,4 @@ def main(): if __name__ == '__main__': - # import logzero - # logzero.loglevel(logging.INFO) main() diff --git a/uiautomator2/image.py b/uiautomator2/image.py index 50944f37..e4545d20 100644 --- a/uiautomator2/image.py +++ b/uiautomator2/image.py @@ -17,7 +17,6 @@ import imutils import numpy as np import requests -from logzero import setup_logger from PIL import Image, ImageDraw from skimage.metrics import structural_similarity @@ -26,6 +25,7 @@ ImageType = typing.Union[np.ndarray, Image.Image] compare_ssim = structural_similarity +logger = logging.getLogger(__name__) def color_bgr2gray(image: ImageType): @@ -223,13 +223,10 @@ def __init__(self, d: "uiautomator2.Device"): Args: d (uiautomator2 instance) """ - self.logger = setup_logger() self._d = d assert hasattr(d, 'click') assert hasattr(d, 'screenshot') - self.logger.setLevel(logging.DEBUG) - def send_click(self, x, y): return self._d.click(x, y) @@ -274,13 +271,13 @@ def __wait(self, imdata, timeout=30.0, threshold=0.8): while time.time() < deadline: m = self.match(imdata) sim = m['similarity'] - self.logger.debug("similarity %.2f [~%.2f], left time: %.1fs", sim, + logger.debug("similarity %.2f [~%.2f], left time: %.1fs", sim, threshold, deadline - time.time()) if sim < threshold: continue time.sleep(.1) return m - self.logger.debug("image not found") + logger.debug("image not found") def wait(self, imdata, timeout=30.0, threshold=0.9): """ wait until image show up """ diff --git a/uiautomator2/init.py b/uiautomator2/init.py index 0284c5f8..d541a387 100644 --- a/uiautomator2/init.py +++ b/uiautomator2/init.py @@ -11,7 +11,6 @@ import adbutils import progress.bar import requests -from logzero import logger, setup_logger from retry import retry from uiautomator2.utils import natualsize @@ -22,6 +21,8 @@ GITHUB_BASEURL = "https://github.com/openatx" +logger = logging.getLogger(__name__) + class DownloadBar(progress.bar.PixelBar): message = "Downloading" suffix = '%(current_size)s/%(total_size)s' @@ -88,7 +89,7 @@ def cache_download(url, filename=None, timeout=None, storepath=None, logger=logg shutil.move(storepath + '.part', storepath) return storepath -def mirror_download(url: str, filename=None, logger: logging.Logger = logger): +def mirror_download(url: str, filename=None): """ Download from mirror, then fallback to origin url """ @@ -150,9 +151,7 @@ def __init__(self, device: adbutils.AdbDevice, loglevel=logging.DEBUG): or self.abi).split(",") self.__atx_listen_addr = "127.0.0.1:7912" - self.logger = setup_logger(level=loglevel) - # self.logger.debug("Initial device %s", device) - self.logger.info("uiautomator2 version: %s", __version__) + logger.info("uiautomator2 version: %s", __version__) def set_atx_agent_addr(self, addr: str): assert ":" in addr @@ -163,7 +162,7 @@ def atx_agent_path(self): return "/data/local/tmp/atx-agent" def shell(self, *args, timeout=60): - self.logger.debug("Shell: %s", args) + logger.debug("Shell: %s", args) return self._device.shell(args, timeout=60) @property @@ -222,9 +221,7 @@ def minitouch_url(self): @retry(tries=2, logger=logger) def push_url(self, url, dest=None, mode=0o755, tgz=False, extract_name=None): # yapf: disable - path = mirror_download(url, - filename=os.path.basename(url), - logger=self.logger) + path = mirror_download(url, filename=os.path.basename(url)) if tgz: tar = tarfile.open(path, 'r:gz') path = os.path.join(os.path.dirname(path), extract_name) @@ -234,7 +231,7 @@ def push_url(self, url, dest=None, mode=0o755, tgz=False, extract_name=None): # if not dest: dest = "/data/local/tmp/" + os.path.basename(path) - self.logger.debug("Push to %s:0%o", dest, mode) + logger.debug("Push to %s:0%o", dest, mode) self._device.sync.push(path, dest, mode=mode) return dest @@ -252,12 +249,12 @@ def is_apk_outdated(self): apk_debug = self._device.package_info("com.github.uiautomator") apk_debug_test = self._device.package_info( "com.github.uiautomator.test") - self.logger.debug("apk-debug package-info: %s", apk_debug) - self.logger.debug("apk-debug-test package-info: %s", apk_debug_test) + logger.debug("apk-debug package-info: %s", apk_debug) + logger.debug("apk-debug-test package-info: %s", apk_debug_test) if not apk_debug or not apk_debug_test: return True if apk_debug['version_name'] != __apk_version__: - self.logger.info( + logger.info( "package com.github.uiautomator version %s, latest %s", apk_debug['version_name'], __apk_version__) return True @@ -268,7 +265,7 @@ def is_apk_outdated(self): max_delta = datetime.timedelta(minutes=3) if abs(apk_debug['first_install_time'] - apk_debug_test['first_install_time']) > max_delta: - self.logger.debug( + logger.debug( "package com.github.uiautomator does not have a signature matching the target com.github.uiautomator" ) return True @@ -281,7 +278,7 @@ def is_atx_agent_outdated(self): """ agent_version = self._device.shell([self.atx_agent_path, "version"]).strip() if agent_version == "dev": - self.logger.info("skip version check for atx-agent dev") + logger.info("skip version check for atx-agent dev") return False # semver major.minor.patch @@ -291,7 +288,7 @@ def is_atx_agent_outdated(self): except ValueError: return True - self.logger.debug("Real version: %s, Expect version: %s", real_ver, + logger.debug("Real version: %s, Expect version: %s", real_ver, want_ver) if real_ver[:2] != want_ver[:2]: @@ -327,7 +324,7 @@ def _install_uiautomator_apks(self): for filename, url in app_uiautomator_apk_urls(): path = self.push_url(url, mode=0o644) self.shell("pm", "install", "-r", "-t", path) - self.logger.info("- %s installed", filename) + logger.info("- %s installed", filename) def _install_jars(self): """ use uiautomator 1.0 to run uiautomator test """ @@ -335,18 +332,18 @@ def _install_jars(self): self.push_url(url, "/data/local/tmp/" + name, mode=0o644) def _install_atx_agent(self): - self.logger.info("Install atx-agent %s", __atx_agent_version__) + logger.info("Install atx-agent %s", __atx_agent_version__) self.push_url(self.atx_agent_url, tgz=True, extract_name="atx-agent") def setup_atx_agent(self): # stop atx-agent first self.shell(self.atx_agent_path, "server", "--stop") if self.is_atx_agent_outdated(): - self.logger.info("Install atx-agent %s", __atx_agent_version__) + logger.info("Install atx-agent %s", __atx_agent_version__) self.push_url(self.atx_agent_url, tgz=True, extract_name="atx-agent") self.shell(self.atx_agent_path, 'server', '--nouia', '-d', "--addr", self.__atx_listen_addr) - self.logger.info("Check atx-agent version") + logger.info("Check atx-agent version") self.check_atx_agent_version() @retry( @@ -355,39 +352,39 @@ def setup_atx_agent(self): tries=10) def check_atx_agent_version(self): port = self._device.forward_port(7912) - self.logger.debug("Forward: local:tcp:%d -> remote:tcp:%d", port, 7912) + logger.debug("Forward: local:tcp:%d -> remote:tcp:%d", port, 7912) version = requests.get("http://%s:%d/version" % (self._device._client.host, port)).text.strip() - self.logger.debug("atx-agent version %s", version) + logger.debug("atx-agent version %s", version) wlan_ip = requests.get("http://%s:%d/wlan/ip" % (self._device._client.host, port)).text.strip() - self.logger.debug("device wlan ip: %s", wlan_ip) + logger.debug("device wlan ip: %s", wlan_ip) return version def install(self): """ TODO: push minicap and minitouch from tgz file """ - self.logger.info("Install minicap, minitouch") + logger.info("Install minicap, minitouch") self.push_url(self.minitouch_url) if self.abi == "x86": - self.logger.info( + logger.info( "abi:x86 not supported well, skip install minicap") elif int(self.sdk) > 30: - self.logger.info("Android R (sdk:30) has no minicap resource") + logger.info("Android R (sdk:30) has no minicap resource") else: for url in self.minicap_urls: self.push_url(url) # self._install_jars() # disable jars if self.is_apk_outdated(): - self.logger.info( + logger.info( "Install com.github.uiautomator, com.github.uiautomator.test %s", __apk_version__) self._install_uiautomator_apks() else: - self.logger.info("Already installed com.github.uiautomator apks") + logger.info("Already installed com.github.uiautomator apks") self.setup_atx_agent() print("Successfully init %s" % self._device) @@ -395,14 +392,14 @@ def install(self): def uninstall(self): self._device.shell([self.atx_agent_path, "server", "--stop"]) self._device.shell(["rm", self.atx_agent_path]) - self.logger.info("atx-agent stopped and removed") + logger.info("atx-agent stopped and removed") self._device.shell(["rm", "/data/local/tmp/minicap"]) self._device.shell(["rm", "/data/local/tmp/minicap.so"]) self._device.shell(["rm", "/data/local/tmp/minitouch"]) - self.logger.info("minicap, minitouch removed") + logger.info("minicap, minitouch removed") self._device.shell(["pm", "uninstall", "com.github.uiautomator"]) self._device.shell(["pm", "uninstall", "com.github.uiautomator.test"]) - self.logger.info("com.github.uiautomator uninstalled, all done !!!") + logger.info("com.github.uiautomator uninstalled, all done !!!") if __name__ == "__main__": diff --git a/uiautomator2/watcher.py b/uiautomator2/watcher.py index 770db565..c06faa2e 100644 --- a/uiautomator2/watcher.py +++ b/uiautomator2/watcher.py @@ -9,18 +9,14 @@ from collections import OrderedDict from typing import Optional -from logzero import setup_logger - import uiautomator2 from uiautomator2.xpath import XPath +from uiautomator2.utils import inject_call -from .utils import inject_call - -logger = logging.getLogger("uiautomator2") +logger = logging.getLogger(__name__) def _callback_click(el): - # print("callback", threading.current_thread()) el.click() @@ -155,18 +151,6 @@ def __init__(self, d: "uiautomator2.Device"): self._watching = False # func start is calling self._triggering = False - self.logger = setup_logger() - self.logger.setLevel(logging.INFO) - - @property - def debug(self): - return self.logger.level == logging.DEBUG - - @debug.setter - def debug(self, v: bool): - assert isinstance(v, bool) - self.logger.setLevel(logging.DEBUG if v else logging.INFO) - @property def _xpath(self) -> XPath: return self._d.xpath @@ -180,7 +164,7 @@ def when(self, xpath=None): def start(self, interval: float = 2.0): """ stop watcher """ if self._watching: - self.logger.warning("already started") + logger.warning("already started") return self._watching = True th = threading.Thread(name="watcher", @@ -193,7 +177,7 @@ def start(self, interval: float = 2.0): def stop(self): """ stop watcher """ if not self._watching: - self.logger.warning("watch already stopped") + logger.warning("watch already stopped") return if self._watch_stopped.is_set(): @@ -239,7 +223,7 @@ def run(self, source: Optional[str] = None): try: return self._run_watchers(source=source) except Exception as e: - self.logger.warning("_run_watchers exception: %s", e) + logger.warning("_run_watchers exception: %s", e) return False def _run_watchers(self, source=None) -> bool: @@ -258,7 +242,7 @@ def _run_watchers(self, source=None) -> bool: break if last_selector: - self.logger.info("XPath(hook:%s): %s", h['name'], h['xpaths']) + logger.info("XPath(hook:%s): %s", h['name'], h['xpaths']) self._triggering = True cb = h['callback'] defaults = { @@ -276,7 +260,7 @@ def _run_watchers(self, source=None) -> bool: try: cb(*ba.args, **ba.kwargs) except Exception as e: - self.logger.warning("watchers exception: %s", e) + logger.warning("watchers exception: %s", e) finally: self._triggering = False return True @@ -292,7 +276,7 @@ def remove(self, name=None): return for w in self._watchers[:]: if w['name'] == name: - self.logger.debug("remove(%s) %s", name, w['xpaths']) + logger.debug("remove(%s) %s", name, w['xpaths']) self._watchers.remove(w) diff --git a/uiautomator2/webview.py b/uiautomator2/webview.py index d277b217..03e09124 100644 --- a/uiautomator2/webview.py +++ b/uiautomator2/webview.py @@ -2,17 +2,16 @@ # # Not implemented yet. # - -import contextlib import json +import logging import string from pprint import pprint import adbutils import pychrome import requests -from logzero import logger +logger = logging.getLogger(__name__) class WebviewDriver(): def __init__(self, url): diff --git a/uiautomator2/widget.py b/uiautomator2/widget.py index 5809e6de..68992ede 100644 --- a/uiautomator2/widget.py +++ b/uiautomator2/widget.py @@ -9,13 +9,13 @@ from typing import Union import requests -from logzero import logger, setup_logger from lxml import etree import uiautomator2 as u2 -import uiautomator2.image as uim from uiautomator2.image import compare_ssim, draw_point, imread +logger = logging.getLogger(__name__) + def xml2nodes(xml_content: Union[str, bytes]): if isinstance(xml_content, str): @@ -92,9 +92,6 @@ def __init__(self, d: "u2.Device"): self.popups = [] - self.logger = setup_logger() - self.logger.setLevel(logging.INFO) - @property def wait_timeout(self): return self._d.settings['wait_timeout'] diff --git a/uiautomator2/xpath.py b/uiautomator2/xpath.py index 4d2e07cc..6f75c549 100644 --- a/uiautomator2/xpath.py +++ b/uiautomator2/xpath.py @@ -3,7 +3,6 @@ from __future__ import absolute_import -import abc import functools import io import json @@ -12,12 +11,10 @@ import threading import time from collections import defaultdict -from types import ModuleType -from typing import Callable, Optional, Union +from typing import Callable, List, Optional, Union import adbutils from deprecated import deprecated -from logzero import logger, setup_logger from PIL import Image from lxml import etree @@ -27,6 +24,8 @@ from uiautomator2.utils import inject_call, swipe_in_bounds +logger = logging.getLogger(__name__) + def safe_xmlstr(s): return s.replace("$", "-") @@ -50,7 +49,7 @@ def is_xpath_syntax_ok(xpath_expression) -> bool: return False # Indicates a syntax error in the XPath expression -def strict_xpath(xpath: str, logger=logger) -> str: +def strict_xpath(xpath: str) -> str: """make xpath to be computer recognized xpath""" orig_xpath = xpath @@ -119,11 +118,6 @@ def __init__(self, d: DeviceInterface): self._alias_strict = False self._dump_lock = threading.Lock() - # 这里setup_logger不可以通过level参数传入logging.INFO - # 不然其StreamHandler都要重新setLevel，没看懂也没关系，反正就是不要这么干. 特此备注 - self._logger = setup_logger() - self._logger.setLevel(logging.INFO) - def global_set(self, key, value): valid_keys = { "timeout", @@ -143,15 +137,6 @@ def implicitly_wait(self, timeout): """set default timeout when click""" self._d.wait_timeout = timeout - @property - def logger(self): - expect_level = ( - logging.DEBUG if self._d.settings["xpath_debug"] else logging.INFO - ) # yapf: disable - if expect_level != self._logger.level: - self._logger.setLevel(expect_level) - return self._logger - @property def wait_timeout(self): return self._d.wait_timeout @@ -176,7 +161,7 @@ def add_event_listener(self, event_name, callback): def send_click(self, x, y): if self._click_before_delay: - self.logger.debug( + logger.debug( "click before delay %.1f seconds", self._click_after_delay ) time.sleep(self._click_before_delay) @@ -189,7 +174,7 @@ def send_click(self, x, y): self._d.click(x, y) if self._click_after_delay: - self.logger.debug("click after delay %.1f seconds", self._click_after_delay) + logger.debug("click after delay %.1f seconds", self._click_after_delay) time.sleep(self._click_after_delay) def send_longclick(self, x, y): @@ -214,44 +199,6 @@ def match(self, xpath, source=None): def when(self, xquery: str): return self._watcher.when(xquery) - @deprecated(version="3.0.0", reason="deprecated") - def apply_watch_from_yaml(self, data): - """ - Examples of argument data - - --- - - when: "@com.example.app/popup" - then: > - def callback(d): - d.click(0.5, 0.5) - - when: 继续 - then: click - """ - try: - import yaml - except ImportError: - self.logger.warning("missing lib pyyaml") - - data = yaml.load(data, Loader=yaml.SafeLoader) - for item in data: - when, then = item["when"], item["then"] - - trigger = lambda: None - self.logger.info("%s, %s", when, then) - if then == "click": - trigger = lambda selector: selector.get_last_match().click() - trigger.__doc__ = "click" - elif then.lstrip().startswith("def callback"): - mod = ModuleType("_inner_module") - exec(then, mod.__dict__) - trigger = mod.callback - trigger.__doc__ = then - else: - self.logger.warning("Unknown then: %r", then) - - self.logger.debug("When: %r, Trigger: %r", when, trigger.__doc__) - self.when(when).call(trigger) - @deprecated(version="3.0.0", reason="use d.watcher.run() instead") def run_watchers(self, source=None): self._watcher.run() @@ -282,7 +229,7 @@ def _get_after_watch(self, xpath: Union[str, list], timeout=None): if timeout == 0: timeout = 0.01 timeout = timeout or self.wait_timeout - self.logger.info("XPath(timeout %.1f) %s", timeout, xpath) + logger.info("XPath(timeout %.1f) %s", timeout, xpath) deadline = time.time() + timeout while True: source = self.dump_hierarchy() @@ -356,21 +303,18 @@ def __alias_get(self, key, default=None): return value def __call__(self, xpath: str, source=None): - print("XPATH:", xpath) return XPathSelector(self, xpath, source) class XPathSelector(object): def __init__(self, parent: XPath, xpath: str = None, source: str = None): - self.logger = parent.logger - self._parent = parent self._d = parent._d self._source = source self._last_source = None self._position = None self._fallback = None - self._xpath_list = (strict_xpath(xpath, self.logger),) if xpath else () + self._xpath_list = (strict_xpath(xpath),) if xpath else () def __str__(self): return f"XPathSelector={'|'.join(self._xpath_list)}" @@ -395,7 +339,7 @@ def xpath(self, _xpath: Union[list, tuple, str]): for xp in _xpath: new = new.xpath(xp) else: - new._xpath_list = new._xpath_list + (strict_xpath(_xpath, self.logger),) + new._xpath_list = new._xpath_list + (strict_xpath(_xpath),) return new def child(self, _xpath: str) -> "XPathSelector": @@ -434,14 +378,13 @@ def fallback(self, func: Optional[Callable[..., bool]] = None, *args, **kwargs): return new @property - def _global_timeout(self): - return self._parent.wait_timeout + def _global_timeout(self) -> float: + if hasattr(self._parent, "wait_timeout") and isinstance(self._parent.wait_timeout, (int, float)): + return self._parent.wait_timeout + return 20.0 - def all(self, source=None): - """ - Returns: - list of XMLElement - """ + def all(self, source=None) -> List["XMLElement"]: + """find all matched elements""" xml_content = source or self._source or self._parent.dump_hierarchy() self._last_source = xml_content @@ -456,7 +399,7 @@ def all(self, source=None): trigger_count += 1 xml_content = self._parent.dump_hierarchy() if trigger_count: - self.logger.debug("watcher triggered %d times", trigger_count) + logger.debug("watcher triggered %d times", trigger_count) if hierarchy is None: root = etree.fromstring(str2bytes(xml_content)) @@ -541,20 +484,14 @@ def set_text(self, text: str = ""): self._parent.send_text(text) def wait(self, timeout=None) -> bool: - """ - Args: - timeout (float): seconds - - Returns: - None or XMLElement - """ + """ wait until element found """ deadline = time.time() + (timeout or self._global_timeout) - while time.time() < deadline: - # self.logger.debug("wait %s left %.1fs", self, deadline-time.time()) + while True: if self.exists: return True + if time.time() > deadline: + return False time.sleep(0.2) - return False def match(self) -> Optional["XMLElement"]: """ @@ -581,7 +518,7 @@ def wait_gone(self, timeout=None) -> bool: def click_nowait(self): x, y = self.all()[0].center() - self.logger.info("click %d, %d", x, y) + logger.info("click %d, %d", x, y) self._parent.send_click(x, y) def click(self, timeout=None): @@ -592,7 +529,7 @@ def click(self, timeout=None): except XPathElementNotFoundError: if not self._fallback: raise - self.logger.info("element not found, run fallback") + logger.info("element not found, run fallback") return inject_call(self._fallback, d=self._d) def click_exists(self, timeout=None) -> bool: @@ -661,24 +598,6 @@ def get_xpath(self, strip_index: bool = False): path = re.sub(r"\[\d+\]", "", path) # remove indexes return path - # 模糊对比方法，后来发现直接对比XPath似乎更好一些 - # def fuzzy_equal(self, xml_element) -> bool: - # root = self.elem.getroottree() - # fullpath = root.getpath(self.elem) - # fullpath = re.sub(r'\[\d+\]', '', fullpath) # remove indexes - - # compared_attrs = ("text", "resource-id", "package", "content-desc") - # for name in compared_attrs: - # if self.elem.attrib[name] != xml_element.attrib[name]: - # return False - - # def _elem2fullpath(el): - # root = el.getroottree() - # fullpath = root.getpath(el) - # return re.sub(r'\[\d+\]', '', fullpath) # remove indexes - - # return _elem2fullpath(self.elem) == _elem2fullpath(xml_element.elem) - def center(self): """ Returns: