Check for stray threads from tests and fail if found

testing_support/thread_watcher.py

+# Copyright (c) 2016 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 sys
+import threading
+import traceback
+import unittest
+
+
+class ThreadWatcherMixIn(object):
+  def setUp(self):
+    self._pre_test_threads = [t.ident for t in threading.enumerate()]
+
+  def tearDown(self):
+    post_test_threads = threading.enumerate()
+    new_threads = [t for t in post_test_threads
+                   if t.ident not in self._pre_test_threads]
+    if new_threads:
+      details = []
+      for th in new_threads:
+        details.append('\nThread %s stacktrace:\n' % th)
+        try:
+          details.extend(traceback.format_stack(sys._current_frames()[t.ident]))
+        except Exception:
+          if t.ident in sys._current_frames():
+            raise
+          # This can happen due to threads starting or stopping concurrently.
+          details.append('  Thread stopped while acquiring stacktrace.\n')
+      details = ''.join(details)
+
+      self.fail('Found %d running thread(s) after the test.\n%s' % (
+        len(new_threads), details))
+
+
+
+class TestCase(unittest.TestCase, ThreadWatcherMixIn):
+  """Base unittest class that fails on leaked threads after the test."""
+  def setUp(self):
+    super(TestCase, self).setUp()
+    ThreadWatcherMixIn.setUp(self)
+
+  def tearDown(self):
+    ThreadWatcherMixIn.tearDown(self)
+    super(TestCase, self).tearDown()