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
5import threading
6import unittest
7
8import stress
9
10
11class StopThreadForTesting(Exception):
12    pass
13
14
15class 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
123if __name__ == '__main__':
124    unittest.main()
125