Add everything from src/tools/isolate r159537.

tests/trace_inputs_test.py

+#!/usr/bin/env python
+# Copyright (c) 2012 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 logging
+import os
+import unittest
+import sys
+
+BASE_DIR = unicode(os.path.dirname(os.path.abspath(__file__)))
+ROOT_DIR = os.path.dirname(BASE_DIR)
+sys.path.insert(0, ROOT_DIR)
+
+FILE_PATH = unicode(os.path.abspath(__file__))
+
+import trace_inputs
+
+
+def join_norm(*args):
+  """Joins and normalizes path in a single step."""
+  return unicode(os.path.normpath(os.path.join(*args)))
+
+
+class TraceInputs(unittest.TestCase):
+  def test_process_quoted_arguments(self):
+    test_cases = (
+      ('"foo"', ['foo']),
+      ('"foo", "bar"', ['foo', 'bar']),
+      ('"foo"..., "bar"', ['foo', 'bar']),
+      ('"foo", "bar"...', ['foo', 'bar']),
+      (
+        '"/browser_tests", "--type=use,comma"',
+        ['/browser_tests', '--type=use,comma']
+      ),
+      (
+        '"/browser_tests", "--ignored=\\" --type=renderer \\""',
+        ['/browser_tests', '--ignored=" --type=renderer "']
+      ),
+    )
+    for actual, expected in test_cases:
+      self.assertEquals(
+          expected, trace_inputs.strace_process_quoted_arguments(actual))
+
+  def test_process_escaped_arguments(self):
+    test_cases = (
+      ('foo\\0', ['foo']),
+      ('foo\\001bar\\0', ['foo', 'bar']),
+      ('\\"foo\\"\\0', ['"foo"']),
+    )
+    for actual, expected in test_cases:
+      self.assertEquals(
+          expected,
+          trace_inputs.Dtrace.Context.process_escaped_arguments(actual))
+
+  def test_variable_abs(self):
+    value = trace_inputs.Results.File(None, '/foo/bar', False, False)
+    actual = value.replace_variables({'$FOO': '/foo'})
+    self.assertEquals('$FOO/bar', actual.path)
+    self.assertEquals('$FOO/bar', actual.full_path)
+    self.assertEquals(True, actual.tainted)
+
+  def test_variable_rel(self):
+    value = trace_inputs.Results.File('/usr', 'foo/bar', False, False)
+    actual = value.replace_variables({'$FOO': 'foo'})
+    self.assertEquals('$FOO/bar', actual.path)
+    self.assertEquals(os.path.join('/usr', '$FOO/bar'), actual.full_path)
+    self.assertEquals(True, actual.tainted)
+
+  def test_native_case_end_with_os_path_sep(self):
+    # Make sure the trailing os.path.sep is kept.
+    path = trace_inputs.get_native_path_case(ROOT_DIR) + os.path.sep
+    self.assertEquals(trace_inputs.get_native_path_case(path), path)
+
+  def test_native_case_non_existing(self):
+    # Make sure it doesn't throw on non-existing files.
+    non_existing = 'trace_input_test_this_file_should_not_exist'
+    path = os.path.expanduser('~/' + non_existing)
+    self.assertFalse(os.path.exists(path))
+    path = trace_inputs.get_native_path_case(ROOT_DIR) + os.path.sep
+    self.assertEquals(trace_inputs.get_native_path_case(path), path)
+
+  if sys.platform in ('darwin', 'win32'):
+    def test_native_case_not_sensitive(self):
+      # The home directory is almost guaranteed to have mixed upper/lower case
+      # letters on both Windows and OSX.
+      # This test also ensures that the output is independent on the input
+      # string case.
+      path = os.path.expanduser('~')
+      self.assertTrue(os.path.isdir(path))
+      # This test assumes the variable is in the native path case on disk, this
+      # should be the case. Verify this assumption:
+      self.assertEquals(path, trace_inputs.get_native_path_case(path))
+      self.assertEquals(
+          trace_inputs.get_native_path_case(path.lower()),
+          trace_inputs.get_native_path_case(path.upper()))
+
+    def test_native_case_not_sensitive_non_existent(self):
+      # This test also ensures that the output is independent on the input
+      # string case.
+      non_existing = os.path.join(
+          'trace_input_test_this_dir_should_not_exist', 'really not', '')
+      path = os.path.expanduser(os.path.join('~', non_existing))
+      self.assertFalse(os.path.exists(path))
+      lower = trace_inputs.get_native_path_case(path.lower())
+      upper = trace_inputs.get_native_path_case(path.upper())
+      # Make sure non-existing element is not modified:
+      self.assertTrue(lower.endswith(non_existing.lower()))
+      self.assertTrue(upper.endswith(non_existing.upper()))
+      self.assertEquals(lower[:-len(non_existing)], upper[:-len(non_existing)])
+
+  if sys.platform != 'win32':
+    def test_symlink(self):
+      # This test will fail if the checkout is in a symlink.
+      actual = trace_inputs.split_at_symlink(None, ROOT_DIR)
+      expected = (ROOT_DIR, None, None)
+      self.assertEquals(expected, actual)
+
+      actual = trace_inputs.split_at_symlink(
+          None, os.path.join(BASE_DIR, 'trace_inputs'))
+      expected = (
+          os.path.join(BASE_DIR, 'trace_inputs'), None, None)
+      self.assertEquals(expected, actual)
+
+      actual = trace_inputs.split_at_symlink(
+          None, os.path.join(BASE_DIR, 'trace_inputs', 'files2'))
+      expected = (
+          os.path.join(BASE_DIR, 'trace_inputs'), 'files2', '')
+      self.assertEquals(expected, actual)
+
+      actual = trace_inputs.split_at_symlink(
+          ROOT_DIR, os.path.join('tests', 'trace_inputs', 'files2'))
+      expected = (
+          os.path.join('tests', 'trace_inputs'), 'files2', '')
+      self.assertEquals(expected, actual)
+      actual = trace_inputs.split_at_symlink(
+          ROOT_DIR, os.path.join('tests', 'trace_inputs', 'files2', 'bar'))
+      expected = (
+          os.path.join('tests', 'trace_inputs'), 'files2', '/bar')
+      self.assertEquals(expected, actual)
+
+    def test_native_case_symlink_right_case(self):
+      actual = trace_inputs.get_native_path_case(
+          os.path.join(BASE_DIR, 'trace_inputs'))
+      self.assertEquals('trace_inputs', os.path.basename(actual))
+
+      # Make sure the symlink is not resolved.
+      actual = trace_inputs.get_native_path_case(
+          os.path.join(BASE_DIR, 'trace_inputs', 'files2'))
+      self.assertEquals('files2', os.path.basename(actual))
+
+  if sys.platform == 'darwin':
+    def test_native_case_symlink_wrong_case(self):
+      actual = trace_inputs.get_native_path_case(
+          os.path.join(BASE_DIR, 'trace_inputs'))
+      self.assertEquals('trace_inputs', os.path.basename(actual))
+
+      # Make sure the symlink is not resolved.
+      actual = trace_inputs.get_native_path_case(
+          os.path.join(BASE_DIR, 'trace_inputs', 'Files2'))
+      self.assertEquals('files2', os.path.basename(actual))
+
+
+if sys.platform != 'win32':
+  class StraceInputs(unittest.TestCase):
+    # Represents the root process pid (an arbitrary number).
+    _ROOT_PID = 27
+    _CHILD_PID = 14
+    _GRAND_CHILD_PID = 70
+
+    @staticmethod
+    def _load_context(lines, initial_cwd):
+      context = trace_inputs.Strace.Context(lambda _: False, initial_cwd)
+      for line in lines:
+        context.on_line(*line)
+      return context.to_results().flatten()
+
+    def _test_lines(self, lines, initial_cwd, files, command=None):
+      filepath = join_norm(initial_cwd, '../out/unittests')
+      command = command or ['../out/unittests']
+      expected = {
+        'root': {
+          'children': [],
+          'command': command,
+          'executable': filepath,
+          'files': files,
+          'initial_cwd': initial_cwd,
+          'pid': self._ROOT_PID,
+        }
+      }
+      if not files:
+        expected['root']['command'] = None
+        expected['root']['executable'] = None
+      self.assertEquals(expected, self._load_context(lines, initial_cwd))
+
+    def test_execve(self):
+      lines = [
+        (self._ROOT_PID,
+          'execve("/home/foo_bar_user/out/unittests", '
+            '["/home/foo_bar_user/out/unittests", '
+            '"--gtest_filter=AtExitTest.Basic"], [/* 44 vars */])         = 0'),
+        (self._ROOT_PID,
+          'open("out/unittests.log", O_WRONLY|O_CREAT|O_APPEND, 0666) = 8'),
+      ]
+      files = [
+        {
+          'path': u'/home/foo_bar_user/out/unittests',
+          'size': -1,
+        },
+        {
+          'path': u'/home/foo_bar_user/src/out/unittests.log',
+          'size': -1,
+        },
+      ]
+      command = [
+        '/home/foo_bar_user/out/unittests', '--gtest_filter=AtExitTest.Basic',
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', files, command)
+
+    def test_empty(self):
+      try:
+        self._load_context([], None)
+        self.fail()
+      except trace_inputs.TracingFailure, e:
+        expected = (
+          'Found internal inconsitency in process lifetime detection '
+          'while finding the root process',
+          None,
+          None,
+          None,
+          [])
+        self.assertEquals(expected, e.args)
+
+    def test_chmod(self):
+      lines = [
+          (self._ROOT_PID, 'chmod("temp/file", 0100644) = 0'),
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', [])
+
+    def test_close(self):
+      lines = [
+        (self._ROOT_PID, 'close(7)                          = 0'),
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', [])
+
+    def test_clone(self):
+      # Grand-child with relative directory.
+      lines = [
+        (self._ROOT_PID,
+          'clone(child_stack=0, flags=CLONE_CHILD_CLEARTID'
+            '|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7f5350f829d0) = %d' %
+            self._CHILD_PID),
+        (self._CHILD_PID,
+          'clone(child_stack=0, flags=CLONE_CHILD_CLEARTID'
+            '|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7f5350f829d0) = %d' %
+            self._GRAND_CHILD_PID),
+        (self._GRAND_CHILD_PID,
+          'open("%s", O_RDONLY)       = 76' % os.path.basename(FILE_PATH)),
+      ]
+      size = os.stat(FILE_PATH).st_size
+      expected = {
+        'root': {
+          'children': [
+            {
+              'children': [
+                {
+                  'children': [],
+                  'command': None,
+                  'executable': None,
+                  'files': [
+                    {
+                      'path': FILE_PATH,
+                      'size': size,
+                    },
+                  ],
+                  'initial_cwd': BASE_DIR,
+                  'pid': self._GRAND_CHILD_PID,
+                },
+              ],
+              'command': None,
+              'executable': None,
+              'files': [],
+              'initial_cwd': BASE_DIR,
+              'pid': self._CHILD_PID,
+            },
+          ],
+          'command': None,
+          'executable': None,
+          'files': [],
+          'initial_cwd': BASE_DIR,
+          'pid': self._ROOT_PID,
+        },
+      }
+      self.assertEquals(expected, self._load_context(lines, BASE_DIR))
+
+    def test_clone_chdir(self):
+      # Grand-child with relative directory.
+      lines = [
+        (self._ROOT_PID,
+          'execve("../out/unittests", '
+            '["../out/unittests"...], [/* 44 vars */])         = 0'),
+        (self._ROOT_PID,
+          'clone(child_stack=0, flags=CLONE_CHILD_CLEARTID'
+            '|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7f5350f829d0) = %d' %
+            self._CHILD_PID),
+        (self._CHILD_PID,
+          'chdir("/home_foo_bar_user/path1") = 0'),
+        (self._CHILD_PID,
+          'clone(child_stack=0, flags=CLONE_CHILD_CLEARTID'
+            '|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7f5350f829d0) = %d' %
+            self._GRAND_CHILD_PID),
+        (self._GRAND_CHILD_PID,
+          'execve("../out/unittests", '
+            '["../out/unittests"...], [/* 44 vars */])         = 0'),
+        (self._ROOT_PID, 'chdir("/home_foo_bar_user/path2") = 0'),
+        (self._GRAND_CHILD_PID,
+          'open("random.txt", O_RDONLY)       = 76'),
+      ]
+      expected = {
+        'root': {
+          'children': [
+            {
+              'children': [
+                {
+                  'children': [],
+                  'command': ['../out/unittests'],
+                  'executable': '/home_foo_bar_user/out/unittests',
+                  'files': [
+                    {
+                      'path': u'/home_foo_bar_user/out/unittests',
+                      'size': -1,
+                    },
+                    {
+                      'path': u'/home_foo_bar_user/path1/random.txt',
+                      'size': -1,
+                    },
+                  ],
+                  'initial_cwd': u'/home_foo_bar_user/path1',
+                  'pid': self._GRAND_CHILD_PID,
+                },
+              ],
+              # clone does not carry over the command and executable so it is
+              # clear if an execve() call was done or not.
+              'command': None,
+              'executable': None,
+              # This is important, since no execve call was done, it didn't
+              # touch the executable file.
+              'files': [],
+              'initial_cwd': unicode(ROOT_DIR),
+              'pid': self._CHILD_PID,
+            },
+          ],
+          'command': ['../out/unittests'],
+          'executable': join_norm(ROOT_DIR, '../out/unittests'),
+          'files': [
+            {
+              'path': join_norm(ROOT_DIR, '../out/unittests'),
+              'size': -1,
+            },
+          ],
+          'initial_cwd': unicode(ROOT_DIR),
+          'pid': self._ROOT_PID,
+        },
+      }
+      self.assertEquals(expected, self._load_context(lines, ROOT_DIR))
+
+    def test_open(self):
+      lines = [
+        (self._ROOT_PID,
+          'execve("../out/unittests", '
+            '["../out/unittests"...], [/* 44 vars */])         = 0'),
+        (self._ROOT_PID,
+          'open("out/unittests.log", O_WRONLY|O_CREAT|O_APPEND, 0666) = 8'),
+      ]
+      files = [
+        {
+          'path': u'/home/foo_bar_user/out/unittests',
+          'size': -1,
+        },
+        {
+          'path': u'/home/foo_bar_user/src/out/unittests.log',
+          'size': -1,
+        },
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', files)
+
+    def test_open_resumed(self):
+      lines = [
+        (self._ROOT_PID,
+          'execve("../out/unittests", '
+            '["../out/unittests"...], [/* 44 vars */])         = 0'),
+        (self._ROOT_PID,
+          'open("out/unittests.log", O_WRONLY|O_CREAT|O_APPEND '
+            '<unfinished ...>'),
+        (self._ROOT_PID, '<... open resumed> )              = 3'),
+      ]
+      files = [
+        {
+          'path': u'/home/foo_bar_user/out/unittests',
+          'size': -1,
+        },
+        {
+          'path': u'/home/foo_bar_user/src/out/unittests.log',
+          'size': -1,
+        },
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', files)
+
+    def test_rmdir(self):
+      lines = [
+          (self._ROOT_PID, 'rmdir("directory/to/delete") = 0'),
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', [])
+
+    def test_setxattr(self):
+      lines = [
+          (self._ROOT_PID,
+           'setxattr("file.exe", "attribute", "value", 0, 0) = 0'),
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', [])
+
+    def test_sig_unexpected(self):
+      lines = [
+        (self._ROOT_PID, 'exit_group(0)                     = ?'),
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', [])
+
+    def test_stray(self):
+      lines = [
+        (self._ROOT_PID,
+          'execve("../out/unittests", '
+            '["../out/unittests"...], [/* 44 vars */])         = 0'),
+        (self._ROOT_PID,
+          ')                                       = ? <unavailable>'),
+      ]
+      files = [
+        {
+          'path': u'/home/foo_bar_user/out/unittests',
+          'size': -1,
+        },
+      ]
+      self._test_lines(lines, '/home/foo_bar_user/src', files)
+
+
+if __name__ == '__main__':
+  VERBOSE = '-v' in sys.argv
+  logging.basicConfig(level=logging.DEBUG if VERBOSE else logging.ERROR)
+  unittest.main()