Home | History | Annotate | Download | only in cros 
      1 # Copyright (c) 2012 The Chromium OS 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 import threading
      6 import unittest
      7 
      8 import stress
      9 
     10 
     11 class StopThreadForTesting(Exception):
     12     pass
     13 
     14 
     15 class StressorTest(unittest.TestCase):
     16 
     17     def testEscalateExceptions(self):
     18         def stress_event():
     19             raise StopThreadForTesting
     20 
     21         stressor = stress.CountedStressor(stress_event)
     22         stressor.start(1)
     23         self.assertRaises(StopThreadForTesting, stressor.wait)
     24 
     25 
     26     def testDontEscalateExceptions(self):
     27         event = threading.Event()
     28         def stress_event():
     29             event.set()
     30             raise StopThreadForTesting
     31 
     32         stressor = stress.CountedStressor(stress_event,
     33                                           escalate_exceptions=False)
     34         stressor.start(1)
     35         stressor.wait()
     36         self.assertTrue(event.is_set(), 'The stress event did not run')
     37 
     38 
     39     def testOnExit(self):
     40         def stress_event():
     41             pass
     42 
     43         event = threading.Event()
     44         def on_exit():
     45             event.set()
     46 
     47         stressor = stress.CountedStressor(stress_event, on_exit=on_exit)
     48         stressor.start(1)
     49         stressor.wait()
     50         self.assertTrue(event.is_set())
     51 
     52 
     53     def testOnExitWithException(self):
     54         def stress_event():
     55             raise StopThreadForTesting
     56 
     57         event = threading.Event()
     58         def on_exit():
     59             event.set()
     60 
     61         stressor = stress.CountedStressor(stress_event, on_exit=on_exit)
     62         stressor.start(1)
     63         self.assertRaises(StopThreadForTesting, stressor.wait)
     64         self.assertTrue(event.is_set())
     65 
     66 
     67     def testCountedStressorStartCondition(self):
     68         event = threading.Event()
     69 
     70         def start_condition():
     71             if event.is_set():
     72                 return True
     73             event.set()
     74             return False
     75 
     76         def stress_event():
     77             raise StopThreadForTesting
     78 
     79         stressor = stress.CountedStressor(stress_event)
     80         stressor.start(1, start_condition=start_condition)
     81         self.assertRaises(StopThreadForTesting, stressor.wait)
     82         self.assertTrue(event.is_set(),
     83                         'Stress event ran despite a False start condition')
     84 
     85 
     86     def testControlledStressorStartCondition(self):
     87         start_event = threading.Event()
     88         ran_event = threading.Event()
     89 
     90         def start_condition():
     91             if start_event.is_set():
     92                 return True
     93             start_event.set()
     94             return False
     95 
     96         def stress_event():
     97             ran_event.set()
     98             raise StopThreadForTesting
     99 
    100         stressor = stress.ControlledStressor(stress_event)
    101         stressor.start(start_condition=start_condition)
    102         ran_event.wait()
    103         self.assertRaises(StopThreadForTesting, stressor.stop)
    104         self.assertTrue(start_event.is_set(),
    105                         'Stress event ran despite a False start condition')
    106 
    107 
    108     def testCountedStressorIterations(self):
    109         # This is a list to get around scoping rules in Python 2.x. See
    110         # 'nonlocal' for the Python 3 remedy.
    111         count = [0]
    112 
    113         def stress_event():
    114             count[0] += 1
    115 
    116         stressor = stress.CountedStressor(stress_event)
    117         stressor.start(10)
    118         stressor.wait()
    119         self.assertEqual(10, count[0], 'Stress event did not run the expected '
    120                                        'number of iterations')
    121 
    122 
    123 if __name__ == '__main__':
    124     unittest.main()
    125