src/third_party/WebKit/Tools/Scripts/webkitpy/common/system/executive_unittest.py

   1 # Copyright (C) 2010 Google Inc. All rights reserved.
   2 # Copyright (C) 2009 Daniel Bates (dbates@intudata.com). All rights reserved.
   3 #
   4 # Redistribution and use in source and binary forms, with or without
   5 # modification, are permitted provided that the following conditions are
   6 # met:
   7 #
   8 #    * Redistributions of source code must retain the above copyright
   9 # notice, this list of conditions and the following disclaimer.
  10 #    * Redistributions in binary form must reproduce the above
  11 # copyright notice, this list of conditions and the following disclaimer
  12 # in the documentation and/or other materials provided with the
  13 # distribution.
  14 #    * Neither the name of Google Inc. nor the names of its
  15 # contributors may be used to endorse or promote products derived from
  16 # this software without specific prior written permission.
  17 #
  18 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  19 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  20 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  21 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  22 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  23 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  24 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  25 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  26 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  27 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  28 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29
  30 import os
  31 import errno
  32 import signal
  33 import subprocess
  34 import sys
  35 import time
  36 import unittest
  37
  38 # Since we execute this script directly as part of the unit tests, we need to ensure
  39 # that Tools/Scripts is in sys.path for the next imports to work correctly.
  40 script_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
  41 if script_dir not in sys.path:
  42     sys.path.append(script_dir)
  43
  44
  45 from webkitpy.common.system.executive import Executive, ScriptError
  46 from webkitpy.common.system.filesystem_mock import MockFileSystem
  47
  48
  49 class ScriptErrorTest(unittest.TestCase):
  50     def test_message_with_output(self):
  51         error = ScriptError('My custom message!', '', -1)
  52         self.assertEqual(error.message_with_output(), 'My custom message!')
  53         error = ScriptError('My custom message!', '', -1, 'My output.')
  54         self.assertEqual(error.message_with_output(), 'My custom message!\n\noutput: My output.')
  55         error = ScriptError('', 'my_command!', -1, 'My output.', '/Users/username/blah')
  56         self.assertEqual(error.message_with_output(), 'Failed to run "\'my_command!\'" exit_code: -1 cwd: /Users/username/blah\n\noutput: My output.')
  57         error = ScriptError('', 'my_command!', -1, 'ab' + '1' * 499)
  58         self.assertEqual(error.message_with_output(), 'Failed to run "\'my_command!\'" exit_code: -1\n\noutput: Last 500 characters of output:\nb' + '1' * 499)
  59
  60     def test_message_with_tuple(self):
  61         error = ScriptError('', ('my', 'command'), -1, 'My output.', '/Users/username/blah')
  62         self.assertEqual(error.message_with_output(), 'Failed to run "(\'my\', \'command\')" exit_code: -1 cwd: /Users/username/blah\n\noutput: My output.')
  63
  64 def never_ending_command():
  65     """Arguments for a command that will never end (useful for testing process
  66     killing). It should be a process that is unlikely to already be running
  67     because all instances will be killed."""
  68     if sys.platform == 'win32':
  69         return ['wmic']
  70     return ['yes']
  71
  72
  73 def command_line(cmd, *args):
  74     return [sys.executable, __file__, '--' + cmd] + list(args)
  75
  76
  77 class ExecutiveTest(unittest.TestCase):
  78     def assert_interpreter_for_content(self, intepreter, content):
  79         fs = MockFileSystem()
  80
  81         tempfile, temp_name = fs.open_binary_tempfile('')
  82         tempfile.write(content)
  83         tempfile.close()
  84         file_interpreter = Executive.interpreter_for_script(temp_name, fs)
  85
  86         self.assertEqual(file_interpreter, intepreter)
  87
  88     def test_interpreter_for_script(self):
  89         self.assert_interpreter_for_content(None, '')
  90         self.assert_interpreter_for_content(None, 'abcd\nefgh\nijklm')
  91         self.assert_interpreter_for_content(None, '##/usr/bin/perl')
  92         self.assert_interpreter_for_content('perl', '#!/usr/bin/env perl')
  93         self.assert_interpreter_for_content('perl', '#!/usr/bin/env perl\nfirst\nsecond')
  94         self.assert_interpreter_for_content('perl', '#!/usr/bin/perl')
  95         self.assert_interpreter_for_content('perl', '#!/usr/bin/perl -w')
  96         self.assert_interpreter_for_content(sys.executable, '#!/usr/bin/env python')
  97         self.assert_interpreter_for_content(sys.executable, '#!/usr/bin/env python\nfirst\nsecond')
  98         self.assert_interpreter_for_content(sys.executable, '#!/usr/bin/python')
  99         self.assert_interpreter_for_content('ruby', '#!/usr/bin/env ruby')
 100         self.assert_interpreter_for_content('ruby', '#!/usr/bin/env ruby\nfirst\nsecond')
 101         self.assert_interpreter_for_content('ruby', '#!/usr/bin/ruby')
 102
 103     def test_run_command_with_bad_command(self):
 104         def run_bad_command():
 105             Executive().run_command(["foo_bar_command_blah"], error_handler=Executive.ignore_error, return_exit_code=True)
 106         self.assertRaises(OSError, run_bad_command)
 107
 108     def test_run_command_args_type(self):
 109         executive = Executive()
 110         self.assertRaises(AssertionError, executive.run_command, "echo")
 111         self.assertRaises(AssertionError, executive.run_command, u"echo")
 112         executive.run_command(command_line('echo', 'foo'))
 113         executive.run_command(tuple(command_line('echo', 'foo')))
 114
 115     def test_auto_stringify_args(self):
 116         executive = Executive()
 117         executive.run_command(command_line('echo', 1))
 118         executive.popen(command_line('echo', 1), stdout=executive.PIPE).wait()
 119         self.assertEqual('echo 1', executive.command_for_printing(['echo', 1]))
 120
 121     def test_popen_args(self):
 122         executive = Executive()
 123         # Explicitly naming the 'args' argument should not thow an exception.
 124         executive.popen(args=command_line('echo', 1), stdout=executive.PIPE).wait()
 125
 126     def test_run_command_with_unicode(self):
 127         """Validate that it is safe to pass unicode() objects
 128         to Executive.run* methods, and they will return unicode()
 129         objects by default unless decode_output=False"""
 130         unicode_tor_input = u"WebKit \u2661 Tor Arne Vestb\u00F8!"
 131         if sys.platform == 'win32':
 132             encoding = 'mbcs'
 133         else:
 134             encoding = 'utf-8'
 135         encoded_tor = unicode_tor_input.encode(encoding)
 136         # On Windows, we expect the unicode->mbcs->unicode roundtrip to be
 137         # lossy. On other platforms, we expect a lossless roundtrip.
 138         if sys.platform == 'win32':
 139             unicode_tor_output = encoded_tor.decode(encoding)
 140         else:
 141             unicode_tor_output = unicode_tor_input
 142
 143         executive = Executive()
 144
 145         output = executive.run_command(command_line('cat'), input=unicode_tor_input)
 146         self.assertEqual(output, unicode_tor_output)
 147
 148         output = executive.run_command(command_line('echo', unicode_tor_input))
 149         self.assertEqual(output, unicode_tor_output)
 150
 151         output = executive.run_command(command_line('echo', unicode_tor_input), decode_output=False)
 152         self.assertEqual(output, encoded_tor)
 153
 154         # Make sure that str() input also works.
 155         output = executive.run_command(command_line('cat'), input=encoded_tor, decode_output=False)
 156         self.assertEqual(output, encoded_tor)
 157
 158     def test_kill_process(self):
 159         executive = Executive()
 160         process = subprocess.Popen(never_ending_command(), stdout=subprocess.PIPE)
 161         self.assertEqual(process.poll(), None)  # Process is running
 162         executive.kill_process(process.pid)
 163
 164         # Killing again should fail silently.
 165         executive.kill_process(process.pid)
 166
 167     def _assert_windows_image_name(self, name, expected_windows_name):
 168         executive = Executive()
 169         windows_name = executive._windows_image_name(name)
 170         self.assertEqual(windows_name, expected_windows_name)
 171
 172     def test_windows_image_name(self):
 173         self._assert_windows_image_name("foo", "foo.exe")
 174         self._assert_windows_image_name("foo.exe", "foo.exe")
 175         self._assert_windows_image_name("foo.com", "foo.com")
 176         # If the name looks like an extension, even if it isn't
 177         # supposed to, we have no choice but to return the original name.
 178         self._assert_windows_image_name("foo.baz", "foo.baz")
 179         self._assert_windows_image_name("foo.baz.exe", "foo.baz.exe")
 180
 181     def test_check_running_pid(self):
 182         executive = Executive()
 183         self.assertTrue(executive.check_running_pid(os.getpid()))
 184         # Maximum pid number on Linux is 32768 by default
 185         self.assertFalse(executive.check_running_pid(100000))
 186
 187     def test_running_pids(self):
 188         if sys.platform in ("win32", "cygwin"):
 189             return  # This function isn't implemented on Windows yet.
 190
 191         executive = Executive()
 192         pids = executive.running_pids()
 193         self.assertIn(os.getpid(), pids)
 194
 195     def test_run_in_parallel_assert_nonempty(self):
 196         self.assertRaises(AssertionError, Executive().run_in_parallel, [])
 197
 198
 199 def main(platform, stdin, stdout, cmd, args):
 200     if platform == 'win32' and hasattr(stdout, 'fileno'):
 201         import msvcrt
 202         msvcrt.setmode(stdout.fileno(), os.O_BINARY)
 203     if cmd == '--cat':
 204         stdout.write(stdin.read())
 205     elif cmd == '--echo':
 206         stdout.write(' '.join(args))
 207     return 0
 208
 209 if __name__ == '__main__' and len(sys.argv) > 1 and sys.argv[1] in ('--cat', '--echo'):
 210     sys.exit(main(sys.platform, sys.stdin, sys.stdout, sys.argv[1], sys.argv[2:]))