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

Source Code for Module tlib.base.AndroidHelper

  1  # noinspection PyPackageRequirements 
  2  import requests 
  3  import time 
  4  # noinspection PyPackageRequirements 
  5  import pytest 
  6  import re 
  7  from _pytest import runner 
  8  from tlib.base import TestHelper 
  9  import logging 
 10   
 11  __author__ = 'Andres Osorio' 
 12   
 13  DEVICE_UNPLUGGED = "disconnected"    # Device is not connected 
 14  DEVICE_OFFLINE = "offline"           # Device is connected but offline 
 15  DEVICE_UNAUTHORIZED = 'unauthorized' # Unauthorized 
 16  DEVICE_ONLINE = "online"             # Device is connected and accessible 
 17   
18 -class UnknownDeviceStatus(Exception):
19 """Exception thrown when adb device command returns an unknown status""" 20 pass
21 22
23 -def is_webdriver_running(tlib_logger, adb_logger, log=True):
24 """ 25 Function to validate if webdriver is running and a connection can be established 26 27 @param tlib_logger: TLib logger 28 @type tlib_logger: logging.logger 29 @param adb_logger: ADB logger 30 @type adb_logger: logging.logger 31 @param log: When true, log debugging information 32 @type log: bool 33 """ 34 try: 35 if log: tlib_logger.debug("Checking if Webdriver is running") 36 response = requests.get("http://localhost:8080/wd/hub/status", timeout=10) 37 except requests.ConnectionError as e: 38 if log: adb_logger.debug("Connection to Webdriver failed:\n%s" % e) 39 return False 40 41 return response.status_code == 200
42 43
44 -def get_device_status(tlib_logger, adb_logger, serial_id):
45 """ 46 Get status of a device using it's serial id\n 47 Serial id can be either a ID (for devices connected to USB) or an IP and port (For devices connected via IP) 48 49 @param tlib_logger: TLib logger 50 @type tlib_logger: logging.logger 51 @param adb_logger: ADB logger 52 @type adb_logger: logging.logger 53 @param serial_id: Device's serial number 54 @type serial_id: str 55 """ 56 57 out = TestHelper.run_command(adb_logger, ["adb", "devices"], fail_on_error=False) 58 if re.search(serial_id, out[0]) is None: 59 tlib_logger.debug("Device {serial_id} is not connected".format(serial_id=serial_id)) 60 return DEVICE_UNPLUGGED 61 elif re.search("{serial_id}\s+offline".format(serial_id=serial_id), out[0]) is not None: 62 tlib_logger.debug("Device {serial_id} is offline".format(serial_id=serial_id)) 63 return DEVICE_OFFLINE 64 elif re.search("{serial_id}\s+unauthorized".format(serial_id=serial_id), out[0]) is not None: 65 tlib_logger.debug("Device {serial_id} is offline".format(serial_id=serial_id)) 66 return DEVICE_UNAUTHORIZED 67 elif re.search("{serial_id}\s+device".format(serial_id=serial_id), out[0]) is not None: 68 tlib_logger.debug("Device {serial_id} is online".format(serial_id=serial_id)) 69 return DEVICE_ONLINE 70 else: 71 tlib_logger.error("Unknown device status\n%s" % out[0]) 72 raise UnknownDeviceStatus
73 74
75 -def setup_ip_connection(tlib_logger, adb_logger, serial_id):
76 """ 77 Connects to an android device via IP and waits for the connection to be established. 78 79 @param tlib_logger: TLib logger 80 @type tlib_logger: logging.logger 81 @param adb_logger: ADB logger 82 @type adb_logger: logging.logger 83 @param serial_id: Device's serial number 84 @type serial_id: str 85 """ 86 # Validate is device is already connected 87 tlib_logger.debug("Setting up IP connection to device {serial_id}".format(serial_id=serial_id)) 88 status = get_device_status(tlib_logger, adb_logger, serial_id) 89 90 if status == DEVICE_ONLINE: 91 tlib_logger.debug("Device is already online".format(serial_id=serial_id)) 92 return 93 94 # If device is offline or unauthorized, disconnect 95 if status == DEVICE_OFFLINE: 96 tlib_logger.warn("Device {serial_id} is offline, disconnecting and retrying".format(serial_id=serial_id)) 97 terminate_ip_connection(tlib_logger, adb_logger, serial_id) 98 elif status == DEVICE_UNAUTHORIZED: 99 terminate_ip_connection(tlib_logger, adb_logger, serial_id) 100 tlib_logger.error("Device {serial_id} is not authorized, aborting".format(serial_id=serial_id)) 101 pytest.fail("Device {serial_id} is not authorized, aborting".format(serial_id=serial_id)) 102 103 timeout = 0.5 104 105 #We wait up to 3 seconds 106 for i in range(1, int(3 / timeout)): 107 out = TestHelper.run_command(adb_logger, ["adb", "connect", serial_id]) 108 log_adb_output(adb_logger, out) 109 110 if get_device_status(tlib_logger, adb_logger, serial_id) == DEVICE_ONLINE: 111 tlib_logger.debug("Connected to device") 112 return 113 else: 114 tlib_logger.debug("Not yet connected") 115 time.sleep(timeout) 116 117 #If we're here, it means connection failed 118 adb_logger.error(r"Timeout while connecting to device {serial_id}\nADB output:\n{out}".format(serial_id=serial_id, out=out)) 119 # noinspection PyUnresolvedReferences 120 pytest.fail(r"Timeout while connecting to device {serial_id}\nADB output:\n{out}".format(serial_id=serial_id, out=out))
121 122
123 -def terminate_ip_connection(tlib_logger, adb_logger, serial_id):
124 """ 125 Disconnects from device 126 127 @param tlib_logger: TLib logger 128 @type tlib_logger: logging.logger 129 @param adb_logger: ADB logger 130 @type adb_logger: logging.logger 131 @param serial_id: Device's serial number 132 @type serial_id: str 133 """ 134 tlib_logger.debug("Disconnecting from device {serial_id}".format(serial_id=serial_id)) 135 out = TestHelper.run_command(adb_logger, ["adb", "disconnect", serial_id]) 136 log_adb_output(adb_logger, out)
137 138
139 -def close_webdriver(tlib_logger, adb_logger, serial_id):
140 """ 141 Stops Webdriver app o the device 142 143 @param tlib_logger: TLib logger 144 @type tlib_logger: logging.logger 145 @param adb_logger: ADB logger 146 @type adb_logger: logging.logger 147 @param serial_id: Device's serial number 148 @type serial_id: str 149 """ 150 tlib_logger.debug("Closing Webdriver on {serial_id}".format(serial_id=serial_id)) 151 if serial_id is None or serial_id == "": 152 out = TestHelper.run_command(adb_logger, 153 ["adb", "shell", "am", "force-stop", "org.openqa.selenium.android.app"]) 154 else: 155 out = TestHelper.run_command(adb_logger, 156 ["adb", "-s", serial_id, "shell", "am", "force-stop", "org.openqa.selenium.android.app"]) 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"]) 174 else: 175 out = TestHelper.run_command(adb_logger, ["adb", "-s", serial_id, "forward", "tcp:8080", "tcp:8080"]) 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"]) 190 log_adb_output(adb_logger, out)
191 192
193 -def stop_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("Stopping ADB") 204 out = TestHelper.run_command(adb_logger, ["adb", "kill-server"], fail_on_error=False) 205 log_adb_output(adb_logger, out)
206 207
208 -def start_webdriver(tlib_logger, adb_logger, serial_id):
209 """ 210 Starts Webdriver app on the device 211 212 @param tlib_logger: TLib logger 213 @type tlib_logger: logging.logger 214 @param adb_logger: ADB logger 215 @type adb_logger: logging.logger 216 @param serial_id: Device's serial number 217 @type serial_id: str 218 """ 219 tlib_logger.info("Starting webdriver on the device") 220 if serial_id is None or serial_id == "": 221 out = TestHelper.run_command(adb_logger, ["adb", "shell", "am", "start", "-a", "android.intent.action.MAIN", 222 "-n", "org.openqa.selenium.android.app/.MainActivity", "-e", "debug", "true"]) 223 else: 224 out = TestHelper.run_command(adb_logger, 225 ["adb", "-s", serial_id, "shell", "am", "start", "-a", "android.intent.action.MAIN", 226 "-n", "org.openqa.selenium.android.app/.MainActivity", 227 "-e", "debug", "true"]) 228 log_adb_output(adb_logger, out)
229 230
231 -def wait_for_connection_to_webdriver(tlib_logger, adb_logger):
232 """ 233 Waits up to 3 seconds for a connection to Webdriver 234 235 @param tlib_logger: TLib logger 236 @type tlib_logger: logging.logger 237 @param adb_logger: ADB logger 238 @type adb_logger: logging.logger 239 """ 240 #Check connection to webdriver can be established. Retry 3 times 241 tlib_logger.debug("Waiting for connection to Webdriver") 242 timeout = 0.5 243 for i in range(1, int(3 / timeout)): 244 if is_webdriver_running(tlib_logger, adb_logger, False): 245 tlib_logger.debug("Webdriver started successfully") 246 break 247 tlib_logger.debug("Can't connect to Webdriver, retrying in {timeout} seconds".format(timeout=timeout)) 248 time.sleep(timeout) 249 250 if not is_webdriver_running(tlib_logger, adb_logger, False): 251 tlib_logger.error("Couldn't start Webdriver. Make sure it's installed and running\n" 252 "See https://code.google.com/p/selenium/wiki/AndroidDriver#Setup_the_Emulator for more details") 253 # noinspection PyUnresolvedReferences 254 pytest.fail("Couldn't start Webdriver. Make sure it's installed and running\n" 255 "See https://code.google.com/p/selenium/wiki/AndroidDriver#Setup_the_Emulator for more details")
256 257 258 # noinspection PyUnresolvedReferences
259 -def setup_webdriver(tlib_logger, adb_logger, serial_id):
260 """ 261 Connects to a device and starts webdriver 262 263 @param tlib_logger: TLib logger 264 @type tlib_logger: logging.logger 265 @param adb_logger: ADB logger 266 @type adb_logger: logging.logger 267 @param serial_id: Device's serial number 268 @type serial_id: str 269 """ 270 #Try to connect to Webdriver and exit if success 271 if is_webdriver_running(tlib_logger, adb_logger, log=False): 272 tlib_logger.debug("Already connected to device") 273 return 274 275 tlib_logger.info("Connecting to Webdriver") 276 277 #Connect to device and setup port forwarding. 278 if TestHelper.is_valid_ip(serial_id): 279 setup_ip_connection(tlib_logger, adb_logger, serial_id) 280 281 setup_port_forwarding(tlib_logger, adb_logger, serial_id) 282 283 #Try again to connect to Webdriver and exit if success 284 if is_webdriver_running(tlib_logger, adb_logger): 285 tlib_logger.debug("Connected to Webdriver") 286 return 287 288 # Webdriver not running, start it 289 start_webdriver(tlib_logger, adb_logger, serial_id) 290 291 #Ensure we're connected 292 wait_for_connection_to_webdriver(tlib_logger, adb_logger) 293 294 tlib_logger.info("Connection to Webdriver established")
295 296 297 # noinspection PyUnresolvedReferences
298 -def teardown_webdriver(tlib_logger, adb_logger, serial_id):
299 """ 300 Closes webdriver and disconnects from device 301 302 @param tlib_logger: TLib logger 303 @type tlib_logger: logging.logger 304 @param adb_logger: ADB logger 305 @type adb_logger: logging.logger 306 @param serial_id: Device's serial number 307 @type serial_id: str 308 """ 309 #close selenium WebDriver 310 tlib_logger.info("Disconnecting from Webdriver") 311 close_webdriver(tlib_logger, adb_logger, serial_id) 312 313 # Terminates IP connection if an IP was given 314 if TestHelper.is_valid_ip(serial_id): 315 terminate_ip_connection(tlib_logger, adb_logger, serial_id) 316 317 #Stop adb service so it won't lock files required by TeamCity 318 try: 319 stop_adb_server(tlib_logger, adb_logger) 320 except runner.Failed: 321 #If ADB fails to stop, don't abort test 322 adb_logger.warn("Error stopping ADB server") 323 324 tlib_logger.info("Disconnected from Webdriver")
325 326
327 -def log_adb_output(logger, out):
328 """ 329 Logs ADB output 330 331 @param logger: ADB logger 332 @type logger: logging.logger 333 """ 334 if out[0] is not None and out[0] != '': 335 logger.debug("\n" + out[0]) 336 337 if out[1] is not None and out[1] != '': 338 logger.debug("\n" + out[1])
339