def _patch_asyncio() -> None:
    # This patches asyncio to add a sync_wait method to the event
    # loop. This method can then be called from within a task
    # including a synchronous function called from a task. Sadly it
    # requires the python Task and Future implementations, which
    # invokes some performance cost.
    asyncio.Task = asyncio.tasks._CTask = asyncio.tasks.Task = asyncio.tasks._PyTask  # type: ignore
    asyncio.Future = (  # type: ignore
        asyncio.futures._CFuture  # type: ignore
    ) = asyncio.futures.Future = asyncio.futures._PyFuture  # type: ignore # noqa

    current_policy = asyncio.get_event_loop_policy()
    if hasattr(asyncio, "unix_events"):
        target_policy = asyncio.unix_events._UnixDefaultEventLoopPolicy
    else:
        target_policy = object  # type: ignore

    if not isinstance(current_policy, target_policy):
        raise RuntimeError("Flask Patching only works with the default event loop policy")

    _patch_loop()
    _patch_task() 

def _init_asyncio_patch():
    """
    Select compatible event loop for Tornado 5+.
    As of Python 3.8, the default event loop on Windows is `proactor`,
    however Tornado requires the old default "selector" event loop.
    As Tornado has decided to leave this to users to set, MkDocs needs
    to set it. See https://github.com/tornadoweb/tornado/issues/2608.
    """
    if sys.platform.startswith("win") and sys.version_info >= (3, 8):
        import asyncio
        try:
            from asyncio import WindowsSelectorEventLoopPolicy
        except ImportError:
            pass  # Can't assign a policy which doesn't exist.
        else:
            if not isinstance(asyncio.get_event_loop_policy(), WindowsSelectorEventLoopPolicy):
                asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy()) 

def event_loop(request):
    """
    Create an instance of the requested event loop.
    """
    default_event_loop(request=request)

    # Close previous event loop
    policy = asyncio.get_event_loop_policy()
    policy.get_event_loop().close()

    # Create new event loop
    _event_loop = policy.new_event_loop()
    policy.set_event_loop(_event_loop)

    def fin():
        _event_loop.close()

    # Add finaliser and return new event loop
    request.addfinalizer(fin)
    return _event_loop 

def event_loop(request):
    """Overwrite `pytest_asyncio` eventloop to fix Windows issue.

    Default implementation causes `NotImplementedError` on Windows with
    Python 3.8, because they changed default eventloop in 3.8.

    NOTE: We do the same thing in the `example/settings.py` because it
    imports (and fails) before we have a chance to invoke this fixture.
    So, we could avoid adding this fixture, but I feel it is better to
    keep the proper solution here as well.

    """
    if sys.platform == "win32" and sys.version_info.minor >= 8:
        asyncio.set_event_loop_policy(
            asyncio.WindowsSelectorEventLoopPolicy()  # pylint: disable=no-member
        )
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close() 

def __init__(self, conf: Config, analytics_manager=None, skip_components=None,
                 peer_manager=None, **override_components):
        self.conf = conf
        self.skip_components = skip_components or []
        self.loop = asyncio.get_event_loop()
        self.analytics_manager = analytics_manager
        self.component_classes = {}
        self.components = set()
        self.started = asyncio.Event(loop=self.loop)
        self.peer_manager = peer_manager or PeerManager(asyncio.get_event_loop_policy().get_event_loop())

        for component_name, component_class in self.default_component_classes.items():
            if component_name in override_components:
                component_class = override_components.pop(component_name)
            if component_name not in self.skip_components:
                self.component_classes[component_name] = component_class

        if override_components:
            raise SyntaxError("unexpected components: %s" % override_components)

        for component_class in self.component_classes.values():
            self.components.add(component_class(self)) 

def _init_asyncio_patch():
    """
    Select compatible event loop for Tornado 5+.

    As of Python 3.8, the default event loop on Windows is `proactor`,
    however Tornado requires the old default "selector" event loop.
    As Tornado has decided to leave this to users to set, MkDocs needs
    to set it. See https://github.com/tornadoweb/tornado/issues/2608.
    """
    if sys.platform.startswith("win") and sys.version_info >= (3, 8):
        import asyncio
        try:
            from asyncio import WindowsSelectorEventLoopPolicy
        except ImportError:
            pass  # Can't assign a policy which doesn't exist.
        else:
            if not isinstance(asyncio.get_event_loop_policy(), WindowsSelectorEventLoopPolicy):
                asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy()) 

def runner(request):
    def _( ga, **kargs ):
        time.sleep(0.5) # leave the time to shutdown previous instance
        if request.param=="serve":
            return getattr(ga,request.param)(port=10000)
        else:
            return getattr(ga,request.param)(**kargs)

    return _

