(self)
| 193 | self._test_generator = None # type: Optional[Union[Generator, Coroutine]] |
| 194 | |
| 195 | def setUp(self) -> None: |
| 196 | setup_with_context_manager(self, warnings.catch_warnings()) |
| 197 | warnings.filterwarnings( |
| 198 | "ignore", |
| 199 | message="There is no current event loop", |
| 200 | category=DeprecationWarning, |
| 201 | module=r"tornado\..*", |
| 202 | ) |
| 203 | super().setUp() |
| 204 | # NOTE: this code attempts to navigate deprecation warnings introduced |
| 205 | # in Python 3.10. The idea of an implicit current event loop is |
| 206 | # deprecated in that version, with the intention that tests like this |
| 207 | # explicitly create a new event loop and run on it. However, other |
| 208 | # packages such as pytest-asyncio (as of version 0.16.0) still rely on |
| 209 | # the implicit current event loop and we want to be compatible with them |
| 210 | # (even when run on 3.10, but not, of course, on the future version of |
| 211 | # python that removes the get/set_event_loop methods completely). |
| 212 | # |
| 213 | # Deprecation warnings were introduced inconsistently: |
| 214 | # asyncio.get_event_loop warns, but |
| 215 | # asyncio.get_event_loop_policy().get_event_loop does not. Similarly, |
| 216 | # none of the set_event_loop methods warn, although comments on |
| 217 | # https://bugs.python.org/issue39529 indicate that they are also |
| 218 | # intended for future removal. |
| 219 | # |
| 220 | # Therefore, we first attempt to access the event loop with the |
| 221 | # (non-warning) policy method, and if it fails, fall back to creating a |
| 222 | # new event loop. We do not have effective test coverage of the |
| 223 | # new event loop case; this will have to be watched when/if |
| 224 | # get_event_loop is actually removed. |
| 225 | self.should_close_asyncio_loop = False |
| 226 | try: |
| 227 | self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop() |
| 228 | except Exception: |
| 229 | self.asyncio_loop = asyncio.new_event_loop() |
| 230 | self.should_close_asyncio_loop = True |
| 231 | |
| 232 | async def get_loop() -> IOLoop: |
| 233 | return self.get_new_ioloop() |
| 234 | |
| 235 | self.io_loop = self.asyncio_loop.run_until_complete(get_loop()) |
| 236 | with warnings.catch_warnings(): |
| 237 | warnings.simplefilter("ignore", DeprecationWarning) |
| 238 | self.io_loop.make_current() |
| 239 | |
| 240 | def tearDown(self) -> None: |
| 241 | # Native coroutines tend to produce warnings if they're not |
no test coverage detected