New run shell implementation for DeviceUtils

build/android/pylib/device/device_utils_test.py

 #!/usr/bin/env python
 # Copyright 2014 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.
 
 """
 Unit tests for the contents of device_utils.py (mostly DeviceUtils).
 """
 
 # pylint: disable=C0321
@@ skipping 270 matching lines @@
   def testEnableRoot_rootFails(self):
     with self.assertCallsSequence([
         ('adb -s 0123456789abcdef shell getprop ro.build.type',
          'userdebug\r\n'),
         ('adb -s 0123456789abcdef root', 'no\r\n'),
         ('adb -s 0123456789abcdef wait-for-device', '')]):
       with self.assertRaises(device_errors.CommandFailedError):
         self.device.EnableRoot()
 
 
-class DeviceUtilsGetExternalStoragePathTest(DeviceUtilsOldImplTest):
+class DeviceUtilsGetExternalStoragePathTest(DeviceUtilsNewImplTest):
 
   def testGetExternalStoragePath_succeeds(self):
     fakeStoragePath = '/fake/storage/path'
-    with self.assertCalls(
-        "adb -s 0123456789abcdef shell 'echo $EXTERNAL_STORAGE'",
-        '%s\r\n' % fakeStoragePath):
+    with mock.patch.object(self.device, '_NewRunShellImpl',
+                           return_value='%s\r\n' % fakeStoragePath):
       self.assertEquals(fakeStoragePath,
                         self.device.GetExternalStoragePath())
 
   def testGetExternalStoragePath_fails(self):
