| 46 | @patch('celery.utils.timer2.sleep') |
| 47 | @patch('os._exit') # To ensure the test fails gracefully |
| 48 | def test_on_tick(self, _exit, sleep): |
| 49 | def next_entry_side_effect(): |
| 50 | # side effect simulating following scenario: |
| 51 | # 3.33, 3.33, 3.33, <shutdown event set> |
| 52 | for _ in range(3): |
| 53 | yield 3.33 |
| 54 | while True: |
| 55 | yield getattr(t, "_Timer__is_shutdown").set() |
| 56 | |
| 57 | on_tick = Mock(name='on_tick') |
| 58 | t = timer2.Timer(on_tick=on_tick) |
| 59 | t._next_entry = Mock( |
| 60 | name='_next_entry', side_effect=next_entry_side_effect() |
| 61 | ) |
| 62 | t.run() |
| 63 | sleep.assert_called_with(3.33) |
| 64 | on_tick.assert_has_calls([call(3.33), call(3.33), call(3.33)]) |
| 65 | _exit.assert_not_called() |
| 66 | |
| 67 | @patch('os._exit') |
| 68 | def test_thread_crash(self, _exit): |