def asyncio_orchestration_debug()

in idb/utils/testing.py [0:0]

• 17 lines of code
• 7 McCabe index (conditional complexity)

    def asyncio_orchestration_debug(self, testMethod, b_log, c_log):
        asyncio.set_event_loop(self.loop)
        real_logger = logging.getLogger("asyncio").error
        c_log.error.side_effect = b_log.error.side_effect = real_logger
        # Don't make testmethods cleanup tasks that existed before them
        before_tasks = asyncio.all_tasks(self.loop)
        _tasks_warning(before_tasks)
        debug_async = self.debug_async(testMethod)
        self.loop.run_until_complete(debug_async)

        if c_log.error.called or b_log.error.called:
            self.fail("asyncio logger.error() called!")
        # Sometimes we end up with a reference to our task for debug_async
        tasks = {
            t
            for t in asyncio.all_tasks(self.loop) - before_tasks
            if not (t._coro == debug_async and t.done())
        }
        del before_tasks
        self.assertEqual(set(), tasks, "left over asyncio tasks!")