(self, block_func, block_args, trigger_func, trigger_args)
| 58 | class BlockingTestMixin: |
| 59 | |
| 60 | def do_blocking_test(self, block_func, block_args, trigger_func, trigger_args): |
| 61 | thread = _TriggerThread(trigger_func, trigger_args) |
| 62 | thread.start() |
| 63 | try: |
| 64 | self.result = block_func(*block_args) |
| 65 | # If block_func returned before our thread made the call, we failed! |
| 66 | if not thread.startedEvent.is_set(): |
| 67 | self.fail("blocking function %r appeared not to block" % |
| 68 | block_func) |
| 69 | return self.result |
| 70 | finally: |
| 71 | threading_helper.join_thread(thread) # make sure the thread terminates |
| 72 | |
| 73 | # Call this instead if block_func is supposed to raise an exception. |
| 74 | def do_exceptional_blocking_test(self,block_func, block_args, trigger_func, |
no test coverage detected