build/android/pylib/utils/reraiser_thread_unittest.py

# Copyright 2013 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.

"""Unittests for reraiser_thread.py."""

import threading
import unittest

from pylib.utils import reraiser_thread
from pylib.utils import watchdog_timer


class TestException(Exception):
  pass


class TestReraiserThread(unittest.TestCase):
  """Tests for reraiser_thread.ReraiserThread."""
  def testNominal(self):
    result = [None, None]

    def f(a, b=None):
      result[0] = a
      result[1] = b

    thread = reraiser_thread.ReraiserThread(f, [1], {'b': 2})
    thread.start()
    thread.join()
    self.assertEqual(result[0], 1)
    self.assertEqual(result[1], 2)

  def testRaise(self):
    def f():
      raise TestException

    thread = reraiser_thread.ReraiserThread(f)
    thread.start()
    thread.join()
    with self.assertRaises(TestException):
      thread.ReraiseIfException()


class TestReraiserThreadGroup(unittest.TestCase):
  """Tests for reraiser_thread.ReraiserThreadGroup."""
  def testInit(self):
    ran = [False] * 5
    def f(i):
      ran[i] = True

    group = reraiser_thread.ReraiserThreadGroup(
      [reraiser_thread.ReraiserThread(f, args=[i]) for i in range(5)])
    group.StartAll()
    group.JoinAll()
    for v in ran:
      self.assertTrue(v)

  def testAdd(self):
    ran = [False] * 5
    def f(i):
      ran[i] = True

    group = reraiser_thread.ReraiserThreadGroup()
    for i in xrange(5):
      group.Add(reraiser_thread.ReraiserThread(f, args=[i]))
    group.StartAll()
    group.JoinAll()
    for v in ran:
      self.assertTrue(v)

  def testJoinRaise(self):
    def f():
      raise TestException
    group = reraiser_thread.ReraiserThreadGroup(
      [reraiser_thread.ReraiserThread(f) for _ in xrange(5)])
    group.StartAll()
    with self.assertRaises(TestException):
      group.JoinAll()

  def testJoinTimeout(self):
    def f():
      pass
    event = threading.Event()
    def g():
      event.wait()
    group = reraiser_thread.ReraiserThreadGroup(
        [reraiser_thread.ReraiserThread(g),
         reraiser_thread.ReraiserThread(f)])
    group.StartAll()
    with self.assertRaises(reraiser_thread.TimeoutError):
      group.JoinAll(watchdog_timer.WatchdogTimer(0.01))
    event.set()


if __name__ == '__main__':
  unittest.main()