Check for stray threads from tests and fail if found

testing_support/tests/thread_watcher_test.py

+# Copyright 2015 The Chromium Authors. All rights reserved.

[tandrii(chromium), 2016/06/13 17:03:28]

nit 2016

[Sergiy Byelozyorov, 2016/06/14 06:12:14]

Done.

+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+import mock
+import threading
+import time
+import unittest
+
+from testing_support import thread_watcher
+
+class TestNoExtraThreads(thread_watcher.TestCase):
+  def mock_test(self):
+    pass
+
+
+class TestExtraThreads(thread_watcher.TestCase):
+  def _start_thread(self, name='foo'):
+    stop = threading.Event()
+    def thread_func():
+      while not stop.is_set():
+        time.sleep(0.01)
+
+    t = threading.Thread(target=thread_func, name=name)
+    t.start()
+    return t, stop
+
+  def mock_test(self):
+    self.t1, self.stop1 = self._start_thread('foo')
+    self.t2, self.stop2 = self._start_thread('bar')
+
+  def stopThreads(self):
+    self.stop1.set()
+    self.stop2.set()
+    self.t1.join()
+    self.t2.join()
+
+
+class TestAutoStubTestCase(thread_watcher.TestCase):
+  @mock.patch('unittest.TestCase.fail')
+  def test_no_extra_threads(self, fail_mock):
+    TestNoExtraThreads('mock_test').run()
+    self.assertFalse(fail_mock.called)
+
+  @mock.patch('unittest.TestCase.fail')
+  def test_extra_threads(self, fail_mock):
+    t = TestExtraThreads('mock_test')
+    t.run()
+    t.stopThreads()
+    self.assertEqual(
+        fail_mock.call_args_list,
+        [mock.call('Found 2 running thread(s) after the test: foo, bar')])
+
+
+if __name__ == '__main__':
+  unittest.main()