1 # Copyright 2013 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.
5 """Unittests for reraiser_thread.py."""
10 from pylib
.utils
import reraiser_thread
11 from pylib
.utils
import watchdog_timer
14 class TestException(Exception):
18 class TestReraiserThread(unittest
.TestCase
):
19 """Tests for reraiser_thread.ReraiserThread."""
20 def testNominal(self
):
27 thread
= reraiser_thread
.ReraiserThread(f
, [1], {'b': 2})
30 self
.assertEqual(result
[0], 1)
31 self
.assertEqual(result
[1], 2)
37 thread
= reraiser_thread
.ReraiserThread(f
)
40 with self
.assertRaises(TestException
):
41 thread
.ReraiseIfException()
44 class TestReraiserThreadGroup(unittest
.TestCase
):
45 """Tests for reraiser_thread.ReraiserThreadGroup."""
51 group
= reraiser_thread
.ReraiserThreadGroup(
52 [reraiser_thread
.ReraiserThread(f
, args
=[i
]) for i
in range(5)])
63 group
= reraiser_thread
.ReraiserThreadGroup()
65 group
.Add(reraiser_thread
.ReraiserThread(f
, args
=[i
]))
71 def testJoinRaise(self
):
74 group
= reraiser_thread
.ReraiserThreadGroup(
75 [reraiser_thread
.ReraiserThread(f
) for _
in xrange(5)])
77 with self
.assertRaises(TestException
):
80 def testJoinTimeout(self
):
83 event
= threading
.Event()
86 group
= reraiser_thread
.ReraiserThreadGroup(
87 [reraiser_thread
.ReraiserThread(g
),
88 reraiser_thread
.ReraiserThread(f
)])
90 with self
.assertRaises(reraiser_thread
.TimeoutError
):
91 group
.JoinAll(watchdog_timer
.WatchdogTimer(0.01))
95 if __name__
== '__main__':