(self, **kwargs: Any)
| 331 | """ |
| 332 | |
| 333 | def initialize(self, **kwargs: Any) -> None: # type: ignore |
| 334 | self.is_current = False |
| 335 | loop = None |
| 336 | if "asyncio_loop" not in kwargs: |
| 337 | kwargs["asyncio_loop"] = loop = asyncio.new_event_loop() |
| 338 | try: |
| 339 | super().initialize(**kwargs) |
| 340 | except Exception: |
| 341 | # If initialize() does not succeed (taking ownership of the loop), |
| 342 | # we have to close it. |
| 343 | if loop is not None: |
| 344 | loop.close() |
| 345 | raise |
| 346 | |
| 347 | def close(self, all_fds: bool = False) -> None: |
| 348 | if self.is_current: |
no test coverage detected