Home | History | Annotate | Download | only in utils 
      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.
      4 
      5 """Unittests for reraiser_thread.py."""
      6 
      7 import threading
      8 import unittest
      9 
     10 from devil.utils import reraiser_thread
     11 from devil.utils import watchdog_timer
     12 
     13 
     14 class TestException(Exception):
     15   pass
     16 
     17 
     18 class TestReraiserThread(unittest.TestCase):
     19   """Tests for reraiser_thread.ReraiserThread."""
     20 
     21   def testNominal(self):
     22     result = [None, None]
     23 
     24     def f(a, b=None):
     25       result[0] = a
     26       result[1] = b
     27 
     28     thread = reraiser_thread.ReraiserThread(f, [1], {'b': 2})
     29     thread.start()
     30     thread.join()
     31     self.assertEqual(result[0], 1)
     32     self.assertEqual(result[1], 2)
     33 
     34   def testRaise(self):
     35     def f():
     36       raise TestException
     37 
     38     thread = reraiser_thread.ReraiserThread(f)
     39     thread.start()
     40     thread.join()
     41     with self.assertRaises(TestException):
     42       thread.ReraiseIfException()
     43 
     44 
     45 class TestReraiserThreadGroup(unittest.TestCase):
     46   """Tests for reraiser_thread.ReraiserThreadGroup."""
     47 
     48   def testInit(self):
     49     ran = [False] * 5
     50 
     51     def f(i):
     52       ran[i] = True
     53 
     54     group = reraiser_thread.ReraiserThreadGroup(
     55       [reraiser_thread.ReraiserThread(f, args=[i]) for i in range(5)])
     56     group.StartAll()
     57     group.JoinAll()
     58     for v in ran:
     59       self.assertTrue(v)
     60 
     61   def testAdd(self):
     62     ran = [False] * 5
     63 
     64     def f(i):
     65       ran[i] = True
     66 
     67     group = reraiser_thread.ReraiserThreadGroup()
     68     for i in xrange(5):
     69       group.Add(reraiser_thread.ReraiserThread(f, args=[i]))
     70     group.StartAll()
     71     group.JoinAll()
     72     for v in ran:
     73       self.assertTrue(v)
     74 
     75   def testJoinRaise(self):
     76     def f():
     77       raise TestException
     78     group = reraiser_thread.ReraiserThreadGroup(
     79       [reraiser_thread.ReraiserThread(f) for _ in xrange(5)])
     80     group.StartAll()
     81     with self.assertRaises(TestException):
     82       group.JoinAll()
     83 
     84   def testJoinTimeout(self):
     85     def f():
     86       pass
     87     event = threading.Event()
     88 
     89     def g():
     90       event.wait()
     91     group = reraiser_thread.ReraiserThreadGroup(
     92         [reraiser_thread.ReraiserThread(g),
     93          reraiser_thread.ReraiserThread(f)])
     94     group.StartAll()
     95     with self.assertRaises(reraiser_thread.TimeoutError):
     96       group.JoinAll(watchdog_timer.WatchdogTimer(0.01))
     97     event.set()
     98 
     99 
    100 class TestRunAsync(unittest.TestCase):
    101   """Tests for reraiser_thread.RunAsync."""
    102 
    103   def testNoArgs(self):
    104     results = reraiser_thread.RunAsync([])
    105     self.assertEqual([], results)
    106 
    107   def testOneArg(self):
    108     results = reraiser_thread.RunAsync([lambda: 1])
    109     self.assertEqual([1], results)
    110 
    111   def testTwoArgs(self):
    112     a, b = reraiser_thread.RunAsync((lambda: 1, lambda: 2))
    113     self.assertEqual(1, a)
    114     self.assertEqual(2, b)
    115 
    116 if __name__ == '__main__':
    117   unittest.main()
    118