text: #!/usr/bin/env python
  # Copyright 2016 The Chromium Authors. All rights reserved.
  # Use of this source code is governed by a BSD-style license that can be
  # found in the LICENSE file.
  
  import contextlib
  import os
  import posixpath
  import random
  import signal
  import sys
  import unittest
  
  _CATAPULT_BASE_DIR = os.path.abspath(
      os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
  
  sys.path.append(os.path.join(_CATAPULT_BASE_DIR, 'devil'))
  from devil import devil_env
  from devil.android import device_errors
  from devil.android import device_test_case
  from devil.android.sdk import adb_wrapper
  from devil.utils import cmd_helper
  from devil.utils import timeout_retry
  
  _TEST_DATA_DIR = os.path.abspath(
      os.path.join(os.path.dirname(__file__), 'test', 'data'))
  
  
  def _hostAdbPids():
    ps_status, ps_output = cmd_helper.GetCmdStatusAndOutput(
        ['pgrep', '-l', 'adb'])
    if ps_status != 0:
      return []
  
    pids_and_names = (line.split() for line in ps_output.splitlines())
    return [int(pid) for pid, name in pids_and_names if name == 'adb']
  
  
  class AdbCompatibilityTest(device_test_case.DeviceTestCase):
    @classmethod
    def setUpClass(cls):
      custom_adb_path = os.environ.get('ADB_PATH')
      custom_deps = {
          'config_type': 'BaseConfig',
          'dependencies': {},
      }
      if custom_adb_path:
        custom_deps['dependencies']['adb'] = {
            'file_info': {
                devil_env.GetPlatform(): {
                    'local_paths': [custom_adb_path],
                },
            },
        }
      devil_env.config.Initialize(configs=[custom_deps])
  
    def testStartServer(self):
      # Manually kill off any instances of adb.
      adb_pids = _hostAdbPids()
      for p in adb_pids:
        os.kill(p, signal.SIGKILL)
  
      self.assertIsNotNone(
          timeout_retry.WaitFor(
              lambda: not _hostAdbPids(), wait_period=0.1, max_tries=10))
  
      # start the adb server
      start_server_status, _ = cmd_helper.GetCmdStatusAndOutput(
          [adb_wrapper.AdbWrapper.GetAdbPath(), 'start-server'])
  
      # verify that the server is now online
      self.assertEquals(0, start_server_status)
      self.assertIsNotNone(
          timeout_retry.WaitFor(
              lambda: bool(_hostAdbPids()), wait_period=0.1, max_tries=10))
  
    def testKillServer(self):
      adb_pids = _hostAdbPids()
      if not adb_pids:
        adb_wrapper.AdbWrapper.StartServer()
  
      adb_pids = _hostAdbPids()
      self.assertGreaterEqual(len(adb_pids), 1)
  
      kill_server_status, _ = cmd_helper.GetCmdStatusAndOutput(
          [adb_wrapper.AdbWrapper.GetAdbPath(), 'kill-server'])
      self.assertEqual(0, kill_server_status)
  
      adb_pids = _hostAdbPids()
      self.assertEqual(0, len(adb_pids))
  
    def testDevices(self):
      devices = adb_wrapper.AdbWrapper.Devices()
      self.assertNotEqual(0, len(devices), 'No devices found.')
  
    def getTestInstance(self):
      """Creates a real AdbWrapper instance for testing."""
      return adb_wrapper.AdbWrapper(self.serial)
  
    def testShell(self):
      under_test = self.getTestInstance()
      shell_ls_result = under_test.Shell('ls')
      self.assertIsInstance(shell_ls_result, str)
      self.assertTrue(bool(shell_ls_result))
  
    def testShell_failed(self):
      under_test = self.getTestInstance()
      with self.assertRaises(device_errors.AdbShellCommandFailedError):
        under_test.Shell('ls /foo/bar/baz')
  
    def testShell_externalStorageDefined(self):
      under_test = self.getTestInstance()
      external_storage = under_test.Shell('echo $EXTERNAL_STORAGE')
      self.assertIsInstance(external_storage, str)
      self.assertTrue(posixpath.isabs(external_storage))
  
    @contextlib.contextmanager
    def getTestPushDestination(self, under_test):
      """Creates a temporary directory suitable for pushing to."""
      external_storage = under_test.Shell('echo $EXTERNAL_STORAGE').strip()
      if not external_storage:
        self.skipTest('External storage not available.')
      while True:
        random_hex = hex(random.randint(0, 2**52))[2:]
        name = 'tmp_push_test%s' % random_hex
        path = posixpath.join(external_storage, name)
        try:
          under_test.Shell('ls %s' % path)
        except device_errors.AdbShellCommandFailedError:
          break
      under_test.Shell('mkdir %s' % path)
      try:
        yield path
      finally:
        under_test.Shell('rm -rf %s' % path)
  
    def testPush_fileToFile(self):
      under_test = self.getTestInstance()
      with self.getTestPushDestination(under_test) as push_target_directory:
        src = os.path.join(_TEST_DATA_DIR, 'push_file.txt')
        dest = posixpath.join(push_target_directory, 'push_file.txt')
        with self.assertRaises(device_errors.AdbShellCommandFailedError):
          under_test.Shell('ls %s' % dest)
        under_test.Push(src, dest)
        self.assertEquals(dest, under_test.Shell('ls %s' % dest).strip())
  
    def testPush_fileToDirectory(self):
      under_test = self.getTestInstance()
      with self.getTestPushDestination(under_test) as push_target_directory:
        src = os.path.join(_TEST_DATA_DIR, 'push_file.txt')
        dest = push_target_directory
        resulting_file = posixpath.join(dest, 'push_file.txt')
        with self.assertRaises(device_errors.AdbShellCommandFailedError):
          under_test.Shell('ls %s' % resulting_file)
        under_test.Push(src, dest)
        self.assertEquals(resulting_file,
                          under_test.Shell('ls %s' % resulting_file).strip())
  
    def testPush_directoryToDirectory(self):
      under_test = self.getTestInstance()
      with self.getTestPushDestination(under_test) as push_target_directory:
        src = os.path.join(_TEST_DATA_DIR, 'push_directory')
        dest = posixpath.join(push_target_directory, 'push_directory')
        with self.assertRaises(device_errors.AdbShellCommandFailedError):
          under_test.Shell('ls %s' % dest)
        under_test.Push(src, dest)
        self.assertEquals(
            sorted(os.listdir(src)),
            sorted(under_test.Shell('ls %s' % dest).strip().split()))
  
    def testPush_directoryToExistingDirectory(self):
      under_test = self.getTestInstance()
      with self.getTestPushDestination(under_test) as push_target_directory:
        src = os.path.join(_TEST_DATA_DIR, 'push_directory')
        dest = push_target_directory
        resulting_directory = posixpath.join(dest, 'push_directory')
        with self.assertRaises(device_errors.AdbShellCommandFailedError):
          under_test.Shell('ls %s' % resulting_directory)
        under_test.Shell('mkdir %s' % resulting_directory)
        under_test.Push(src, dest)
        self.assertEquals(
            sorted(os.listdir(src)),
            sorted(under_test.Shell('ls %s' % resulting_directory).split()))
  
    # TODO(jbudorick): Implement tests for the following:
    # taskset -c
    # devices [-l]
    # pull
    # shell
    # ls
    # logcat [-c] [-d] [-v] [-b]
    # forward [--remove] [--list]
    # jdwp
    # install [-l] [-r] [-s] [-d]
    # install-multiple [-l] [-r] [-s] [-d] [-p]
    # uninstall [-k]
    # backup -f [-apk] [-shared] [-nosystem] [-all]
    # restore
    # wait-for-device
    # get-state (BROKEN IN THE M SDK)
    # get-devpath
    # remount
    # reboot
    # reboot-bootloader
    # root
    # emu
  
    @classmethod
    def tearDownClass(cls):
      print
      print
      print 'tested %s' % adb_wrapper.AdbWrapper.GetAdbPath()
      print '  %s' % adb_wrapper.AdbWrapper.Version()
      print 'connected devices:'
      try:
        for d in adb_wrapper.AdbWrapper.Devices():
          print '  %s' % d
      except device_errors.AdbCommandFailedError:
        print '  <failed to list devices>'
        raise
      finally:
        print
  
  
  if __name__ == '__main__':
    sys.exit(unittest.main())
  