sdk/platform-tools/systrace/catapult/devil/devil/android/sdk/adb_compatibility_devicetest.py

   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
   6 import contextlib
   7 import os
   8 import posixpath
   9 import random
  10 import signal
  11 import sys
  12 import unittest
  13
  14 _CATAPULT_BASE_DIR = os.path.abspath(os.path.join(
  15     os.path.dirname(__file__), '..', '..', '..', '..'))
  16
  17 sys.path.append(os.path.join(_CATAPULT_BASE_DIR, 'devil'))
  18 from devil import devil_env
  19 from devil.android import device_errors
  20 from devil.android import device_test_case
  21 from devil.android.sdk import adb_wrapper
  22 from devil.utils import cmd_helper
  23 from 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
  30 def _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
  41 class 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
 229 if __name__ == '__main__':
 230   sys.exit(unittest.main())