| 39 | """ |
| 40 | |
| 41 | def setup_test_reactor(self, use_asyncio=False, auto_tear_down=True) -> None: |
| 42 | if use_asyncio: |
| 43 | warn_deprecated('4.2.0', 'use_asyncio=True is deprecated') |
| 44 | if not auto_tear_down: |
| 45 | warn_deprecated('4.2.0', 'auto_tear_down=False is deprecated') |
| 46 | |
| 47 | self.patch(threadpool, 'ThreadPool', NonThreadPool) |
| 48 | self.patch(twisted, 'ThreadPool', NonThreadPool) |
| 49 | self.reactor = TestReactor() |
| 50 | self.reactor.set_test_case(self) |
| 51 | |
| 52 | _setReactor(self.reactor) |
| 53 | |
| 54 | def deferToThread(f, *args, **kwargs): |
| 55 | return threads.deferToThreadPool( |
| 56 | self.reactor, self.reactor.getThreadPool(), f, *args, **kwargs |
| 57 | ) |
| 58 | |
| 59 | self.patch(threads, 'deferToThread', deferToThread) |
| 60 | |
| 61 | if auto_tear_down: |
| 62 | self.addCleanup(self.tear_down_test_reactor) |
| 63 | self._reactor_tear_down_called = False |
| 64 | |
| 65 | def tear_down_test_reactor(self) -> None: |
| 66 | if self._reactor_tear_down_called: |