1
2 import requests
3 import time
4
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"
14 DEVICE_OFFLINE = "offline"
15 DEVICE_UNAUTHORIZED = 'unauthorized'
16 DEVICE_ONLINE = "online"
17
19 """Exception thrown when adb device command returns an unknown status"""
20 pass
21
22
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
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
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
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
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
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
118 adb_logger.error(r"Timeout while connecting to device {serial_id}\nADB output:\n{out}".format(serial_id=serial_id, out=out))
119
120 pytest.fail(r"Timeout while connecting to device {serial_id}\nADB output:\n{out}".format(serial_id=serial_id, out=out))
121
122
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
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
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
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
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
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
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
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
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
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
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
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
284 if is_webdriver_running(tlib_logger, adb_logger):
285 tlib_logger.debug("Connected to Webdriver")
286 return
287
288
289 start_webdriver(tlib_logger, adb_logger, serial_id)
290
291
292 wait_for_connection_to_webdriver(tlib_logger, adb_logger)
293
294 tlib_logger.info("Connection to Webdriver established")
295
296
297
325
326
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