-    with self.assertCalls(
-        "adb -s 0123456789abcdef shell 'echo $EXTERNAL_STORAGE'", '\r\n'):
+    with mock.patch.object(self.device, '_NewRunShellImpl',

[jbudorick, 2014/10/14 16:55:19]

I'm not sure I like mocking here vs mocking between DeviceUtils and AdbWrapper -
in this case, mocking AdbWrapper.Shell.

+                           return_value='\r\n'):
       with self.assertRaises(device_errors.CommandFailedError):
         self.device.GetExternalStoragePath()
 
 
 class DeviceUtilsWaitUntilFullyBootedTest(DeviceUtilsOldImplTest):
 
   def testWaitUntilFullyBooted_succeedsNoWifi(self):
     with self.assertCallsSequence([
         # AndroidCommands.WaitForSystemBootCompleted
         ('adb -s 0123456789abcdef wait-for-device', ''),
@@ skipping 249 matching lines @@
       self.device.RunShellCommand('echo $ANDROID_DATA', check_return=True)
 
   def testRunShellCommand_checkReturn_failure(self):
     with self.assertCalls(
         "adb -s 0123456789abcdef shell 'echo $ANDROID_DATA; echo %$?'",
         '\r\n%1\r\n'):
       with self.assertRaises(device_errors.CommandFailedError):
         self.device.RunShellCommand('echo $ANDROID_DATA', check_return=True)
 
 
+class DeviceUtilsNewRunShellImplTest(DeviceUtilsNewImplTest):
+  def assert_called_shell_with(self, cmd):
+    self.adb.Shell.assert_called_once_with(cmd, expect_rc=0,
+      timeout=device_utils._DEFAULT_TIMEOUT,
+      retries=device_utils._DEFAULT_RETRIES)
+
+  def testNewRunShellImpl_simple(self):
+    self.device._NewRunShellImpl('echo hello')
+    self.assert_called_shell_with('echo hello')
+
+  def testNewRunShellImpl_withlist(self):
+    self.device._NewRunShellImpl(['echo', 'hello'])
+    self.assert_called_shell_with('echo hello')
+
+  def testNewRunShellImpl_withfancylist(self):
+    self.device._NewRunShellImpl(['echo', 'hello world'])
+    self.assert_called_shell_with("echo 'hello world'")
+
+  def testNewRunShellImpl_withenv(self):
+    self.device._NewRunShellImpl('echo "$VAR"', env={'VAR': 'some_string'})
+    self.assert_called_shell_with('VAR=some_string echo "$VAR"')
+
+  def testNewRunShellImpl_withfancyenv(self):
+    self.device._NewRunShellImpl('echo "$VAR"', env={'VAR': 'hello world'})
+    self.assert_called_shell_with('''VAR='hello world' echo "$VAR"''')
+
+  def testNewRunShellImpl_withenv_invalid(self):
+    with self.assertRaises(KeyError):
+      self.device._NewRunShellImpl('some_cmd', env={'INVALID NAME': 'value'})
+
+  def testNewRunShellImpl_withcwd(self):
+    self.device._NewRunShellImpl('ls', cwd='/some/test/path')
+    self.assert_called_shell_with('cd /some/test/path && ls')
+
+  def testNewRunShellImpl_withfancycwd(self):
+    self.device._NewRunShellImpl('ls', cwd='/some test/path with/spaces')
+    self.assert_called_shell_with("cd '/some test/path with/spaces' && ls")
+
+  def testNewRunShellImpl_asroot_withoutroot(self):
+    with mock.patch.object(self.device, '_HasRootImpl', return_value=False):
+      self.device._NewRunShellImpl('ls /path/to/protected', as_root=True)
+      self.assert_called_shell_with('su -c ls /path/to/protected')
+
+  def testNewRunShellImpl_asroot_withroot(self):
+    with mock.patch.object(self.device, '_HasRootImpl', return_value=True):
+      self.device._NewRunShellImpl('ls /path/to/protected', as_root=True)
+      self.assert_called_shell_with('ls /path/to/protected')
+
+
 class DeviceUtilsKillAllTest(DeviceUtilsOldImplTest):
 
   def testKillAll_noMatchingProcesses(self):
     with self.assertCalls(
         "adb -s 0123456789abcdef shell 'ps'",
         'USER   PID   PPID  VSIZE  RSS   WCHAN    PC       NAME\r\n'):
       with self.assertRaises(device_errors.CommandFailedError):
         self.device.KillAll('test_process')
 
   def testKillAll_nonblocking(self):
@@ skipping 321 matching lines @@
     self.device._InstallCommands = mock.Mock()
 
   def testPushChangedFilesZipped_empty(self):
     test_files = []
     self.device._PushChangedFilesZipped(test_files)
     self.assertEqual(0, self.adb.Push.call_count)
 
   def testPushChangedFilesZipped_single(self):
     test_files = [('/test/host/path/file1', '/test/device/path/file1')]
 
-    self.device._GetExternalStoragePathImpl = mock.Mock(
+    self.device.GetExternalStoragePath = mock.Mock(
         return_value='/test/device/external_dir')
     self.device._IsOnlineImpl = mock.Mock(return_value=True)
     self.device._RunShellCommandImpl = mock.Mock()
     mock_zip_temp = mock.mock_open()
     mock_zip_temp.return_value.name = '/test/temp/file/tmp.zip'
     with mock.patch('multiprocessing.Process') as mock_zip_proc, (
          mock.patch('tempfile.NamedTemporaryFile', mock_zip_temp)):
       self.device._PushChangedFilesZipped(test_files)
 
     mock_zip_proc.assert_called_once_with(
         target=device_utils.DeviceUtils._CreateDeviceZip,
         args=('/test/temp/file/tmp.zip', test_files))
     self.adb.Push.assert_called_once_with(
         '/test/temp/file/tmp.zip', '/test/device/external_dir/tmp.zip')
     self.assertEqual(2, self.device._RunShellCommandImpl.call_count)
     self.device._RunShellCommandImpl.assert_any_call(
         ['unzip', '/test/device/external_dir/tmp.zip'],
         as_root=True, check_return=True,
         env={'PATH': '$PATH:/data/local/tmp/bin'})
     self.device._RunShellCommandImpl.assert_any_call(
         ['rm', '/test/device/external_dir/tmp.zip'])
 
   def testPushChangedFilesZipped_multiple(self):
     test_files = [('/test/host/path/file1', '/test/device/path/file1'),
                   ('/test/host/path/file2', '/test/device/path/file2')]
 
-    self.device._GetExternalStoragePathImpl = mock.Mock(
+    self.device.GetExternalStoragePath = mock.Mock(
         return_value='/test/device/external_dir')
     self.device._IsOnlineImpl = mock.Mock(return_value=True)
     self.device._RunShellCommandImpl = mock.Mock()
     mock_zip_temp = mock.mock_open()
     mock_zip_temp.return_value.name = '/test/temp/file/tmp.zip'
     with mock.patch('multiprocessing.Process') as mock_zip_proc, (
          mock.patch('tempfile.NamedTemporaryFile', mock_zip_temp)):
       self.device._PushChangedFilesZipped(test_files)
 
     mock_zip_proc.assert_called_once_with(
@@ skipping 559 matching lines @@
     self.device = device_utils.DeviceUtils(None)
     with self.assertCalls('adb  get-serialno', 'unknown'), (
          self.assertRaises(device_errors.NoDevicesError)):
       str(self.device)
 
 
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.DEBUG)
   unittest.main(verbosity=2)