| # -*- coding: utf-8 -*- |
| ''' |
| Copyright (C) 2012-2015 Diego Torres Milano |
| Created on Feb 2, 2015 |
| |
| Licensed under the Apache License, Version 2.0 (the "License"); |
| you may not use this file except in compliance with the License. |
| You may obtain a copy of the License at |
| |
| http://www.apache.org/licenses/LICENSE-2.0 |
| |
| Unless required by applicable law or agreed to in writing, software |
| distributed under the License is distributed on an "AS IS" BASIS, |
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| See the License for the specific language governing permissions and |
| limitations under the License. |
| |
| @author: Diego Torres Milano |
| ''' |
| |
| __version__ = '11.5.9' |
| |
| import os |
| import subprocess |
| import sys |
| import platform |
| import threading |
| import re |
| import json |
| try: |
| import requests |
| REQUESTS_AVAILABLE = True |
| except: |
| REQUESTS_AVAILABLE = False |
| import time |
| from com.dtmilano.android.adb.adbclient import AdbClient |
| from com.dtmilano.android.common import obtainAdbPath |
| |
| __author__ = 'diego' |
| |
| |
| DEBUG = False |
| |
| lock = threading.Lock() |
| |
| class RunTestsThread(threading.Thread): |
| """ |
| Runs the instrumentation for the specified package in a new thread. |
| """ |
| def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, verbose=None, adbClient=None, testClass=None, testRunner=None): |
| threading.Thread.__init__(self, group=group, target=target, name=name, verbose=verbose) |
| self.adbClient = adbClient |
| self.testClass = testClass |
| self.testRunner = testRunner |
| self.pkg = re.sub('\.test$', '', self.testClass) |
| |
| def run(self): |
| if DEBUG: |
| print >> sys.stderr, "RunTestsThread: Acquiring lock" |
| lock.acquire() |
| if DEBUG: |
| print >> sys.stderr, "RunTestsThread: Lock acquired" |
| self.forceStop() |
| time.sleep(3) |
| if DEBUG: |
| print >> sys.stderr, "Starting test..." |
| print >> sys.stderr, "RunTestsThread: Releasing lock" |
| lock.release() |
| out = self.adbClient.shell('am instrument -w ' + self.testClass + '/' + self.testRunner + '; echo "ERROR: $?"') |
| if DEBUG: |
| print >> sys.stderr, "\nFinished test." |
| errmsg = out.splitlines()[-1] |
| m = re.match('ERROR: (\d+)', errmsg) |
| if m: |
| exitval = int(m.group(1)) |
| if exitval != 0: |
| raise RuntimeError('Cannot start test on device: ' + out) |
| else: |
| raise RuntimeError('Unknown message') |
| |
| def forceStop(self): |
| if DEBUG: |
| print >> sys.stderr, "Cleaning up before start. Stopping '%s'" % self.pkg |
| self.adbClient.shell('am force-stop ' + self.pkg) |
| |
| |
| class UiAutomatorHelper: |
| PACKAGE = 'com.dtmilano.android.culebratester' |
| TEST_CLASS = PACKAGE + '.test' |
| TEST_RUNNER = 'com.dtmilano.android.uiautomatorhelper.UiAutomatorHelperTestRunner' |
| |
| def __init__(self, adbclient, adb=None, localport=9999, remoteport=9999, hostname='localhost'): |
| if not REQUESTS_AVAILABLE: |
| raise Exception('''Python Requests is needed for UiAutomatorHelper to work. |
| |
| On Ubuntu install |
| |
| $ sudo apt-get install python-requests |
| |
| On OSX install |
| |
| $ easy_install requests |
| ''') |
| |
| self.adbClient = adbclient |
| ''' The adb client (a.k.a. device) ''' |
| instrumentation = self.adbClient.shell('pm list instrumentation %s' % self.PACKAGE) |
| if not re.match('instrumentation:%s/%s \(target=%s\)' % (self.TEST_CLASS, self.TEST_RUNNER, self.PACKAGE), instrumentation): |
| raise RuntimeError('The target device does not contain the instrumentation for %s' % self.PACKAGE) |
| self.adb = self.__whichAdb(adb) |
| ''' The adb command ''' |
| self.osName = platform.system() |
| ''' The OS name. We sometimes need specific behavior. ''' |
| self.isDarwin = (self.osName == 'Darwin') |
| ''' Is it Mac OSX? ''' |
| self.hostname = hostname |
| ''' The hostname we are connecting to. ''' |
| if hostname in ['localhost', '127.0.0.1']: |
| self.__redirectPort(localport, remoteport) |
| self.__runTests() |
| self.baseUrl = 'http://%s:%d' % (hostname, localport) |
| try: |
| self.session = self.__connectSession() |
| except RuntimeError, ex: |
| self.thread.forceStop() |
| raise ex |
| |
| |
| def __connectSession(self): |
| if DEBUG: |
| print >> sys.stderr, "UiAutomatorHelper: Acquiring lock" |
| lock.acquire() |
| if DEBUG: |
| print >> sys.stderr, "UiAutomatorHelper: Lock acquired" |
| print >> sys.stderr, "UiAutomatorHelper: Connecting session" |
| session = requests.Session() |
| if not session: |
| raise RuntimeError("Cannot create session") |
| tries = 10 |
| while tries > 0: |
| time.sleep(0.5) |
| if DEBUG: |
| print >> sys.stderr, "UiAutomatorHelper: Attempting to connect to", self.baseUrl, '(tries=%s)' % tries |
| try: |
| response = session.head(self.baseUrl) |
| if response.status_code == 200: |
| break |
| except requests.exceptions.ConnectionError, ex: |
| tries -= 1 |
| lock.release() |
| if tries == 0: |
| raise RuntimeError("Cannot connect to " + self.baseUrl) |
| if DEBUG: |
| print >> sys.stderr, "UiAutomatorHelper: HEAD", response |
| print >> sys.stderr, "UiAutomatorHelper: Releasing lock" |
| #lock.release() |
| return session |
| |
| def __whichAdb(self, adb): |
| if adb: |
| if not os.access(adb, os.X_OK): |
| raise Exception('adb="%s" is not executable' % adb) |
| else: |
| # Using adbclient we don't need adb executable yet (maybe it's needed if we want to |
| # start adb if not running) or to redirect ports |
| adb = obtainAdbPath() |
| |
| return adb |
| |
| def __redirectPort(self, localport, remoteport): |
| self.localPort = localport |
| self.remotePort = remoteport |
| subprocess.check_call([self.adb, '-s', self.adbClient.serialno, 'forward', 'tcp:%d' % self.localPort, |
| 'tcp:%d' % self.remotePort]) |
| |
| def __runTests(self): |
| if DEBUG: |
| print >> sys.stderr, "__runTests: start" |
| # We need a new AdbClient instance with timeout=None (means, no timeout) for the long running test service |
| newAdbClient = AdbClient(self.adbClient.serialno, self.adbClient.hostname, self.adbClient.port, timeout=None) |
| self.thread = RunTestsThread(adbClient=newAdbClient, testClass=self.TEST_CLASS, testRunner=self.TEST_RUNNER) |
| if DEBUG: |
| print >> sys.stderr, "__runTests: starting thread" |
| self.thread.start() |
| if DEBUG: |
| print >> sys.stderr, "__runTests: end" |
| |
| |
| def __httpCommand(self, url, params=None, method='GET'): |
| if method == 'GET': |
| if params: |
| response = self.session.get(self.baseUrl + url, params=params) |
| else: |
| response = self.session.get(self.baseUrl + url) |
| elif method == 'PUT': |
| response = self.session.put(self.baseUrl + url, params=params) |
| else: |
| raise RuntimeError("method not supported: " + method) |
| return response.content |
| |
| # |
| # Device |
| # |
| def getDisplayRealSize(self): |
| return self.__httpCommand('/Device/getDisplayRealSize') |
| |
| # |
| # UiAutomatorHelper internal commands |
| # |
| def quit(self): |
| try: |
| self.__httpCommand('/UiAutomatorHelper/quit') |
| except: |
| pass |
| self.session.close() |
| |
| # |
| # UiDevice |
| # |
| def click(self, **kwargs): |
| params = kwargs |
| if not ((params.has_key('x') and params.has_key('y')) or params.has_key('oid')): |
| raise RuntimeError('click: (x, y) or oid must have a value') |
| if params.has_key('oid'): |
| return self.__httpCommand('/UiObject2/%d/click' % params['oid']) |
| else: |
| return self.__httpCommand('/UiDevice/click', params) |
| |
| def dumpWindowHierarchy(self): |
| dump = self.__httpCommand('/UiDevice/dumpWindowHierarchy').decode(encoding='UTF-8', errors='replace') |
| if DEBUG: |
| print >> sys.stderr, "DUMP: ", dump |
| return dump |
| |
| def findObject(self, **kwargs): |
| params = kwargs |
| if not (params.has_key('resourceId') or params.has_key('bySelector')): |
| raise RuntimeError('findObject: resourceId or bySelector must have a value') |
| response = self.__httpCommand('/UiDevice/findObject', params) |
| # { "status": "OK", "oid": 1, "className": "android.view.View"} |
| if DEBUG: |
| print >> sys.stderr, "UiAutomatorHelper: findObject: response=", response |
| r = json.loads(response) |
| if r[u'status'] == 'OK': |
| if DEBUG: |
| print >> sys.stderr, "UiAutomatorHelper: findObject: returning", int(r[u'oid']) |
| return UiObject(self, int(r[u'oid'])) |
| raise RuntimeError("Error: " + response) |
| |
| def longClick(self, **kwargs): |
| params = kwargs |
| if not ((params.has_key('x') and params.has_key('y')) or params.has_key('oid')): |
| raise RuntimeError('longClick: (x, y) or oid must have a value') |
| if params.has_key('oid'): |
| return self.__httpCommand('/UiObject2/%d/longClick' % params['oid']) |
| else: |
| return self.__httpCommand('/UiDevice/longClick', params) |
| |
| def pressBack(self): |
| return self.__httpCommand('/UiDevice/pressBack') |
| |
| def pressHome(self): |
| return self.__httpCommand('/UiDevice/pressHome') |
| |
| def pressKeyCode(self, keyCode, metaState=0): |
| params = {'keyCode': keyCode, 'metaState': metaState} |
| return self.__httpCommand('/UiDevice/pressKeyCode', params) |
| |
| def pressRecentApps(self): |
| return self.__httpCommand('/UiDevice/pressRecentApps') |
| |
| def swipe(self, startX=-1, startY=-1, endX=-1, endY=-1, steps=10, segments=[], segmentSteps=5): |
| if startX != -1 and startY != -1: |
| params = {'startX': startX, 'startY': startY, 'endX': endX, 'endY': endY, 'steps': steps} |
| elif segments: |
| params = {'segments': ','.join(str(p) for p in segments), "segmentSteps": segmentSteps} |
| else: |
| raise RuntimeError("Cannot determine method invocation from provided parameters. startX and startY or segments must be provided.") |
| return self.__httpCommand('/UiDevice/swipe', params) |
| |
| def takeScreenshot(self, scale=1.0, quality=90): |
| params = {'scale': scale, 'quality': quality} |
| return self.__httpCommand('/UiDevice/takeScreenshot', params) |
| |
| # |
| # UiObject |
| # |
| def setText(self, uiObject, text): |
| params = {'text': text} |
| return self.__httpCommand('/UiObject/0x%x/setText' % (uiObject.oid), params) |
| |
| |
| class UiObject: |
| def __init__(self, uiAutomatorHelper, oid): |
| self.uiAutomatorHelper = uiAutomatorHelper |
| self.oid = oid |
| |
| def click(self): |
| self.uiAutomatorHelper.click(oid=self.oid) |
| |
| def longClick(self): |
| self.uiAutomatorHelper.longClick(oid=self.oid) |
| |
| def setText(self, text): |
| self.uiAutomatorHelper.setText(uiObject=self, text=text) |
