(self,block_func, block_args, trigger_func,
trigger_args, expected_exception_class)
| 74 | |
| 75 | # Call this instead if block_func is supposed to raise an exception. |
| 76 | def do_exceptional_blocking_test(self,block_func, block_args, trigger_func, |
| 77 | trigger_args, expected_exception_class): |
| 78 | thread = _TriggerThread(trigger_func, trigger_args) |
| 79 | thread.start() |
| 80 | try: |
| 81 | try: |
| 82 | block_func(*block_args) |
| 83 | except expected_exception_class: |
| 84 | raise |
| 85 | else: |
| 86 | self.fail("expected exception of kind %r" % |
| 87 | expected_exception_class) |
| 88 | finally: |
| 89 | threading_helper.join_thread(thread) # make sure the thread terminates |
| 90 | if not thread.startedEvent.is_set(): |
| 91 | self.fail("trigger thread ended but event never set") |
| 92 | |
| 93 | |
| 94 | class BaseQueueTestMixin(BlockingTestMixin): |
no test coverage detected