Home | History | Annotate | Download | only in sdk 
      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())
    231