Move auto_stub.py from commit-queue

Use it in subprocess2 and trial_dir to remove manually bookeeping.

R=dpranke@chromium.org
BUG=
TEST=


Review URL: http://codereview.chromium.org/8688011

git-svn-id: svn://svn.chromium.org/chrome/trunk/tools/depot_tools@111941 0039d316-1c4b-4281-b951-d872f2087c98

.gitignore

@@ -7,6 +7,7 @@
 /svn_bin
 /svnversion.bat
 /depot_tools_testing_lib/_rietveld
+/testing_support/_rietveld
 /tests/subversion_config/README.txt
 /tests/subversion_config/auth
 /tests/subversion_config/servers

testing_support/auto_stub.py

+# Copyright (c) 2011 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 inspect
+import unittest
+
+
+class OrderedDict(object):
+  """Incomplete and inefficient implementation of collections.OrderedDict."""
+  def __init__(self):
+    self._keys = []
+
+  def setdefault(self, key, value):
+    try:
+      self._getindex(key)
+    except KeyError:
+      self._keys.append((key, value))
+    return self[key]
+
+  def _getindex(self, key):
+    for i, v in enumerate(self._keys):
+      if v[0] == key:
+        return i
+    raise KeyError(key)
+
+  def __getitem__(self, key):
+    return self._keys[self._getindex(key)][1]
+
+  def iteritems(self):
+    for i in self._keys:
+      yield i
+
+
+class AutoStubMixIn(object):
+  """Automatically restores stubbed functions on unit test teardDown.
+
+  It's an extremely lightweight mocking class that doesn't require bookeeping.
+  """
+  _saved = None
+
+  def mock(self, obj, member, mock):
+    self._saved = self._saved or OrderedDict()
+    old_value = self._saved.setdefault(
+        obj, OrderedDict()).setdefault(member, getattr(obj, member))
+    setattr(obj, member, mock)
+    return old_value
+
+  def tearDown(self):
+    """Restore all the mocked members."""
+    if self._saved:
+      for obj, items in self._saved.iteritems():
+        for member, previous_value in items.iteritems():
+          setattr(obj, member, previous_value)
+
+
+class SimpleMock(object):
+  """Really simple manual class mock."""
+  calls = []
+
+  def __init__(self, unit_test):
+    """Do not call __init__ if you want to use the global call list to detect
+    ordering across different instances.
+    """
+    self.calls = []
+    self.unit_test = unit_test
+    self.assertEquals = unit_test.assertEquals
+
+  def pop_calls(self):
+    """Returns the list of calls up to date.
+
+    Good to do self.assertEquals(expected, mock.pop_calls()).
+    """
+    calls = self.calls
+    self.calls = []
+    return calls
+
+  def check_calls(self, expected):
+    self.assertEquals(expected, self.pop_calls())
+
+  def _register_call(self, *args, **kwargs):
+    """Registers the name of the caller function."""
+    caller_name = kwargs.pop('caller_name', None) or inspect.stack()[1][3]
+    str_args = ', '.join(repr(arg) for arg in args)
+    str_kwargs = ', '.join('%s=%r' % (k, v) for k, v in kwargs.iteritems())
+    self.calls.append('%s(%s)' % (
+        caller_name, ', '.join(filter(None, [str_args, str_kwargs]))))
+
+
+class TestCase(unittest.TestCase, AutoStubMixIn):
+  """Adds python 2.7 functionality."""
+
+  def tearDown(self):
+    AutoStubMixIn.tearDown(self)
+    unittest.TestCase.tearDown(self)
+
+  def has_failed(self):
+    """Returns True if the test has failed."""
+    if hasattr(self, '_exc_info'):
+      # Only present in python <= 2.6
+      # pylint: disable=E1101
+      return bool(self._exc_info()[0])
+
+    # Only present in python >= 2.7
+    # pylint: disable=E1101
+    return not self._resultForDoCleanups.wasSuccessful()
+
+  def assertIs(self, expr1, expr2, msg=None):
+    if hasattr(super(TestCase, self), 'assertIs'):
+      return super(TestCase, self).assertIs(expr1, expr2, msg)
+    if expr1 is not expr2:
+      self.fail(msg or '%r is not %r' % (expr1, expr2))
+
+  def assertIsNot(self, expr1, expr2, msg=None):
+    if hasattr(super(TestCase, self), 'assertIsNot'):
+      return super(TestCase, self).assertIsNot(expr1, expr2, msg)
+    if expr1 is expr2:
+      self.fail(msg or 'unexpectedly identical: %r' % expr1)
+
+  def assertIn(self, expr1, expr2, msg=None):
+    if hasattr(super(TestCase, self), 'assertIn'):
+      return super(TestCase, self).assertIn(expr1, expr2, msg)
+    if expr1 not in expr2:
+      self.fail(msg or '%r not in %r' % (expr1, expr2))
+
+  def assertLess(self, a, b, msg=None):
+    if hasattr(super(TestCase, self), 'assertLess'):
+      return super(TestCase, self).assertLess(a, b, msg)
+    if not a < b:
+      self.fail(msg or '%r not less than %r' % (a, b))