# @pytest.yield_fixture(scope='session')
# def event_loop(request):
#     """Create an instance of the default event loop for each test case."""
#     loop = asyncio.get_event_loop_policy().new_event_loop()
#     yield loop
#     loop.close() 

def test_get_event_loop_returns_running_loop(self):
        class Policy(asyncio.DefaultEventLoopPolicy):
            def get_event_loop(self):
                raise NotImplementedError

        loop = None

        old_policy = asyncio.get_event_loop_policy()
        try:
            asyncio.set_event_loop_policy(Policy())
            loop = asyncio.new_event_loop()
            self.assertIs(asyncio._get_running_loop(), None)

            async def func():
                self.assertIs(asyncio.get_event_loop(), loop)
                self.assertIs(asyncio._get_running_loop(), loop)

            loop.run_until_complete(func())
        finally:
            asyncio.set_event_loop_policy(old_policy)
            if loop is not None:
                loop.close()

        self.assertIs(asyncio._get_running_loop(), None) 

def event_loop(request):
    """
    Create an instance of the requested event loop.
    """
    default_event_loop(request=request)

    # Close previous event loop
    policy = asyncio.get_event_loop_policy()
    policy.get_event_loop().close()

    # Create new event loop
    _event_loop = policy.new_event_loop()
    policy.set_event_loop(_event_loop)

    def fin():
        _event_loop.close()

    # Add finaliser and return new event loop
    request.addfinalizer(fin)
    return _event_loop 

def event_loop(request):
    """Create an instance of the default event loop for each test case.
    Also catches all warnings and raises exception if there was
    'coroutine was never awaited' wargning.
    """
    loop = asyncio.get_event_loop_policy().new_event_loop()

    with warnings.catch_warnings(record=True) as catched_warnings:
        yield loop
        # Collecting garbage should trigger warning for non-awaited coroutines
        gc.collect()

    for w in catched_warnings:
        if (
                isinstance(w.message, RuntimeWarning) and
                str(w.message).endswith('was never awaited')
        ):
            raise w.message
        else:
            warnings.showwarning(w.message, w.category, w.filename, w.lineno)

    loop.close() 

def _init_asyncio_patch():
    """
    Select compatible event loop for Tornado 5+.

    As of Python 3.8, the default event loop on Windows is `proactor`,
    however Tornado requires the old default "selector" event loop.
    As Tornado has decided to leave this to users to set, MkDocs needs
    to set it. See https://github.com/tornadoweb/tornado/issues/2608.
    """
    if sys.platform.startswith("win") and sys.version_info >= (3, 8):
        import asyncio
        try:
            from asyncio import WindowsSelectorEventLoopPolicy
        except ImportError:
            pass  # Can't assign a policy which doesn't exist.
        else:
            if not isinstance(asyncio.get_event_loop_policy(), WindowsSelectorEventLoopPolicy):
                asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy()) 

def test_event_loop_policy(caplog, policy, policy_name):
    """Test that a the runner switches to a different event loop policy when instructed to."""
    caplog.set_level(logging.INFO)
    component = ShutdownComponent()
    old_policy = asyncio.get_event_loop_policy()
    try:
        run_application(component, event_loop_policy=policy)
    except DistributionNotFound as e:
        pytest.skip(str(e))
    finally:
        asyncio.set_event_loop_policy(old_policy)

    records = [record for record in caplog.records if record.name == 'asphalt.core.runner']
    assert len(records) == 6
    assert records[0].message == 'Running in development mode'
    assert records[1].message == 'Switched event loop policy to %s' % policy_name
    assert records[2].message == 'Starting application'
    assert records[3].message == 'Application started'
    assert records[4].message == 'Stopping application'
    assert records[5].message == 'Application stopped' 

def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close() 

def check(self, src, value):
        policy = asyncio.get_event_loop_policy()
        loop = policy.new_event_loop()
        asyncio.set_event_loop_policy(ForbiddenEventLoopPolicy())
        try:
            loop.run_until_complete(self._check(src, value, loop))
        finally:
            loop.close()
            asyncio.set_event_loop_policy(policy) 

def event_loop():
    from moler.asyncio_runner import cancel_remaining_feeders

    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    # event_loop fixture is autoloaded by @pytest.mark.asyncio decorator
    # and inside some of our async tests we just submit() observer inside runner without stopping it
    # so, we need to stop all submitted futures
    cancel_remaining_feeders(loop)
    loop.close() 

