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