testing_support/trial_dir.py

-#!/usr/bin/env python
 # Copyright (c) 2011 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.
@@ -9,7 +8,8 @@ import logging
 import os
 import sys
 import tempfile
-import unittest
+
+from testing_support import auto_stub
 
 import gclient_utils
 
@@ -79,15 +79,15 @@ class TrialDirMixIn(object):
     return self.trial.root_dir
 
 
-class TestCase(unittest.TestCase, TrialDirMixIn):
+class TestCase(auto_stub.TestCase, TrialDirMixIn):
   """Base unittest class that cleans off a trial directory in tearDown()."""
   def setUp(self):
-    unittest.TestCase.setUp(self)
+    auto_stub.TestCase.setUp(self)
     TrialDirMixIn.setUp(self)
 
   def tearDown(self):
     TrialDirMixIn.tearDown(self)
-    unittest.TestCase.tearDown(self)
+    auto_stub.TestCase.tearDown(self)
 
 
 if '-l' in sys.argv:

tests/subprocess2_test.py

@@ -21,6 +21,8 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 import subprocess2
 
+from testing_support import auto_stub
+
 # Method could be a function
 # pylint: disable=R0201
 
@@ -42,29 +44,10 @@ def convert_win(string):
   return string
 
 
-class DefaultsTest(unittest.TestCase):
-  # Can be mocked in a test.
-  TO_SAVE = {
-    subprocess2: [
-      'Popen', 'communicate', 'call', 'check_call', 'capture', 'check_output'],
-    subprocess2.subprocess.Popen: ['__init__', 'communicate'],
-  }
-
-  def setUp(self):
-    self.saved = {}
-    for module, names in self.TO_SAVE.iteritems():
-      self.saved[module] = dict(
-          (name, getattr(module, name)) for name in names)
-    # TODO(maruel): Do a reopen() on sys.__stdout__ and sys.__stderr__ so they
-    # can be trapped in the child process for better coverage.
-
-  def tearDown(self):
-    for module, saved in self.saved.iteritems():
-      for name, value in saved.iteritems():
-        setattr(module, name, value)
-
-  @staticmethod
-  def _fake_communicate():
+class DefaultsTest(auto_stub.TestCase):
+  # TODO(maruel): Do a reopen() on sys.__stdout__ and sys.__stderr__ so they
+  # can be trapped in the child process for better coverage.
+  def _fake_communicate(self):
     """Mocks subprocess2.communicate()."""
     results = {}
     def fake_communicate(args, **kwargs):
@@ -72,11 +55,10 @@ class DefaultsTest(unittest.TestCase):
       results.update(kwargs)
       results['args'] = args
       return ('stdout', 'stderr'), 0
-    subprocess2.communicate = fake_communicate
+    self.mock(subprocess2, 'communicate', fake_communicate)
     return results
 
-  @staticmethod
-  def _fake_Popen():
+  def _fake_Popen(self):
     """Mocks the whole subprocess2.Popen class."""
     results = {}
     class fake_Popen(object):
@@ -88,11 +70,10 @@ class DefaultsTest(unittest.TestCase):
       @staticmethod
       def communicate():
         return None, None
-    subprocess2.Popen = fake_Popen
+    self.mock(subprocess2, 'Popen', fake_Popen)
     return results
 
-  @staticmethod
-  def _fake_subprocess_Popen():
+  def _fake_subprocess_Popen(self):
     """Mocks the base class subprocess.Popen only."""
     results = {}
     def __init__(self, args, **kwargs):
@@ -101,8 +82,8 @@ class DefaultsTest(unittest.TestCase):
       results['args'] = args
     def communicate():
       return None, None
-    subprocess2.subprocess.Popen.__init__ = __init__
-    subprocess2.subprocess.Popen.communicate = communicate
+    self.mock(subprocess2.subprocess.Popen, '__init__', __init__)
+    self.mock(subprocess2.subprocess.Popen, 'communicate', communicate)
     return results
 
   def test_check_call_defaults(self):