def _raise_if_asyncio_warnings()

in testslide/__init__.py [0:0]

• 49 lines of code
• 6 McCabe index (conditional complexity)

    def _raise_if_asyncio_warnings(self, context_data: _ContextData) -> Iterator[None]:
        if sys.version_info < (3, 7):
            yield
            return
        original_showwarning = warnings.showwarning
        caught_failures: List[Union[Exception, str]] = []

        def showwarning(
            message: str,
            category: Type[Warning],
            filename: str,
            lineno: int,
            file: Optional[TextIO] = None,
            line: Optional[str] = None,
        ) -> None:
            failure_warning_messages: Dict[Any, str] = {
                RuntimeWarning: "^coroutine '.+' was never awaited"
            }
            warning_class = type(message)
            pattern = failure_warning_messages.get(warning_class, None)
            if pattern and re.compile(pattern).match(str(message)):
                caught_failures.append(message)
            else:
                original_showwarning(message, category, filename, lineno, file, line)

        warnings.showwarning = showwarning  # type: ignore

        original_logger_warning = asyncio.log.logger.warning

        def logger_warning(msg: str, *args: Any, **kwargs: Any) -> None:
            if re.compile("^Executing .+ took .+ seconds$").match(str(msg)):
                msg = (
                    f"{msg}\n"
                    "During the execution of the async test a slow callback "
                    "that blocked the event loop was detected.\n"
                    "Tip: you can customize the detection threshold with:\n"
                    "  asyncio.get_running_loop().slow_callback_duration = seconds"
                )
                caught_failures.append(SlowCallback(msg % args))
            else:
                original_logger_warning(msg, *args, **kwargs)

        asyncio.log.logger.warning = logger_warning  # type: ignore

        aggregated_exceptions = AggregatedExceptions()

        try:
            with aggregated_exceptions.catch():
                yield
        finally:
            warnings.showwarning = original_showwarning
            asyncio.log.logger.warning = original_logger_warning  # type: ignore
            for failure in caught_failures:
                with aggregated_exceptions.catch():
                    raise failure  # type: ignore
            aggregated_exceptions.raise_correct_exception()