#!/usr/bin/env python # coding: utf-8 # """ packageName: com.netease.cloudmusic # Optional activity: .MainActivity # Optional launch: true # Optional, default true close: true # Optional, close app when finished plugins: ocr: $ocr-api-url perf: interval: 1 debug: true filename: perf.csv steps: - q: ~hell - q: ^hello w - q: =hello world - q: hello world - q: foo timeout: 10 # seconds - text: the input text - code: | time.sleep(10) # only "d, time" can be used watchers: # click when show up - q: ~跳过 timeout: 0 # default """ import argparse import time import re import logging import uiautomator2 as u2 import logzero from logzero import logger class JSONRunner(object): __alias = { 'q': 'query', } def __init__(self, cnf): self._cnf = cnf self._title = cnf.get('title') self._pkg_name = cnf.get('packageName') self._activity = cnf.get('activity') self._launch = cnf.get('launch', True) self._close = cnf.get('close', True) self._clear = cnf.get('clear') self._steps = cnf.get('steps') self._watchers = cnf.get('watchers', []) self._udid = cnf.get('udid') self._plugins = cnf.get('plugins', {}) d = self._d = u2.connect(self._udid) self.session = None # plugins if 'ocr' in self._plugins: logger.info("load plugin: ocr") import ssat_sdk.uiautomator2.ext.ocr as ocr ocr.API = self._plugins['ocr'] logger.info("Use ocr plugin: %s", ocr.API) u2.plugin_register('ocr', ocr.OCR) if self._pkg_name and 'perf' in self._plugins: logger.info("load plugin: perf") import ssat_sdk.uiautomator2.ext.perf as perf u2.plugin_register('perf', perf.Perf, self._pkg_name) perfcnf = self._plugins['perf'] d.ext_perf.interval = perfcnf.get('interval', 1.0) d.ext_perf.debug = perfcnf.get('debug', True) d.ext_perf.csv_output = perfcnf.get('filename', 'perf.csv') def _find_xpath(self, xpath, hierarchy): el = self.session.xpath(xpath, hierarchy) if el.exists: return el.wait().center() def _find_text(self, text, hierarchy): elements = self.session.xpath( '//*[re:match(name(), "\.(TextView|Button|ImageView)$")]', hierarchy).all() if text.startswith('~') or text.startswith('^'): # Regexp pattern = re.compile(text[1:] if text.startswith('~') else text) for el in elements: if el.text and pattern.match(el.text): logger.debug("find match: %s, %s", el.text, el.attrib) return el.center() elif text.startswith("="): text = text[1:] for el in elements: if el.text == text: logger.debug("find exactly match text: %s", el.text) return el.center() def _oper_input(self, text): logger.info("input text: %s", text) self.session.set_fastinput_ime(True) self.session.send_keys(text) self.session.press("enter") def _run_watchers(self, hierarchy): for kwargs in self._watchers: kwargs['timeout'] = kwargs.get('timeout', 0) if 'q' in kwargs: kwargs['query'] = kwargs.pop('q') if self._run_onestep(hierarchy, **kwargs): logger.info("trigger watcher: %s", kwargs) def _run_onestep(self, hierarchy, action='click', query=None, text=None, code=None, ocr=None, timeout=10): """ Returns: bool: if step handled """ if text: self._oper_input(text) return True if ocr: self._d.ext_ocr(ocr).click(timeout=timeout) return True # find element and click if query: pos = None if query.startswith("/"): pos = self._find_xpath(query, hierarchy) else: pos = self._find_text(query, hierarchy) if pos is None: if action == 'assertNotExists': return True return False if action == 'click': logger.info("click: %s", pos) self.session.click(*pos) return True elif action == 'assertExists': return True if code: logger.info("exec: |\n%s", code) exec(code, {'d': self.session, 'time': time}) return True # raise NotImplementedError("only support click action") def _handle_step(self, **kwargs): retry_cnt = 0 timeout = kwargs.pop('timeout', 10) deadline = time.time() + timeout # alias if 'q' in kwargs: kwargs['query'] = kwargs.pop('q') while retry_cnt == 0 or time.time() < deadline: retry_cnt += 1 hierarchy = self.session.dump_hierarchy() self._run_watchers(hierarchy) # 处理弹框 if self._run_onestep(hierarchy, **kwargs): break logger.debug("process %s, retry %d", kwargs, retry_cnt) time.sleep(.5) else: raise RuntimeError("element not found: %s", kwargs) def prepare_session(self): d = self._d if not self._launch or not self._pkg_name: self.session = self._d.session(self._pkg_name, attach=True) return if self._clear: d.app_clear(self._pkg_name) s = None if self._activity: d.app_start(self._pkg_name, self._activity, stop=True) s = d.session(self._pkg_name, attach=True) else: s = d.session(self._pkg_name) self.session = s def run(self): logger.info("test begins: %s", self._title) logger.info("launch app: %s", self._pkg_name) self.prepare_session() if 'perf' in self._plugins: # add report import easyhtmlreport as htmlreport hrp = htmlreport.HTMLReport(self._d) hrp.patch_click() self._d.ext_perf.csv_output = 'report/perf.csv' self._d.ext_perf.start() try: for step in self._steps: logger.info("==> %s", step) self._handle_step(**step) logger.info("Finished") finally: if 'perf' in self._plugins: self._d.ext_perf.stop() if self._launch and self._pkg_name and self._close: self._d.app_stop(self._pkg_name) def main(filename, debug=False, onlystep=False): if not debug: logzero.loglevel(logging.INFO) import yaml with open(filename, 'rb') as f: cnf = yaml.load(f) if onlystep: cnf['launch'] = False tc = JSONRunner(cnf) tc.run() # if __name__ == '__main__': # parser = argparse.ArgumentParser() # parser.add_argument( # '-d', '--debug', help='set loglevel to debug', action='store_true') # parser.add_argument('--step', help='run only step', action='store_true') # parser.add_argument('yamlfile') # args = parser.parse_args() # main(args.yamlfile, args.debug, onlystep=args.step)