codekingpro/portable-devtools

1# Protocol Buffers - Google's data interchange format2# Copyright 2008 Google Inc.  All rights reserved.3#4# Use of this source code is governed by a BSD-style5# license that can be found in the LICENSE file or at6# https://developers.google.com/open-source/licenses/bsd7 8"""A subclass of unittest.TestCase which checks for reference leaks.9 10To use:11- Use testing_refleak.BaseTestCase instead of unittest.TestCase12- Configure and compile Python with --with-pydebug13 14If sys.gettotalrefcount() is not available (because Python was built without15the Py_DEBUG option), then this module is a no-op and tests will run normally.16"""17 18import copyreg19import gc20import sys21import unittest22 23 24class LocalTestResult(unittest.TestResult):25  """A TestResult which forwards events to a parent object, except for Skips."""26 27  def __init__(self, parent_result):28    unittest.TestResult.__init__(self)29    self.parent_result = parent_result30 31  def addError(self, test, error):32    self.parent_result.addError(test, error)33 34  def addFailure(self, test, error):35    self.parent_result.addFailure(test, error)36 37  def addSkip(self, test, reason):38    pass39 40  def addDuration(self, test, duration):41    pass42 43 44class ReferenceLeakCheckerMixin(object):45  """A mixin class for TestCase, which checks reference counts."""46 47  NB_RUNS = 348 49  def run(self, result=None):50    testMethod = getattr(self, self._testMethodName)51    expecting_failure_method = getattr(testMethod, "__unittest_expecting_failure__", False)52    expecting_failure_class = getattr(self, "__unittest_expecting_failure__", False)53    if expecting_failure_class or expecting_failure_method:54      return55 56    # python_message.py registers all Message classes to some pickle global57    # registry, which makes the classes immortal.58    # We save a copy of this registry, and reset it before we could references.59    self._saved_pickle_registry = copyreg.dispatch_table.copy()60 61    # Run the test twice, to warm up the instance attributes.62    super(ReferenceLeakCheckerMixin, self).run(result=result)63    super(ReferenceLeakCheckerMixin, self).run(result=result)64 65    local_result = LocalTestResult(result)66    num_flakes = 067    refcount_deltas = []68 69    # Observe the refcount, then create oldrefcount which actually makes the70    # refcount 1 higher than the recorded value immediately71    oldrefcount = self._getRefcounts()72    while len(refcount_deltas) < self.NB_RUNS:73      oldrefcount = self._getRefcounts()74      super(ReferenceLeakCheckerMixin, self).run(result=local_result)75      newrefcount = self._getRefcounts()76      # If the GC was able to collect some objects after the call to run() that77      # it could not collect before the call, then the counts won't match.78      if newrefcount < oldrefcount and num_flakes < 2:79        # This result is (probably) a flake -- garbage collectors aren't very80        # predictable, but a lower ending refcount is the opposite of the81        # failure we are testing for. If the result is repeatable, then we will82        # eventually report it, but not after trying to eliminate it.83        num_flakes += 184        continue85      num_flakes = 086      refcount_deltas.append(newrefcount - oldrefcount)87    print(refcount_deltas, self)88 89    try:90      self.assertEqual(refcount_deltas, [0] * self.NB_RUNS)91    except Exception:  # pylint: disable=broad-except92      result.addError(self, sys.exc_info())93 94  def _getRefcounts(self):95    if hasattr(sys, "_clear_internal_caches"):  # Since 3.1396      sys._clear_internal_caches()  # pylint: disable=protected-access97    else:98      sys._clear_type_cache()  # pylint: disable=protected-access99    copyreg.dispatch_table.clear()100    copyreg.dispatch_table.update(self._saved_pickle_registry)101    # It is sometimes necessary to gc.collect() multiple times, to ensure102    # that all objects can be collected.103    gc.collect()104    gc.collect()105    gc.collect()106    return sys.gettotalrefcount()107 108 109if hasattr(sys, 'gettotalrefcount'):110 111  def TestCase(test_class):112    new_bases = (ReferenceLeakCheckerMixin,) + test_class.__bases__113    new_class = type(test_class)(114        test_class.__name__, new_bases, dict(test_class.__dict__))115    return new_class116  SkipReferenceLeakChecker = unittest.skip117 118else:119  # When PyDEBUG is not enabled, run the tests normally.120 121  def TestCase(test_class):122    return test_class123 124  def SkipReferenceLeakChecker(reason):125    del reason  # Don't skip, so don't need a reason.126    def Same(func):127      return func128    return Same129