Add NeedsSu for RunShellCommand to check whether it needs SU

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 295 matching lines @@
   def assertShellCallSequence(self, calls):
     '''Assert that we expect a sequence of calls to adb.Shell.
 
     Args:
       calls: a sequence of (cmd, return_value) pairs, where |cmd| is the
         expected shell command to run on the device (with any quoting already
         applied), and |return_value| is either a string to give as mock output
         or a _ShellError object to raise an AdbShellCommandFailedError.
     '''
     def mk_expected_call(cmd, return_value):
-      expected_args = Args(cmd, expect_rc=0, timeout=10, retries=0)
+      expected_args = Args(cmd, expect_rc=0)
       if isinstance(return_value, _ShellError):
         return_value = device_errors.AdbShellCommandFailedError(cmd,
             return_value.return_code, return_value.output, str(self.device))
       elif isinstance(return_value, _CmdTimeout):
         return_value = device_errors.CommandTimeoutError(return_value.msg,
                                                          str(self.device))
       return (expected_args, return_value)
 
     expected_calls = (mk_expected_call(a, r) for a, r in calls)
     return MockCallSequence(self, self.adb, 'Shell', expected_calls)
@@ skipping 356 matching lines @@
   def testNewRunShellImpl_withCwd(self):
     with self.assertShellCall('cd /some/test/path && ls'):
       self.device.RunShellCommand('ls', cwd='/some/test/path')
 
   def testNewRunShellImpl_withCwdQuoted(self):
     with self.assertShellCall("cd '/some test/path with/spaces' && ls"):
       self.device.RunShellCommand('ls', cwd='/some test/path with/spaces')
 
   def testRunShellCommand_withSu(self):
     with self.assertShellCallSequence([
-        ('ls /root', _ShellError()),
+        ('su -c ls /root && ! ls /root', ''),
         ('su -c setprop service.adb.root 0', '')]):
       self.device.RunShellCommand('setprop service.adb.root 0', as_root=True)
 
   def testRunShellCommand_withRoot(self):
     with self.assertShellCallSequence([
-        ('ls /root', '\r\n'),
+        ('su -c ls /root && ! ls /root', _ShellError()),
         ('setprop service.adb.root 0', '')]):
       self.device.RunShellCommand('setprop service.adb.root 0', as_root=True)
 
   def testRunShellCommand_manyLines(self):
     cmd = 'ls /some/path'
     with self.assertShellCall(cmd, 'file1\r\nfile2\r\nfile3\r\n'):
       self.assertEquals(['file1', 'file2', 'file3'],
                         self.device.RunShellCommand(cmd))
 
   def testRunShellCommand_singleLine_success(self):
@@ skipping 77 matching lines @@
                'this.is.a.test.process\r\n'),
         ('ps', 'USER   PID   PPID  VSIZE  RSS   WCHAN    PC       NAME\r\n')]):
         self.assertEquals(1,
             self.device.KillAll('this.is.a.test.process', blocking=True))
 
   def testKillAll_root(self):
     with self.assertShellCallSequence([
         ('ps', 'USER   PID   PPID  VSIZE  RSS   WCHAN    PC       NAME\r\n'
                'u0_a1  1234  174   123456 54321 ffffffff 456789ab '
                'this.is.a.test.process\r\n'),
-        ('ls /root', _ShellError()),
+        ('su -c ls /root && ! ls /root', ''),
         ('su -c kill -9 1234', '')]):
       self.assertEquals(1,
           self.device.KillAll('this.is.a.test.process', as_root=True))
 
   def testKillAll_sigterm(self):
     with self.assertShellCallSequence([
         ('ps', 'USER   PID   PPID  VSIZE  RSS   WCHAN    PC       NAME\r\n'
                'u0_a1  1234  174   123456 54321 ffffffff 456789ab '
                'this.is.a.test.process\r\n'),
         ('kill -15 1234', '')]):
@@ skipping 280 matching lines @@
     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(
         return_value='/test/device/external_dir')
     self.device.IsOnline = mock.Mock(return_value=True)
-    self.device._RunShellCommandImpl = mock.Mock()
+    self.device.RunShellCommand = 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(
+    self.assertEqual(2, self.device.RunShellCommand.call_count)
+    self.device.RunShellCommand.assert_any_call(
         ['unzip', '/test/device/external_dir/tmp.zip'],
         as_root=True,
         env={'PATH': '$PATH:/data/local/tmp/bin'},
         check_return=True)
-    self.device._RunShellCommandImpl.assert_any_call(
+    self.device.RunShellCommand.assert_any_call(
         ['rm', '/test/device/external_dir/tmp.zip'], check_return=True)
 
   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(
         return_value='/test/device/external_dir')
     self.device.IsOnline = mock.Mock(return_value=True)
-    self.device._RunShellCommandImpl = mock.Mock()
+    self.device.RunShellCommand = 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(
+    self.assertEqual(2, self.device.RunShellCommand.call_count)
+    self.device.RunShellCommand.assert_any_call(
         ['unzip', '/test/device/external_dir/tmp.zip'],
         as_root=True,
         env={'PATH': '$PATH:/data/local/tmp/bin'},
         check_return=True)
-    self.device._RunShellCommandImpl.assert_any_call(
+    self.device.RunShellCommand.assert_any_call(
         ['rm', '/test/device/external_dir/tmp.zip'], check_return=True)
 
 
 class DeviceUtilsFileExistsTest(DeviceUtilsOldImplTest):
 
   def testFileExists_usingTest_fileExists(self):
     with self.assertCalls(
         "adb -s 0123456789abcdef shell "
             "'test -e \"/data/app/test.file.exists\"; echo $?'",
         '0\r\n'):
@@ skipping 221 matching lines @@
     with self.assertShellCall('echo some.string > /test/file/to.write'):
       self.device.WriteTextFile('/test/file/to.write', 'some.string')
 
   def testWriteTextFileTest_quoted(self):
     with self.assertShellCall(
         "echo 'some other string' > '/test/file/to write'"):
       self.device.WriteTextFile('/test/file/to write', 'some other string')
 
   def testWriteTextFileTest_asRoot(self):
     with self.assertShellCallSequence([
-        ('ls /root', _ShellError()),
+        ('su -c ls /root && ! ls /root', ''),
         ('su -c echo string > /test/file', '')]):
       self.device.WriteTextFile('/test/file', 'string', as_root=True)
 
 class DeviceUtilsLsTest(DeviceUtilsOldImplTest):
 
   def testLs_nothing(self):
     with self.assertCallsSequence([
         ("adb -s 0123456789abcdef shell 'ls -lR /this/file/does.not.exist'",
          '/this/file/does.not.exist: No such file or directory\r\n'),
         ("adb -s 0123456789abcdef shell 'date +%z'", '+0000')]):
@@ skipping 284 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)