xref: /aosp_15_r20/system/extras/torq/tests/device_unit_test.py (revision 288bf5226967eb3dac5cce6c939ccc2a7f2b4fe5) 
1#
2# Copyright (C) 2024 The Android Open Source Project
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8#      http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16
17import unittest
18import os
19import subprocess
20from unittest import mock
21from command import ProfilerCommand
22from device import AdbDevice
23
24TEST_DEVICE_SERIAL = "test-device-serial"
25TEST_DEVICE_SERIAL2 = "test-device-serial2"
26TEST_FILE_PATH = "test-file-path"
27TEST_STRING_FILE = "test-string-file"
28TEST_FAILURE_MSG = "test-failure"
29TEST_EXCEPTION = Exception(TEST_FAILURE_MSG)
30TEST_USER_ID_1 = 0
31TEST_USER_ID_2 = 1
32TEST_USER_ID_3 = 2
33TEST_PACKAGE_1 = "test-package-1"
34TEST_PACKAGE_2 = "test-package-2"
35TEST_PROP = "test-prop"
36TEST_PROP_VALUE = "test-prop-value"
37TEST_PID_OUTPUT = b"8241\n"
38BOOT_COMPLETE_OUTPUT = b"1\n"
39ANDROID_SDK_VERSION_T = 33
40
41class DeviceUnitTest(unittest.TestCase):
42
43  @staticmethod
44  def generate_adb_devices_result(devices, adb_started=True):
45    devices = [device.encode('utf-8') for device in devices]
46    stdout_string = b'List of devices attached\n'
47    if not adb_started:
48      stdout_string = (b'* daemon not running; starting now at tcp:1234\n'
49                       b'* daemon started successfully\n') + stdout_string
50    if len(devices) > 0:
51      stdout_string += b'\tdevice\n'.join(devices) + b'\tdevice\n'
52      stdout_string += b'\n'
53    return subprocess.CompletedProcess(args=['adb', 'devices'], returncode=0,
54                                       stdout=stdout_string)
55
56  @staticmethod
57  def generate_mock_completed_process(stdout_string=b'\n', stderr_string=b'\n'):
58    return mock.create_autospec(subprocess.CompletedProcess, instance=True,
59                                stdout=stdout_string, stderr=stderr_string)
60
61  @staticmethod
62  def subprocess_output(first_return_value, polling_return_value):
63    # Mocking the return value of a call to adb root and the return values of
64    # many followup calls to adb devices
65    yield first_return_value
66    while True:
67      yield polling_return_value
68
69  @staticmethod
70  def mock_users():
71    return mock.create_autospec(subprocess.CompletedProcess, instance=True,
72                                stdout=(b'Users:\n\tUserInfo{%d:Driver:813}'
73                                        b' running\n\tUserInfo{%d:Driver:412}\n'
74                                        % (TEST_USER_ID_1, TEST_USER_ID_2)))
75
76  @staticmethod
77  def mock_packages():
78    return mock.create_autospec(subprocess.CompletedProcess, instance=True,
79                                stdout=(b'package:%b\npackage:%b\n'
80                                        % (TEST_PACKAGE_1.encode("utf-8"),
81                                           TEST_PACKAGE_2.encode("utf-8"))))
82
83  @mock.patch.object(subprocess, "run", autospec=True)
84  def test_get_adb_devices_returns_devices(self, mock_subprocess_run):
85    mock_subprocess_run.return_value = (
86        self.generate_adb_devices_result([TEST_DEVICE_SERIAL,
87                                          TEST_DEVICE_SERIAL2]))
88    adbDevice = AdbDevice(None)
89
90    devices = adbDevice.get_adb_devices()
91
92    self.assertEqual(len(devices), 2)
93    self.assertEqual(devices[0], TEST_DEVICE_SERIAL)
94    self.assertEqual(devices[1], TEST_DEVICE_SERIAL2)
95
96  @mock.patch.object(subprocess, "run", autospec=True)
97  def test_get_adb_devices_returns_devices_and_adb_not_started(self,
98      mock_subprocess_run):
99    mock_subprocess_run.return_value = (
100        self.generate_adb_devices_result([TEST_DEVICE_SERIAL,
101                                          TEST_DEVICE_SERIAL2], False))
102    adbDevice = AdbDevice(None)
103
104    devices = adbDevice.get_adb_devices()
105
106    self.assertEqual(len(devices), 2)
107    self.assertEqual(devices[0], TEST_DEVICE_SERIAL)
108    self.assertEqual(devices[1], TEST_DEVICE_SERIAL2)
109
110  @mock.patch.object(subprocess, "run", autospec=True)
111  def test_get_adb_devices_returns_no_device(self, mock_subprocess_run):
112    mock_subprocess_run.return_value = self.generate_adb_devices_result([])
113    adbDevice = AdbDevice(None)
114
115    devices = adbDevice.get_adb_devices()
116
117    self.assertEqual(devices, [])
118
119  @mock.patch.object(subprocess, "run", autospec=True)
120  def test_get_adb_devices_returns_no_device_and_adb_not_started(self,
121      mock_subprocess_run):
122    mock_subprocess_run.return_value = (
123        self.generate_adb_devices_result([], False))
124    adbDevice = AdbDevice(None)
125
126    devices = adbDevice.get_adb_devices()
127
128    self.assertEqual(devices, [])
129
130  @mock.patch.object(subprocess, "run", autospec=True)
131  def test_get_adb_devices_command_failure_error(self, mock_subprocess_run):
132    mock_subprocess_run.side_effect = TEST_EXCEPTION
133    adbDevice = AdbDevice(None)
134
135    with self.assertRaises(Exception) as e:
136      adbDevice.get_adb_devices()
137
138    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
139
140  @mock.patch.object(subprocess, "run", autospec=True)
141  def test_check_device_connection_serial_arg_in_devices(self,
142      mock_subprocess_run):
143    mock_subprocess_run.return_value = (
144        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
145    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
146
147    error = adbDevice.check_device_connection()
148
149    self.assertEqual(error, None)
150
151  @mock.patch.object(subprocess, "run", autospec=True)
152  def test_check_device_connection_serial_arg_not_in_devices_error(self,
153      mock_subprocess_run):
154    mock_subprocess_run.return_value = (
155        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
156    invalid_device_serial = "invalid-device-serial"
157    adbDevice = AdbDevice(invalid_device_serial)
158
159    error = adbDevice.check_device_connection()
160
161    self.assertNotEqual(error, None)
162    self.assertEqual(error.message, ("Device with serial %s is not connected."
163                                     % invalid_device_serial))
164    self.assertEqual(error.suggestion, None)
165
166  @mock.patch.dict(os.environ, {"ANDROID_SERIAL": TEST_DEVICE_SERIAL},
167                   clear=True)
168  @mock.patch.object(subprocess, "run", autospec=True)
169  def test_check_device_connection_env_variable_in_devices(self,
170      mock_subprocess_run):
171    mock_subprocess_run.return_value = (
172        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
173    adbDevice = AdbDevice(None)
174
175    error = adbDevice.check_device_connection()
176
177    self.assertEqual(error, None)
178    self.assertEqual(adbDevice.serial, TEST_DEVICE_SERIAL)
179
180  @mock.patch.dict(os.environ, {"ANDROID_SERIAL": "invalid-device-serial"},
181                   clear=True)
182  @mock.patch.object(subprocess, "run", autospec=True)
183  def test_check_device_connection_env_variable_not_in_devices_error(self,
184      mock_subprocess_run):
185    mock_subprocess_run.return_value = (
186        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
187    adbDevice = AdbDevice(None)
188
189    error = adbDevice.check_device_connection()
190
191    self.assertNotEqual(error, None)
192    self.assertEqual(error.message, ("Device with serial invalid-device-serial"
193                                     " is set as environment variable,"
194                                     " ANDROID_SERIAL, but is not connected."))
195    self.assertEqual(error.suggestion, None)
196
197  @mock.patch.object(subprocess, "run", autospec=True)
198  def test_check_device_connection_adb_devices_command_fails_error(self,
199      mock_subprocess_run):
200    mock_subprocess_run.side_effect = TEST_EXCEPTION
201    adbDevice = AdbDevice(None)
202
203    with self.assertRaises(Exception) as e:
204      adbDevice.check_device_connection()
205
206    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
207
208  @mock.patch.object(subprocess, "run", autospec=True)
209  def test_check_device_connection_no_devices_connected_error(self,
210      mock_subprocess_run):
211    mock_subprocess_run.return_value = (self.generate_adb_devices_result([]))
212    adbDevice = AdbDevice(None)
213
214    error = adbDevice.check_device_connection()
215
216    self.assertNotEqual(error, None)
217    self.assertEqual(error.message, "There are currently no devices connected.")
218    self.assertEqual(error.suggestion, None)
219
220  @mock.patch.object(subprocess, "run", autospec=True)
221  def test_check_device_connection_no_devices_connected_adb_not_started_error(
222      self, mock_subprocess_run):
223    mock_subprocess_run.return_value = (
224        self.generate_adb_devices_result([], False))
225    adbDevice = AdbDevice(None)
226
227    error = adbDevice.check_device_connection()
228
229    self.assertNotEqual(error, None)
230    self.assertEqual(error.message, "There are currently no devices connected.")
231    self.assertEqual(error.suggestion, None)
232
233  @mock.patch.dict(os.environ, {}, clear=True)
234  @mock.patch.object(subprocess, "run", autospec=True)
235  def test_check_device_connection_only_one_device(self, mock_subprocess_run):
236    mock_subprocess_run.return_value = (
237        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
238    adbDevice = AdbDevice(None)
239
240    error = adbDevice.check_device_connection()
241
242    self.assertEqual(error, None)
243    self.assertEqual(adbDevice.serial, TEST_DEVICE_SERIAL)
244
245  @mock.patch.dict(os.environ, {}, clear=True)
246  @mock.patch.object(subprocess, "run", autospec=True)
247  def test_check_device_connection_multiple_devices_error(self,
248      mock_subprocess_run):
249    mock_subprocess_run.return_value = (
250        self.generate_adb_devices_result([TEST_DEVICE_SERIAL,
251                                          TEST_DEVICE_SERIAL2]))
252    adbDevice = AdbDevice(None)
253
254    error = adbDevice.check_device_connection()
255
256    self.assertNotEqual(error, None)
257    self.assertEqual(error.message, ("There is more than one device currently"
258                                     " connected."))
259    self.assertEqual(error.suggestion, ("Run one of the following commands to"
260                                        " choose one of the connected devices:"
261                                        "\n\t torq --serial %s"
262                                        "\n\t torq --serial %s"
263                                        % (TEST_DEVICE_SERIAL,
264                                           TEST_DEVICE_SERIAL2)))
265
266  @mock.patch.object(subprocess, "run", autospec=True)
267  def test_root_device_success(self, mock_subprocess_run):
268    mock_subprocess_run.side_effect = [
269        self.generate_mock_completed_process(),
270        self.generate_adb_devices_result([TEST_DEVICE_SERIAL])]
271    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
272
273    # No exception is expected to be thrown
274    adbDevice.root_device()
275
276  @mock.patch.object(subprocess, "run", autospec=True)
277  def test_root_device_failure(self, mock_subprocess_run):
278    mock_subprocess_run.side_effect = TEST_EXCEPTION
279    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
280
281    with self.assertRaises(Exception) as e:
282      adbDevice.root_device()
283
284    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
285
286  @mock.patch.object(subprocess, "run", autospec=True)
287  def test_root_device_times_out_error(self, mock_subprocess_run):
288    mock_subprocess_run.side_effect = lambda args, capture_output=True: (
289        next(self.subprocess_output(self.generate_adb_devices_result([]),
290                                    self.generate_mock_completed_process())))
291    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
292
293    with self.assertRaises(Exception) as e:
294      adbDevice.root_device()
295
296    self.assertEqual(str(e.exception), ("Device with serial %s took too long to"
297                                        " reconnect after being rooted."
298                                        % TEST_DEVICE_SERIAL))
299
300  @mock.patch.object(subprocess, "run", autospec=True)
301  def test_root_device_and_adb_devices_fails_error(self, mock_subprocess_run):
302    mock_subprocess_run.side_effect = [self.generate_mock_completed_process(),
303                                       TEST_EXCEPTION]
304    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
305
306    with self.assertRaises(Exception) as e:
307      adbDevice.root_device()
308
309    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
310
311  @mock.patch.object(subprocess, "run", autospec=True)
312  def test_remove_file_success(self, mock_subprocess_run):
313    mock_subprocess_run.return_value = self.generate_mock_completed_process()
314    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
315
316    # No exception is expected to be thrown
317    adbDevice.remove_file(TEST_FILE_PATH)
318
319  @mock.patch.object(subprocess, "run", autospec=True)
320  def test_remove_file_failure(self, mock_subprocess_run):
321    mock_subprocess_run.side_effect = TEST_EXCEPTION
322    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
323
324    with self.assertRaises(Exception) as e:
325      adbDevice.remove_file(TEST_FILE_PATH)
326
327    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
328
329  @mock.patch.object(subprocess, "Popen", autospec=True)
330  def test_start_perfetto_trace_success(self, mock_subprocess_popen):
331    # Mocking the return value of subprocess.Popen to ensure it's
332    # not modified and returned by AdbDevice.start_perfetto_trace
333    mock_subprocess_popen.return_value = mock.Mock()
334    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
335
336    mock_process = adbDevice.start_perfetto_trace(None)
337
338    # No exception is expected to be thrown
339    self.assertEqual(mock_process, mock_subprocess_popen.return_value)
340
341  @mock.patch.object(subprocess, "Popen", autospec=True)
342  def test_start_perfetto_trace_failure(self, mock_subprocess_popen):
343    mock_subprocess_popen.side_effect = TEST_EXCEPTION
344    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
345
346    with self.assertRaises(Exception) as e:
347      adbDevice.start_perfetto_trace(None)
348
349    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
350
351  @mock.patch.object(subprocess, "Popen", autospec=True)
352  def test_start_simpleperf_trace_success(self, mock_subprocess_popen):
353    # Mocking the return value of subprocess.Popen to ensure it's
354    # not modified and returned by AdbDevice.start_simpleperf_trace
355    mock_subprocess_popen.return_value = mock.Mock()
356    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
357    command = ProfilerCommand("profiler", "custom", None, None,
358                              10000, None, None, ["cpu-cycles"], None, None,
359                              None, None, None, None, None)
360    mock_process = adbDevice.start_simpleperf_trace(command)
361
362    # No exception is expected to be thrown
363    self.assertEqual(mock_process, mock_subprocess_popen.return_value)
364
365  @mock.patch.object(subprocess, "Popen", autospec=True)
366  def test_start_simpleperf_trace_failure(self, mock_subprocess_popen):
367    mock_subprocess_popen.side_effect = TEST_EXCEPTION
368    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
369
370    command = ProfilerCommand("profiler", "custom", None, None,
371                              10000, None, None, ["cpu-cycles"], None, None,
372                              None, None, None, None, None)
373    with self.assertRaises(Exception) as e:
374      adbDevice.start_simpleperf_trace(command)
375
376    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
377
378
379  @mock.patch.object(subprocess, "run", autospec=True)
380  def test_pull_file_success(self, mock_subprocess_run):
381    mock_subprocess_run.return_value = self.generate_mock_completed_process()
382    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
383
384    # No exception is expected to be thrown
385    adbDevice.pull_file(TEST_FILE_PATH, TEST_FILE_PATH)
386
387  @mock.patch.object(subprocess, "run", autospec=True)
388  def test_pull_file_failure(self, mock_subprocess_run):
389    mock_subprocess_run.side_effect = TEST_EXCEPTION
390    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
391
392    with self.assertRaises(Exception) as e:
393      adbDevice.pull_file(TEST_FILE_PATH, TEST_FILE_PATH)
394
395    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
396
397  @mock.patch.object(subprocess, "run", autospec=True)
398  def test_get_all_users_success(self, mock_subprocess_run):
399    mock_subprocess_run.return_value = self.mock_users()
400
401    users = AdbDevice(TEST_DEVICE_SERIAL).get_all_users()
402
403    self.assertEqual(users, [TEST_USER_ID_1, TEST_USER_ID_2])
404
405  @mock.patch.object(subprocess, "run", autospec=True)
406  def test_get_all_users_failure(self, mock_subprocess_run):
407    mock_subprocess_run.side_effect = TEST_EXCEPTION
408    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
409
410    with self.assertRaises(Exception) as e:
411      adbDevice.get_all_users()
412
413    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
414
415  @mock.patch.object(subprocess, "run", autospec=True)
416  def test_user_exists_success(self, mock_subprocess_run):
417    mock_subprocess_run.return_value = self.mock_users()
418    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
419
420    error = adbDevice.user_exists(TEST_USER_ID_1)
421
422    self.assertEqual(error, None)
423
424  @mock.patch.object(subprocess, "run", autospec=True)
425  def test_user_exists_and_user_does_not_exist_failure(self,
426      mock_subprocess_run):
427    mock_subprocess_run.return_value = self.mock_users()
428    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
429
430    error = adbDevice.user_exists(TEST_USER_ID_3)
431
432    self.assertNotEqual(error, None)
433    self.assertEqual(error.message, ("User ID %s does not exist on device with"
434                                     " serial %s." % (TEST_USER_ID_3,
435                                                      TEST_DEVICE_SERIAL)))
436    self.assertEqual(error.suggestion,
437                     ("Select from one of the following user IDs on device with"
438                      " serial %s: %s, %s"
439                      % (TEST_DEVICE_SERIAL, TEST_USER_ID_1, TEST_USER_ID_2)))
440
441  @mock.patch.object(subprocess, "run", autospec=True)
442  def test_user_exists_and_get_all_users_fails_error(self, mock_subprocess_run):
443    mock_subprocess_run.side_effect = TEST_EXCEPTION
444    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
445
446    with self.assertRaises(Exception) as e:
447      adbDevice.user_exists(TEST_USER_ID_1)
448
449    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
450
451  @mock.patch.object(subprocess, "run", autospec=True)
452  def test_get_current_user_success(self, mock_subprocess_run):
453    mock_subprocess_run.return_value = (
454        mock.create_autospec(subprocess.CompletedProcess, instance=True,
455                             stdout=b'%d\n' % TEST_USER_ID_1))
456    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
457
458    user = adbDevice.get_current_user()
459
460    self.assertEqual(user, TEST_USER_ID_1)
461
462  @mock.patch.object(subprocess, "run", autospec=True)
463  def test_get_current_user_failure(self, mock_subprocess_run):
464    mock_subprocess_run.side_effect = TEST_EXCEPTION
465    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
466
467    with self.assertRaises(Exception) as e:
468      adbDevice.get_current_user()
469
470    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
471
472  @mock.patch.object(subprocess, "run", autospec=True)
473  def test_perform_user_switch_success(self, mock_subprocess_run):
474    mock_subprocess_run.return_value = self.generate_mock_completed_process()
475    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
476
477    # No exception is expected to be thrown
478    adbDevice.perform_user_switch(TEST_USER_ID_1)
479
480  @mock.patch.object(subprocess, "run", autospec=True)
481  def test_perform_user_switch_failure(self, mock_subprocess_run):
482    mock_subprocess_run.side_effect = TEST_EXCEPTION
483    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
484
485    with self.assertRaises(Exception) as e:
486      adbDevice.perform_user_switch(TEST_USER_ID_1)
487
488    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
489
490  @mock.patch.object(subprocess, "run", autospec=True)
491  def test_write_to_file_success(self, mock_subprocess_run):
492    mock_subprocess_run.return_value = self.generate_mock_completed_process()
493    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
494
495    # No exception is expected to be thrown
496    adbDevice.write_to_file(TEST_FILE_PATH, TEST_STRING_FILE)
497
498  @mock.patch.object(subprocess, "run", autospec=True)
499  def test_write_to_file_failure(self, mock_subprocess_run):
500    mock_subprocess_run.side_effect = TEST_EXCEPTION
501    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
502
503    with self.assertRaises(Exception) as e:
504      adbDevice.write_to_file(TEST_FILE_PATH, TEST_STRING_FILE)
505
506    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
507
508  @mock.patch.object(subprocess, "run", autospec=True)
509  def test_set_prop_success(self, mock_subprocess_run):
510    mock_subprocess_run.return_value = self.generate_mock_completed_process()
511    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
512
513    # No exception is expected to be thrown
514    adbDevice.set_prop(TEST_PROP, TEST_PROP_VALUE)
515
516  @mock.patch.object(subprocess, "run", autospec=True)
517  def test_set_prop_failure(self, mock_subprocess_run):
518    mock_subprocess_run.side_effect = TEST_EXCEPTION
519    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
520
521    with self.assertRaises(Exception) as e:
522      adbDevice.set_prop(TEST_PROP, TEST_PROP_VALUE)
523
524    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
525
526  @mock.patch.object(subprocess, "run", autospec=True)
527  def test_reboot_success(self, mock_subprocess_run):
528    mock_subprocess_run.return_value = self.generate_mock_completed_process()
529    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
530
531    # No exception is expected to be thrown
532    adbDevice.reboot()
533
534  @mock.patch.object(subprocess, "run", autospec=True)
535  def test_reboot_failure(self, mock_subprocess_run):
536    mock_subprocess_run.side_effect = TEST_EXCEPTION
537    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
538
539    with self.assertRaises(Exception) as e:
540      adbDevice.reboot()
541
542    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
543
544  @mock.patch.object(subprocess, "run", autospec=True)
545  def test_wait_for_device_success(self, mock_subprocess_run):
546    mock_subprocess_run.return_value = self.generate_mock_completed_process()
547    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
548
549    # No exception is expected to be thrown
550    adbDevice.wait_for_device()
551
552  @mock.patch.object(subprocess, "run", autospec=True)
553  def test_wait_for_device_failure(self, mock_subprocess_run):
554    mock_subprocess_run.side_effect = TEST_EXCEPTION
555    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
556
557    with self.assertRaises(Exception) as e:
558      adbDevice.wait_for_device()
559
560    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
561
562  @mock.patch.object(subprocess, "run", autospec=True)
563  def test_is_boot_completed_and_is_completed(self, mock_subprocess_run):
564    mock_subprocess_run.return_value = (
565        self.generate_mock_completed_process(BOOT_COMPLETE_OUTPUT))
566    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
567
568    is_completed = adbDevice.is_boot_completed()
569
570    self.assertEqual(is_completed, True)
571
572  @mock.patch.object(subprocess, "run", autospec=True)
573  def test_is_boot_completed_and_is_not_completed(self, mock_subprocess_run):
574    mock_subprocess_run.return_value = self.generate_mock_completed_process()
575    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
576
577    is_completed = adbDevice.is_boot_completed()
578
579    self.assertEqual(is_completed, False)
580
581  @mock.patch.object(subprocess, "run", autospec=True)
582  def test_is_boot_completed_failure(self, mock_subprocess_run):
583    mock_subprocess_run.side_effect = TEST_EXCEPTION
584    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
585
586    with self.assertRaises(Exception) as e:
587      adbDevice.is_boot_completed()
588
589    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
590
591  @mock.patch.object(subprocess, "run", autospec=True)
592  def test_wait_for_boot_to_complete_success(self, mock_subprocess_run):
593    mock_subprocess_run.return_value = (
594        self.generate_mock_completed_process(BOOT_COMPLETE_OUTPUT))
595    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
596
597    # No exception is expected to be thrown
598    adbDevice.wait_for_boot_to_complete()
599
600  @mock.patch.object(subprocess, "run", autospec=True)
601  def test_wait_for_boot_to_complete_and_is_boot_completed_fails_error(self,
602      mock_subprocess_run):
603    mock_subprocess_run.side_effect = TEST_EXCEPTION
604    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
605
606    with self.assertRaises(Exception) as e:
607      adbDevice.wait_for_boot_to_complete()
608
609    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
610
611  @mock.patch.object(subprocess, "run", autospec=True)
612  def test_wait_for_boot_to_complete_times_out_error(self, mock_subprocess_run):
613    mock_subprocess_run.return_value = self.generate_mock_completed_process()
614    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
615
616    with self.assertRaises(Exception) as e:
617      adbDevice.wait_for_boot_to_complete()
618
619    self.assertEqual(str(e.exception), ("Device with serial %s took too long to"
620                                        " finish rebooting."
621                                        % adbDevice.serial))
622
623  @mock.patch.object(subprocess, "run", autospec=True)
624  def test_get_packages_success(self, mock_subprocess_run):
625    mock_subprocess_run.return_value = self.mock_packages()
626    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
627
628    packages = adbDevice.get_packages()
629
630    self.assertEqual(packages, [TEST_PACKAGE_1, TEST_PACKAGE_2])
631
632  @mock.patch.object(subprocess, "run", autospec=True)
633  def test_get_packages_failure(self, mock_subprocess_run):
634    mock_subprocess_run.side_effect = TEST_EXCEPTION
635    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
636
637    with self.assertRaises(Exception) as e:
638      adbDevice.get_packages()
639
640    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
641
642  @mock.patch.object(subprocess, "run", autospec=True)
643  def test_get_pid_success(self, mock_subprocess_run):
644    mock_subprocess_run.return_value = self.generate_mock_completed_process(
645        TEST_PID_OUTPUT)
646    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
647
648    process_id = adbDevice.get_pid(TEST_PACKAGE_1)
649
650    self.assertEqual(process_id, "8241")
651
652  @mock.patch.object(subprocess, "run", autospec=True)
653  def test_get_pid_failure(self, mock_subprocess_run):
654    mock_subprocess_run.side_effect = TEST_EXCEPTION
655    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
656
657    with self.assertRaises(Exception) as e:
658      adbDevice.get_pid(TEST_PACKAGE_1)
659
660    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
661
662  @mock.patch.object(subprocess, "run", autospec=True)
663  def test_package_running(self, mock_subprocess_run):
664    mock_subprocess_run.return_value = self.generate_mock_completed_process(
665        TEST_PID_OUTPUT)
666    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
667
668    is_running = adbDevice.is_package_running(TEST_PACKAGE_1)
669
670    self.assertEqual(is_running, True)
671
672  @mock.patch.object(subprocess, "run", autospec=True)
673  def test_package_not_running(self, mock_subprocess_run):
674    mock_subprocess_run.return_value = self.generate_mock_completed_process()
675    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
676
677    is_running = adbDevice.is_package_running(TEST_PACKAGE_1)
678
679    self.assertEqual(is_running, False)
680
681  @mock.patch.object(subprocess, "run", autospec=True)
682  def test_package_running_and_get_pid_failure(self, mock_subprocess_run):
683    mock_subprocess_run.side_effect = TEST_EXCEPTION
684    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
685
686    with self.assertRaises(Exception) as e:
687      adbDevice.is_package_running(TEST_PACKAGE_1)
688
689    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
690
691  @mock.patch.object(subprocess, "run", autospec=True)
692  def test_start_package_success(self, mock_subprocess_run):
693    mock_subprocess_run.return_value = self.generate_mock_completed_process()
694    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
695
696    error = adbDevice.start_package(TEST_PACKAGE_1)
697
698    self.assertEqual(error, None)
699
700  @mock.patch.object(subprocess, "run", autospec=True)
701  def test_start_package_fails_with_service_app(self, mock_subprocess_run):
702    mock_subprocess_run.return_value = self.generate_mock_completed_process(
703        stderr_string=b'%s\n' % TEST_FAILURE_MSG.encode("utf-8"))
704    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
705
706    error = adbDevice.start_package(TEST_PACKAGE_1)
707
708    self.assertNotEqual(error, None)
709    self.assertEqual(error.message, ("Cannot start package %s on device with"
710                                     " serial %s because %s is a service"
711                                     " package, which doesn't implement a MAIN"
712                                     " activity." % (TEST_PACKAGE_1,
713                                                     TEST_DEVICE_SERIAL,
714                                                     TEST_PACKAGE_1)))
715    self.assertEqual(error.suggestion, None)
716
717  @mock.patch.object(subprocess, "run", autospec=True)
718  def test_start_package_failure(self, mock_subprocess_run):
719    mock_subprocess_run.side_effect = TEST_EXCEPTION
720    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
721
722    with self.assertRaises(Exception) as e:
723      adbDevice.start_package(TEST_PACKAGE_1)
724
725    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
726
727  @mock.patch.object(subprocess, "run", autospec=True)
728  def test_kill_pid_success(self, mock_subprocess_run):
729    mock_subprocess_run.side_effect = [
730        self.generate_mock_completed_process(TEST_PID_OUTPUT), None]
731    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
732
733    # No exception is expected to be thrown
734    adbDevice.kill_pid(TEST_PACKAGE_1)
735
736  @mock.patch.object(subprocess, "run", autospec=True)
737  def test_kill_pid_and_get_pid_failure(self, mock_subprocess_run):
738    mock_subprocess_run.side_effect = TEST_EXCEPTION
739    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
740
741    with self.assertRaises(Exception) as e:
742      adbDevice.kill_pid(TEST_PACKAGE_1)
743
744    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
745
746  @mock.patch.object(subprocess, "run", autospec=True)
747  def test_kill_pid_failure(self, mock_subprocess_run):
748    mock_subprocess_run.side_effect = [
749        self.generate_mock_completed_process(TEST_PID_OUTPUT), TEST_EXCEPTION]
750    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
751
752    with self.assertRaises(Exception) as e:
753      adbDevice.kill_pid(TEST_PACKAGE_1)
754
755    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
756
757  @mock.patch.object(subprocess, "run", autospec=True)
758  def test_force_stop_package_success(self, mock_subprocess_run):
759    mock_subprocess_run.return_value = None
760    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
761
762    # No exception is expected to be thrown
763    adbDevice.force_stop_package(TEST_PACKAGE_1)
764
765  @mock.patch.object(subprocess, "run", autospec=True)
766  def test_force_stop_package_failure(self, mock_subprocess_run):
767    mock_subprocess_run.side_effect = TEST_EXCEPTION
768    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
769
770    with self.assertRaises(Exception) as e:
771      adbDevice.force_stop_package(TEST_PACKAGE_1)
772
773    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
774
775  @mock.patch.object(subprocess, "run", autospec=True)
776  def test_get_prop_success(self, mock_subprocess_run):
777    test_prop_value = ANDROID_SDK_VERSION_T
778    mock_subprocess_run.return_value = self.generate_mock_completed_process(
779        stdout_string=b'%d\n' % test_prop_value)
780    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
781
782    prop_value = int(adbDevice.get_prop(TEST_PROP))
783
784    self.assertEqual(prop_value, test_prop_value)
785
786  @mock.patch.object(subprocess, "run", autospec=True)
787  def test_get_prop_package_failure(self, mock_subprocess_run):
788    mock_subprocess_run.side_effect = TEST_EXCEPTION
789    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
790
791    with self.assertRaises(Exception) as e:
792      adbDevice.get_prop(TEST_PROP)
793
794    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
795
796  @mock.patch.object(subprocess, "run", autospec=True)
797  def test_get_android_sdk_version_success(self, mock_subprocess_run):
798    mock_subprocess_run.return_value = self.generate_mock_completed_process(
799        stdout_string=b'%d\n' % ANDROID_SDK_VERSION_T)
800    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
801
802    prop_value = adbDevice.get_android_sdk_version()
803
804    self.assertEqual(prop_value, ANDROID_SDK_VERSION_T)
805
806  @mock.patch.object(subprocess, "run", autospec=True)
807  def test_get_android_sdk_version_failure(self, mock_subprocess_run):
808    mock_subprocess_run.side_effect = TEST_EXCEPTION
809    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
810
811    with self.assertRaises(Exception) as e:
812      adbDevice.get_android_sdk_version()
813
814    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)
815
816  @mock.patch.object(subprocess, "run", autospec=True)
817  def test_simpleperf_event_exists_success(self, mock_subprocess_run):
818    mock_subprocess_run.return_value = (
819        self.generate_mock_completed_process(b'List of software events:\n  '
820                                             b'alignment-faults\n  '
821                                             b'context-switches\n  '
822                                             b'cpu-clock\n  '
823                                             b'cpu-migrations\n  '
824                                             b'emulation-faults\n  '
825                                             b'major-faults\n  '
826                                             b'minor-faults\n  page-faults\n  '
827                                             b'task-clock'))
828    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
829
830    events = ["cpu-clock", "minor-faults"]
831    # No exception is expected to be thrown
832    error = adbDevice.simpleperf_event_exists(events)
833
834    self.assertEqual(error, None)
835    # Check that the list passed to the function is unchanged
836    self.assertEqual(events, ["cpu-clock", "minor-faults"])
837
838  @mock.patch.object(subprocess, "run", autospec=True)
839  def test_simpleperf_event_exists_failure(self, mock_subprocess_run):
840    mock_subprocess_run.return_value = (
841        self.generate_mock_completed_process(b'List of software events:\n  '
842                                             b'alignment-faults\n  '
843                                             b'context-switches\n  '
844                                             b'cpu-clock\n  '
845                                             b'cpu-migrations\n  '
846                                             b'emulation-faults\n  '
847                                             b'major-faults\n  '
848                                             b'minor-faults\n  page-faults\n  '
849                                             b'task-clock'))
850    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
851
852    error = adbDevice.simpleperf_event_exists(["cpu-clock", "minor-faults",
853                                               "List"])
854
855    self.assertEqual(error.message, "The following simpleperf event(s) are "
856                                    "invalid: ['List'].")
857    self.assertEqual(error.suggestion, "Run adb shell simpleperf list to"
858                                       " see valid simpleperf events.")
859
860
861if __name__ == '__main__':
862  unittest.main()
863