adb_compatibility_devicetest.py revision 33259e44c8229f70ffe0cf3bb5ca9375c4feb2f9 
1#!/usr/bin/env python
2# Copyright 2016 The Chromium Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6import contextlib
7import os
8import posixpath
9import random
10import signal
11import sys
12import unittest
13
14_CATAPULT_BASE_DIR = os.path.abspath(os.path.join(
15    os.path.dirname(__file__), '..', '..', '..', '..'))
16
17sys.path.append(os.path.join(_CATAPULT_BASE_DIR, 'devil'))
18from devil import devil_env
19from devil.android import device_errors
20from devil.android import device_test_case
21from devil.android.sdk import adb_wrapper
22from devil.utils import cmd_helper
23from devil.utils import timeout_retry
24
25
26_TEST_DATA_DIR = os.path.abspath(os.path.join(
27    os.path.dirname(__file__), 'test', 'data'))
28
29
30def _hostAdbPids():
31  ps_status, ps_output = cmd_helper.GetCmdStatusAndOutput(
32      ['pgrep', '-l', 'adb'])
33  if ps_status != 0:
34    return []
35
36  pids_and_names = (line.split() for line in ps_output.splitlines())
37  return [int(pid) for pid, name in pids_and_names
38          if name == 'adb']
39
40
41class AdbCompatibilityTest(device_test_case.DeviceTestCase):
42
43  @classmethod
44  def setUpClass(cls):
45    custom_adb_path = os.environ.get('ADB_PATH')
46    custom_deps = {
47      'config_type': 'BaseConfig',
48      'dependencies': {},
49    }
50    if custom_adb_path:
51      custom_deps['dependencies']['adb'] = {
52        'file_info': {
53          devil_env.GetPlatform(): {
54            'local_paths': [custom_adb_path],
55          },
56        },
57      }
58    devil_env.config.Initialize(configs=[custom_deps])
59
60  def testStartServer(self):
61    # Manually kill off any instances of adb.
62    adb_pids = _hostAdbPids()
63    for p in adb_pids:
64      os.kill(p, signal.SIGKILL)
65
66    self.assertIsNotNone(
67        timeout_retry.WaitFor(
68            lambda: not _hostAdbPids(), wait_period=0.1, max_tries=10))
69
70    # start the adb server
71    start_server_status, _ = cmd_helper.GetCmdStatusAndOutput(
72        [adb_wrapper.AdbWrapper.GetAdbPath(), 'start-server'])
73
74    # verify that the server is now online
75    self.assertEquals(0, start_server_status)
76    self.assertIsNotNone(
77        timeout_retry.WaitFor(
78            lambda: bool(_hostAdbPids()), wait_period=0.1, max_tries=10))
79
80  def testKillServer(self):
81    adb_pids = _hostAdbPids()
82    if not adb_pids:
83      adb_wrapper.AdbWrapper.StartServer()
84
85    adb_pids = _hostAdbPids()
86    self.assertGreaterEqual(len(adb_pids), 1)
87
88    kill_server_status, _ = cmd_helper.GetCmdStatusAndOutput(
89        [adb_wrapper.AdbWrapper.GetAdbPath(), 'kill-server'])
90    self.assertEqual(0, kill_server_status)
91
92    adb_pids = _hostAdbPids()
93    self.assertEqual(0, len(adb_pids))
94
95  def testDevices(self):
96    devices = adb_wrapper.AdbWrapper.Devices()
97    self.assertNotEqual(0, len(devices), 'No devices found.')
98
99  def getTestInstance(self):
100    """Creates a real AdbWrapper instance for testing."""
101    return adb_wrapper.AdbWrapper(self.serial)
102
103  def testShell(self):
104    under_test = self.getTestInstance()
105    shell_ls_result = under_test.Shell('ls')
106    self.assertIsInstance(shell_ls_result, str)
107    self.assertTrue(bool(shell_ls_result))
108
109  def testShell_failed(self):
110    under_test = self.getTestInstance()
111    with self.assertRaises(device_errors.AdbShellCommandFailedError):
112      under_test.Shell('ls /foo/bar/baz')
113
114  def testShell_externalStorageDefined(self):
115    under_test = self.getTestInstance()
116    external_storage = under_test.Shell('echo $EXTERNAL_STORAGE')
117    self.assertIsInstance(external_storage, str)
118    self.assertTrue(posixpath.isabs(external_storage))
119
120  @contextlib.contextmanager
121  def getTestPushDestination(self, under_test):
122    """Creates a temporary directory suitable for pushing to."""
123    external_storage = under_test.Shell('echo $EXTERNAL_STORAGE').strip()
124    if not external_storage:
125      self.skipTest('External storage not available.')
126    while True:
127      random_hex = hex(random.randint(0, 2 ** 52))[2:]
128      name = 'tmp_push_test%s' % random_hex
129      path = posixpath.join(external_storage, name)
130      try:
131        under_test.Shell('ls %s' % path)
132      except device_errors.AdbShellCommandFailedError:
133        break
134    under_test.Shell('mkdir %s' % path)
135    try:
136      yield path
137    finally:
138      under_test.Shell('rm -rf %s' % path)
139
140  def testPush_fileToFile(self):
141    under_test = self.getTestInstance()
142    with self.getTestPushDestination(under_test) as push_target_directory:
143      src = os.path.join(_TEST_DATA_DIR, 'push_file.txt')
144      dest = posixpath.join(push_target_directory, 'push_file.txt')
145      with self.assertRaises(device_errors.AdbShellCommandFailedError):
146        under_test.Shell('ls %s' % dest)
147      under_test.Push(src, dest)
148      self.assertEquals(dest, under_test.Shell('ls %s' % dest).strip())
149
150  def testPush_fileToDirectory(self):
151    under_test = self.getTestInstance()
152    with self.getTestPushDestination(under_test) as push_target_directory:
153      src = os.path.join(_TEST_DATA_DIR, 'push_file.txt')
154      dest = push_target_directory
155      resulting_file = posixpath.join(dest, 'push_file.txt')
156      with self.assertRaises(device_errors.AdbShellCommandFailedError):
157        under_test.Shell('ls %s' % resulting_file)
158      under_test.Push(src, dest)
159      self.assertEquals(
160          resulting_file,
161          under_test.Shell('ls %s' % resulting_file).strip())
162
163  def testPush_directoryToDirectory(self):
164    under_test = self.getTestInstance()
165    with self.getTestPushDestination(under_test) as push_target_directory:
166      src = os.path.join(_TEST_DATA_DIR, 'push_directory')
167      dest = posixpath.join(push_target_directory, 'push_directory')
168      with self.assertRaises(device_errors.AdbShellCommandFailedError):
169        under_test.Shell('ls %s' % dest)
170      under_test.Push(src, dest)
171      self.assertEquals(
172          sorted(os.listdir(src)),
173          sorted(under_test.Shell('ls %s' % dest).strip().split()))
174
175  def testPush_directoryToExistingDirectory(self):
176    under_test = self.getTestInstance()
177    with self.getTestPushDestination(under_test) as push_target_directory:
178      src = os.path.join(_TEST_DATA_DIR, 'push_directory')
179      dest = push_target_directory
180      resulting_directory = posixpath.join(dest, 'push_directory')
181      with self.assertRaises(device_errors.AdbShellCommandFailedError):
182        under_test.Shell('ls %s' % resulting_directory)
183      under_test.Shell('mkdir %s' % resulting_directory)
184      under_test.Push(src, dest)
185      self.assertEquals(
186          sorted(os.listdir(src)),
187          sorted(under_test.Shell('ls %s' % resulting_directory).split()))
188
189  # TODO(jbudorick): Implement tests for the following:
190  # taskset -c
191  # devices [-l]
192  # pull
193  # shell
194  # ls
195  # logcat [-c] [-d] [-v] [-b]
196  # forward [--remove] [--list]
197  # jdwp
198  # install [-l] [-r] [-s] [-d]
199  # install-multiple [-l] [-r] [-s] [-d] [-p]
200  # uninstall [-k]
201  # backup -f [-apk] [-shared] [-nosystem] [-all]
202  # restore
203  # wait-for-device
204  # get-state (BROKEN IN THE M SDK)
205  # get-devpath
206  # remount
207  # reboot
208  # reboot-bootloader
209  # root
210  # emu
211
212  @classmethod
213  def tearDownClass(cls):
214    print
215    print
216    print 'tested %s' % adb_wrapper.AdbWrapper.GetAdbPath()
217    print '  %s' % adb_wrapper.AdbWrapper.Version()
218    print 'connected devices:'
219    try:
220      for d in adb_wrapper.AdbWrapper.Devices():
221        print '  %s' % d
222    except device_errors.AdbCommandFailedError:
223      print '  <failed to list devices>'
224      raise
225    finally:
226      print
227
228
229if __name__ == '__main__':
230  sys.exit(unittest.main())
231