OLD | NEW |
---|---|
(Empty) | |
1 # Copyright (c) 2016 The Chromium Authors. All rights reserved. | |
2 # Use of this source code is governed by a BSD-style license that can be | |
3 # found in the LICENSE file. | |
4 | |
5 | |
6 import sys | |
7 import threading | |
8 import traceback | |
9 import unittest | |
10 | |
11 | |
12 class ThreadWatcherMixIn(object): | |
13 def setUp(self): | |
14 self._pre_test_threads = [t.ident for t in threading.enumerate()] | |
15 | |
16 def tearDown(self): | |
17 post_test_threads = threading.enumerate() | |
18 new_threads = [t for t in post_test_threads | |
19 if t.ident not in self._pre_test_threads] | |
20 if new_threads: | |
21 details = [] | |
22 for th in new_threads: | |
23 details.append('\nThread %s stacktrace:\n' % th) | |
24 try: | |
25 details.extend(traceback.format_stack(sys._current_frames()[t.ident])) | |
26 except Exception: | |
27 if t.ident in sys._current_frames(): | |
28 raise | |
29 # This can happen due to threads starting or stopping concurrently. | |
30 details.append(' Thread already stopped.\n') | |
tandrii(chromium)
2016/06/14 20:26:57
s/already stopped/stopped while acquiring stacktra
Sergiy Byelozyorov
2016/06/14 20:28:24
Done.
| |
31 details = ''.join(details) | |
32 | |
33 self.fail('Found %d running thread(s) after the test.\n%s' % ( | |
34 len(new_threads), details)) | |
35 | |
36 | |
37 | |
38 class TestCase(unittest.TestCase, ThreadWatcherMixIn): | |
39 """Base unittest class that fails on leaked threads after the test.""" | |
40 def setUp(self): | |
41 super(TestCase, self).setUp() | |
42 ThreadWatcherMixIn.setUp(self) | |
43 | |
44 def tearDown(self): | |
45 ThreadWatcherMixIn.tearDown(self) | |
46 super(TestCase, self).tearDown() | |
OLD | NEW |