Package tlib :: Package base :: Module AndroidHelper
[hide private]
[frames] | no frames]

Source Code for Module tlib.base.AndroidHelper

  1  import requests 
  2  import time 
  3  import pytest 
  4  import re 
  5  from _pytest import runner 
  6  from tlib.base import TestHelper 
  7  import jsonpath_rw 
  8   
  9  DEVICE_UNPLUGGED = "disconnected"     # Device is not connected 
 10  DEVICE_OFFLINE = "offline"            # Device is connected but offline 
 11  DEVICE_UNAUTHORIZED = "unauthorized"  # Unauthorized 
 12  DEVICE_ONLINE = "online"              # Device is connected and accessible 
 13   
 14   
15 -def is_webdriver_running(tlib_logger, adb_logger, log=True):
16 """ 17 Function to validate if webdriver is running and a connection can be established 18 19 @param tlib_logger: TLib logger 20 @type tlib_logger: logging.logger 21 @param adb_logger: ADB logger 22 @type adb_logger: logging.logger 23 @param log: When true, log debugging information 24 @type log: bool 25 """ 26 try: 27 if log: 28 tlib_logger.debug("Checking if Webdriver is running") 29 response = requests.get("http://localhost:8080/wd/hub/status", timeout=10) 30 except requests.ConnectionError as e: 31 if log: 32 adb_logger.debug("Connection to Webdriver failed:\n%s" % e) 33 return False 34 35 return response.status_code == 200
36 37
38 -def get_device_status(tlib_logger, adb_logger, serial_id):
39 """ 40 Get status of a device using it's serial id\n 41 Serial id can be either a ID (for devices connected to USB) or an IP and port (For devices connected via IP) 42 43 @param tlib_logger: TLib logger 44 @type tlib_logger: logging.logger 45 @param adb_logger: ADB logger 46 @type adb_logger: logging.logger 47 @param serial_id: Device's serial number 48 @type serial_id: str 49 """ 50 51 out = TestHelper.run_command(adb_logger, ["adb", "devices"], fail_on_error=False) 52 if re.search(serial_id, out[0]) is None: 53 tlib_logger.debug("Device {serial_id} is not connected".format(serial_id=serial_id)) 54 return DEVICE_UNPLUGGED 55 elif re.search("{serial_id}\s+offline".format(serial_id=serial_id), out[0]) is not None: 56 tlib_logger.debug("Device {serial_id} is offline".format(serial_id=serial_id)) 57 return DEVICE_OFFLINE 58 elif re.search("{serial_id}\s+unauthorized".format(serial_id=serial_id), out[0]) is not None: 59 tlib_logger.debug("Device {serial_id} is offline".format(serial_id=serial_id)) 60 return DEVICE_UNAUTHORIZED 61 elif re.search("{serial_id}\s+device".format(serial_id=serial_id), out[0]) is not None: 62 tlib_logger.debug("Device {serial_id} is online".format(serial_id=serial_id)) 63 return DEVICE_ONLINE 64 else: 65 tlib_logger.error("Unknown device status\n%s" % out[0]) 66 pytest.fail("Unknown device status\n%s" % out[0])
67 68
69 -def setup_ip_connection(tlib_logger, adb_logger, serial_id):
70 """ 71 Connects to an android device via IP and waits for the connection to be established. 72 73 @param tlib_logger: TLib logger 74 @type tlib_logger: logging.logger 75 @param adb_logger: ADB logger 76 @type adb_logger: logging.logger 77 @param serial_id: Device's serial number 78 @type serial_id: str 79 """ 80 # Validate is device is already connected 81 tlib_logger.debug("Setting up IP connection to device {serial_id}".format(serial_id=serial_id)) 82 status = get_device_status(tlib_logger, adb_logger, serial_id) 83 84 if status == DEVICE_ONLINE: 85 tlib_logger.debug("Device is already online".format(serial_id=serial_id)) 86 return 87 88 # If device is offline or unauthorized, disconnect 89 if status == DEVICE_OFFLINE: 90 tlib_logger.warn("Device {serial_id} is offline, disconnecting and retrying".format(serial_id=serial_id)) 91 terminate_ip_connection(tlib_logger, adb_logger, serial_id) 92 elif status == DEVICE_UNAUTHORIZED: 93 terminate_ip_connection(tlib_logger, adb_logger, serial_id) 94 tlib_logger.error("Device {serial_id} is not authorized, aborting".format(serial_id=serial_id)) 95 pytest.fail("Device {serial_id} is not authorized, aborting".format(serial_id=serial_id)) 96 elif status == DEVICE_UNPLUGGED: 97 tlib_logger.warn("Device {serial_id} is disconecting, resetting connection".format(serial_id=serial_id)) 98 terminate_ip_connection(tlib_logger, adb_logger, serial_id) 99 100 timeout = 0.5 101 out = None 102 #We wait up to 3 seconds 103 for i in range(1, int(3 / timeout)): 104 out = TestHelper.run_command(adb_logger, ["adb", "connect", serial_id]) 105 log_adb_output(adb_logger, out) 106 107 if get_device_status(tlib_logger, adb_logger, serial_id) == DEVICE_ONLINE: 108 tlib_logger.debug("Connected to device") 109 return 110 else: 111 tlib_logger.debug("Not yet connected") 112 time.sleep(timeout) 113 114 #If we're here, it means connection failed 115 adb_logger.error(r"Timeout while connecting to device {serial_id}\nADB output:\n{out}". 116 format(serial_id=serial_id, out=out)) 117 # noinspection PyUnresolvedReferences 118 pytest.fail(r"Timeout while connecting to device {serial_id}\nADB output:\n{out}". 119 format(serial_id=serial_id, out=out))
120 121
122 -def terminate_ip_connection(tlib_logger, adb_logger, serial_id):
123 """ 124 Disconnects from device 125 126 @param tlib_logger: TLib logger 127 @type tlib_logger: logging.logger 128 @param adb_logger: ADB logger 129 @type adb_logger: logging.logger 130 @param serial_id: Device's serial number 131 @type serial_id: str 132 """ 133 tlib_logger.debug("Disconnecting from device {serial_id}".format(serial_id=serial_id)) 134 out = TestHelper.run_command(adb_logger, ["adb", "disconnect", serial_id]) 135 log_adb_output(adb_logger, out)
136 137
138 -def close_webdriver(tlib_logger, adb_logger, serial_id):
139 """ 140 Stops Webdriver app o the device 141 142 @param tlib_logger: TLib logger 143 @type tlib_logger: logging.logger 144 @param adb_logger: ADB logger 145 @type adb_logger: logging.logger 146 @param serial_id: Device's serial number 147 @type serial_id: str 148 """ 149 tlib_logger.debug("Closing Webdriver on {serial_id}".format(serial_id=serial_id)) 150 if serial_id is None or serial_id == "": 151 out = TestHelper.run_command(adb_logger, 152 "adb shell am force-stop org.openqa.selenium.android.app", shell=True) 153 else: 154 out = TestHelper.run_command(adb_logger, 155 "adb -s %s shell am force-stop org.openqa.selenium.android.app" % serial_id, 156 shell=True) 157 log_adb_output(adb_logger, out)
158 159
160 -def setup_port_forwarding(tlib_logger, adb_logger, serial_id):
161 """ 162 Setup port forwarding between computer and device. 163 164 @param tlib_logger: TLib logger 165 @type tlib_logger: logging.logger 166 @param adb_logger: ADB logger 167 @type adb_logger: logging.logger 168 @param serial_id: Device's serial number 169 @type serial_id: str 170 """ 171 tlib_logger.info("Setting up port forwarding") 172 if serial_id is None or serial_id == "": 173 out = TestHelper.run_command(adb_logger, "adb forward tcp:8080 tcp:8080", shell=True) 174 else: 175 out = TestHelper.run_command(adb_logger, "adb -s %s forward tcp:8080 tcp:8080" % serial_id, shell=True) 176 log_adb_output(adb_logger, out)
177 178
179 -def teardown_port_forwarding(tlib_logger, adb_logger):
180 """ 181 Terminates all port forwarding connections 182 183 @param tlib_logger: TLib logger 184 @type tlib_logger: logging.logger 185 @param adb_logger: ADB logger 186 @type adb_logger: logging.logger 187 """ 188 tlib_logger.info("Tearing down port forwarding") 189 out = TestHelper.run_command(adb_logger, "adb forward --remove-all", shell=True) 190 log_adb_output(adb_logger, out)
191 192
193 -def start_adb_server(tlib_logger, adb_logger):
194 """ 195 Stops adb on the machine\n 196 This can be required by TeamCity so some folders are not locked 197 198 @param tlib_logger: TLib logger 199 @type tlib_logger: logging.logger 200 @param adb_logger: ADB logger 201 @type adb_logger: logging.logger 202 """ 203 tlib_logger.info("Starting ADB") 204 out = TestHelper.run_command(adb_logger, ["adb", "start-server"], fail_on_error=False) 205 log_adb_output(adb_logger, out)
206 207
208 -def stop_adb_server(tlib_logger, adb_logger):
209 """ 210 Stops adb on the machine\n 211 This can be required by TeamCity so some folders are not locked 212 213 @param tlib_logger: TLib logger 214 @type tlib_logger: logging.logger 215 @param adb_logger: ADB logger 216 @type adb_logger: logging.logger 217 """ 218 tlib_logger.info("Stopping ADB") 219 out = TestHelper.run_command(adb_logger, ["adb", "kill-server"], fail_on_error=False) 220 log_adb_output(adb_logger, out)
221 222
223 -def start_webdriver(tlib_logger, adb_logger, serial_id):
224 """ 225 Starts Webdriver app on the device 226 227 @param tlib_logger: TLib logger 228 @type tlib_logger: logging.logger 229 @param adb_logger: ADB logger 230 @type adb_logger: logging.logger 231 @param serial_id: Device's serial number 232 @type serial_id: str 233 """ 234 tlib_logger.info("Starting webdriver on the device") 235 if serial_id is None or serial_id == "": 236 out = TestHelper.run_command(adb_logger, "adb shell am start -a android.intent.action.MAIN -n " 237 "org.openqa.selenium.android.app/.MainActivity -e debug true", shell=True) 238 else: 239 out = TestHelper.run_command(adb_logger, "adb -s %s shell am start -a android.intent.action.MAIN " 240 "-n org.openqa.selenium.android.app/.MainActivity -e debug true" % 241 serial_id, shell=True) 242 log_adb_output(adb_logger, out)
243 244
245 -def wait_for_connection_to_webdriver(tlib_logger, adb_logger):
246 """ 247 Waits up to 3 seconds for a connection to Webdriver 248 249 @param tlib_logger: TLib logger 250 @type tlib_logger: logging.logger 251 @param adb_logger: ADB logger 252 @type adb_logger: logging.logger 253 """ 254 #Check connection to webdriver can be established. Retry 3 times 255 tlib_logger.debug("Waiting for connection to Webdriver") 256 timeout = 0.5 257 for i in range(1, int(3 / timeout)): 258 if is_webdriver_running(tlib_logger, adb_logger, False): 259 tlib_logger.debug("Webdriver started successfully") 260 break 261 tlib_logger.debug("Can't connect to Webdriver, retrying in {timeout} seconds".format(timeout=timeout)) 262 time.sleep(timeout) 263 264 if not is_webdriver_running(tlib_logger, adb_logger, False): 265 tlib_logger.error("Couldn't start Webdriver. Make sure it's installed and running\n" 266 "See https://code.google.com/p/selenium/wiki/AndroidDriver#Setup_the_Emulator " 267 "for more details") 268 # noinspection PyUnresolvedReferences 269 pytest.fail("Couldn't start Webdriver. Make sure it's installed and running\n" 270 "See https://code.google.com/p/selenium/wiki/AndroidDriver#Setup_the_Emulator for more details")
271 272 273 # noinspection PyUnresolvedReferences
274 -def setup_webdriver(tlib_logger, adb_logger, serial_id):
275 """ 276 Connects to a device and starts webdriver 277 278 @param tlib_logger: TLib logger 279 @type tlib_logger: logging.logger 280 @param adb_logger: ADB logger 281 @type adb_logger: logging.logger 282 @param serial_id: Device's serial number 283 @type serial_id: str 284 """ 285 #Try to connect to Webdriver and exit if success 286 if is_webdriver_running(tlib_logger, adb_logger, log=False): 287 tlib_logger.debug("Already connected to device") 288 return 289 290 tlib_logger.info("Connecting to Webdriver") 291 292 #Connect to device and setup port forwarding. 293 if TestHelper.is_valid_ip(serial_id): 294 setup_ip_connection(tlib_logger, adb_logger, serial_id) 295 296 setup_port_forwarding(tlib_logger, adb_logger, serial_id) 297 298 #Try again to connect to Webdriver and exit if success 299 if is_webdriver_running(tlib_logger, adb_logger): 300 tlib_logger.debug("Connected to Webdriver") 301 return 302 303 # Webdriver not running, start it 304 start_webdriver(tlib_logger, adb_logger, serial_id) 305 306 #Ensure we're connected 307 wait_for_connection_to_webdriver(tlib_logger, adb_logger) 308 309 tlib_logger.info("Connection to Webdriver established")
310 311 312 # noinspection PyUnresolvedReferences
313 -def teardown_webdriver(tlib_logger, adb_logger, serial_id):
314 """ 315 Closes webdriver and disconnects from device 316 317 @param tlib_logger: TLib logger 318 @type tlib_logger: logging.logger 319 @param adb_logger: ADB logger 320 @type adb_logger: logging.logger 321 @param serial_id: Device's serial number 322 @type serial_id: str 323 """ 324 #close selenium WebDriver 325 tlib_logger.info("Disconnecting from Webdriver") 326 close_webdriver(tlib_logger, adb_logger, serial_id) 327 328 # Terminates IP connection if an IP was given 329 if TestHelper.is_valid_ip(serial_id): 330 terminate_ip_connection(tlib_logger, adb_logger, serial_id) 331 332 #Stop adb service so it won't lock files required by TeamCity 333 try: 334 stop_adb_server(tlib_logger, adb_logger) 335 except runner.Failed: 336 #If ADB fails to stop, don't abort test 337 adb_logger.warn("Error stopping ADB server") 338 339 tlib_logger.info("Disconnected from Webdriver")
340 341 342 # noinspection PyUnresolvedReferences
343 -def start_selendroid_server(tlib_logger, adb_logger, app_path):
344 """ 345 Start selendroid server and point to apk file 346 347 @param tlib_logger: TLib logger 348 @type tlib_logger: logging.logger 349 @param adb_logger: Adb logger 350 @type adb_logger: logging.logger 351 @param app_path: Location of the apk file 352 @type app_path: str 353 """ 354 tlib_logger.info("Starting selendroid server") 355 stop_adb_server(tlib_logger, adb_logger) 356 start_adb_server(tlib_logger, adb_logger) 357 process = TestHelper.start_process(tlib_logger, "java -jar %s -aut %s" % (TestHelper.selendroid_server_jar(), 358 app_path), shell=False) 359 trial_left = 20 360 while not is_selendroid_running(tlib_logger, adb_logger, False) and trial_left > 0: 361 time.sleep(0.5) 362 trial_left -= 1 363 if trial_left > 0: 364 return process 365 return None
366 367
368 -def is_selendroid_running(tlib_logger, adb_logger, log=True):
369 """ 370 Function to validate if webdriver is running and a connection can be established 371 372 @param tlib_logger: TLib logger 373 @type tlib_logger: logging.logger 374 @param adb_logger: ADB logger 375 @type adb_logger: logging.logger 376 @param log: When true, log debugging information 377 @type log: bool 378 """ 379 try: 380 if log: 381 tlib_logger.debug("Checking if Selendroid server is running") 382 response = requests.get("http://localhost:4444/wd/hub/status", timeout=10) 383 except requests.ConnectionError as e: 384 if log: 385 adb_logger.debug("Connection to Selendroid failed:\n%s" % e) 386 return False 387 388 return response.status_code == 200
389 390
391 -def get_android_app_id():
392 response = requests.get("http://localhost:4444/wd/hub/status") 393 if not response: 394 return None 395 396 parser = jsonpath_rw.parse('$.value.supportedApps[*].appId') 397 apps = parser.find(response.json())[0].value 398 for app in apps: 399 if not app.startswith('io.selendroid.androiddriver'): 400 return app 401 return None
402 403
404 -def log_adb_output(logger, out):
405 """ 406 Logs ADB output 407 408 @param logger: ADB logger 409 @type logger: logging.logger 410 """ 411 if out[0] is not None and out[0] != '': 412 logger.debug("\n" + out[0]) 413 414 if out[1] is not None and out[1] != '': 415 logger.debug("\n" + out[1])
416