def event_loop(request):
    """Create an instance of the default event loop for each test case."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close() 

def setUp(self):
        self.old = dict(_SCENARIO)
        self.oldsetup = dict(_FIXTURES)
        util._STOP = False
        util._TIMER = None
        self.policy = asyncio.get_event_loop_policy() 

def setUp(self):
        # Trigger a cleanup of the mapping so we start with a clean slate.
        AsyncIOLoop().close()
        # If we don't clean up after ourselves other tests may fail on
        # py34.
        self.orig_policy = asyncio.get_event_loop_policy()
        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy()) 

def setUp(self):
        self.orig_policy = asyncio.get_event_loop_policy()
        self.executor = ThreadPoolExecutor(1) 

def setup_test_loop(loop_factory=asyncio.new_event_loop):
    """Create and return an asyncio.BaseEventLoop instance.

    The caller should also call teardown_test_loop, once they are done
    with the loop.
    """
    loop = loop_factory()
    asyncio.set_event_loop(loop)
    if sys.platform != "win32":
        policy = asyncio.get_event_loop_policy()
        watcher = asyncio.SafeChildWatcher()
        watcher.attach_loop(loop)
        policy.set_child_watcher(watcher)
    return loop 

def close(self):
        def __close(conn):
            self.executor.submit(conn.close)
            self.executor.shutdown(wait=True)
        conn = self.connection
        if not conn:
            return
        self.connection = None
        return asyncio.get_event_loop_policy().get_event_loop().call_later(0.01, __close, conn) 

def setUp(self):
        # new event loop for each test
        policy = asyncio.get_event_loop_policy()
        self.loop = policy.new_event_loop()
        policy.set_event_loop(self.loop) 

def make_proto(self):
        conn = MagicMock()
        conn.nick = "me"
        conn.loop = asyncio.get_event_loop_policy().new_event_loop()
        out = []

        async def func(e):
            out.append(self._filter_event(e))

        conn.bot.process = func
        proto = irc._IrcProtocol(conn)
        return conn, out, proto 

def setUp(self):
            policy = asyncio.get_event_loop_policy()
            self.loop = policy.new_event_loop()
            self.set_event_loop(self.loop)

            watcher = self.Watcher()
            watcher.attach_loop(self.loop)
            policy.set_child_watcher(watcher)
            self.addCleanup(policy.set_child_watcher, None) 

def test_handle_source_traceback(self):
        loop = asyncio.get_event_loop_policy().new_event_loop()
        loop.set_debug(True)
        self.set_event_loop(loop)

        def check_source_traceback(h):
            lineno = sys._getframe(1).f_lineno - 1
            self.assertIsInstance(h._source_traceback, list)
            self.assertEqual(h._source_traceback[-1][:3],
                             (__file__,
                              lineno,
                              'test_handle_source_traceback'))

        # call_soon
        h = loop.call_soon(noop)
        check_source_traceback(h)

        # call_soon_threadsafe
        h = loop.call_soon_threadsafe(noop)
        check_source_traceback(h)

        # call_later
        h = loop.call_later(0, noop)
        check_source_traceback(h)

        # call_at
        h = loop.call_later(0, noop)
        check_source_traceback(h) 

def test_get_event_loop_policy(self):
        policy = asyncio.get_event_loop_policy()
        self.assertIsInstance(policy, asyncio.AbstractEventLoopPolicy)
        self.assertIs(policy, asyncio.get_event_loop_policy()) 

def test_set_event_loop_policy(self):
        self.assertRaises(
            AssertionError, asyncio.set_event_loop_policy, object())

        old_policy = asyncio.get_event_loop_policy()

        policy = asyncio.DefaultEventLoopPolicy()
        asyncio.set_event_loop_policy(policy)
        self.assertIs(policy, asyncio.get_event_loop_policy())
        self.assertIsNot(policy, old_policy) 

def setUp(self):
            policy = asyncio.get_event_loop_policy()
            self.loop = policy.new_event_loop()
            self.set_event_loop(self.loop)

            watcher = self.Watcher()
            watcher.attach_loop(self.loop)
            policy.set_child_watcher(watcher)
            self.addCleanup(policy.set_child_watcher, None) 

def test_handle_source_traceback(self):
        loop = asyncio.get_event_loop_policy().new_event_loop()
        loop.set_debug(True)
        self.set_event_loop(loop)

        def check_source_traceback(h):
            lineno = sys._getframe(1).f_lineno - 1
            self.assertIsInstance(h._source_traceback, list)
            self.assertEqual(h._source_traceback[-1][:3],
                             (__file__,
                              lineno,
                              'test_handle_source_traceback'))

        # call_soon
        h = loop.call_soon(noop)
        check_source_traceback(h)

        # call_soon_threadsafe
        h = loop.call_soon_threadsafe(noop)
        check_source_traceback(h)

        # call_later
        h = loop.call_later(0, noop)
        check_source_traceback(h)

        # call_at
        h = loop.call_later(0, noop)
        check_source_traceback(h) 

def test_get_event_loop_policy(self):
        policy = asyncio.get_event_loop_policy()
        self.assertIsInstance(policy, asyncio.AbstractEventLoopPolicy)
        self.assertIs(policy, asyncio.get_event_loop_policy())