Method tearDown

asyncpg/_testbase/__init__.py:119–140

(self)

        self.__unhandled_exceptions = []

    def tearDown(self):
        excs = []
        for exc in self.__unhandled_exceptions:
            if isinstance(exc, ConnectionResetError):
                texc = traceback.TracebackException.from_exception(
                    exc, lookup_lines=False)
                if texc.stack[-1].name == "_call_connection_lost":
                    # On Windows calling socket.shutdown may raise
                    # ConnectionResetError, which happens in the
                    # finally block of _call_connection_lost.
                    continue
            excs.append(exc)

        if excs:
            formatted = []

            for i, context in enumerate(excs):
                formatted.append(self._format_loop_exception(context, i + 1))

            self.fail(
                'unexpected exceptions in asynchronous code:\n' +
                '\n'.join(formatted))

    @contextlib.contextmanager
    def assertRunUnder(self, delta):