def _callTestMethod()

in later/unittest/case.py [0:0]
29 lines of code
11 McCabe index (conditional complexity)

    def _callTestMethod(self, testMethod) -> None:
        ignore_error = getattr(
            self,
            _IGNORE_AIO_ERRS_ATTR,
            getattr(testMethod, _IGNORE_AIO_ERRS_ATTR, False),
        )
        ignore_tasks = getattr(
            self,
            _IGNORE_TASK_LEAKS_ATTR,
            getattr(testMethod, _IGNORE_TASK_LEAKS_ATTR, False),
        )

        # pyre-fixme[16]: `TestCase` has no attribute `_asyncioTestLoop`.
        loop = self._asyncioTestLoop
        if not ignore_tasks:
            # install our own task factory for monitoring usage
            loop.set_task_factory(task_factory)

        # Track existing tasks
        start_tasks = all_tasks(loop)
        # Setup a patch for the asyncio logger
        real_logger = asyncio.log.logger.error
        with mock.patch.object(
            asyncio.log.logger, "error", side_effect=real_logger
        ) as error:
            # pyre-fixme[16]: `AsyncioTestCase` has no attribute `_callTestMethod`.
            super()._callTestMethod(testMethod)

        # Lets join the queue to insure all the tasks created by this case
        # are cleaned up
        # pyre-fixme[16]: `TestCase` has no attribute `_asyncioCallsQueue`.
        loop.run_until_complete(self._asyncioCallsQueue.join())
        left_over_tasks = set(all_tasks(loop)) - set(start_tasks)
        for task in list(left_over_tasks):
            if isinstance(task, TestTask) and task.was_managed():
                left_over_tasks.remove(task)
        if left_over_tasks and not ignore_tasks:
            self.assertEqual(set(), left_over_tasks, "left over un-awaited tasks!")
        if error.called and not ignore_error:
            self.fail(f"asyncio logger.error() was called!\n{error.call_args_list}")