Repository: sanic-org/sanic

Issue:
SANIC_NO_UVLOOP=true doesn't seem to work
**Describe the bug**
Installing sanic with a SANIC_NO_UVLOOP=true in my environment still results in uvloop being installed


**Environment (please complete the following information):**
 - OS: macOS Big Sur
 - Version 11.5.2


**Additional context**
Python 3.9.6

<img width="1494" alt="Screen Shot 2021-08-22 at 9 15 49 am" src="https://user-images.githubusercontent.com/109301/130337105-7599c107-1311-4ae0-8187-46667afe1220.png">



Discussion:
prryplatypus: I've made it `NO_UVLOOP` for consistency, but it's not super user friendly when we do `if not self.config.NO_UVLOOP:`. I'd suggest changing it to `USE_UVLOOP` (both in the install and in config), but make it default to `True` instead.
ahopkins: That's a breaking change. Not idea but I think keeping it negative might be the way to go.
prryplatypus: So we're not willing to do this? 🙈 
```py
import os

USE_UVLOOP = os.getenv('USE_UVLOOP', True)
no_uvloop = os.getenv('NO_UVLOOP', None)

if no_uvloop is not None:
    # Raise DeprecationWarning
    USE_UVLOOP = no_uvloop
```
ahopkins: Why do they need to be connected? The `SANIC_NO_UVLOOP` variable is meant to be an install time switch. Something that itself needs to be fixed (if possible). But, if it is set, then theoretically, you should not have uvloop installed, and it should not matter.

I think we can just go forward with `Config.USE_UVLOOP` and default to that. If there is a conflict (``USE_UVLOOP == True` and `NO_UVLOOP == True`), then this is a situation that we could raise a warning. But, this should only matter if uvloop is installed. If it is not installed, and you have `USE_UVLOOP == True`, then that is another scenario we need to raise a startup warning.
prryplatypus: Alright, that sounds good to me!
prryplatypus: I uhhh... may need some help with these tests. I'm not too used to writing tests, and I think this is not one of the easiest things to write them for either 😅. Could anyone lend me a hand, please?
prryplatypus: I'll probably wait for #2295 to be merged before completing this one. There's some things in that PR that would make this change easier/cleaner.
prryplatypus: Hey y'all, sorry for the radio silence lately! There was something with this change I wasn't too sure about and wanted to test before actually trying to merge it into the repo, and I really didn't have time to try it out... until today!

My main concern was whether changing the event loop policy within an asynchronous function (e.g.: `create_server`) was useful at all, or whether the loop itself was not going to change because it was already running. After testing it out today, I believe my suspicions are correct, and if an event loop is already running, `asyncio.get_event_loop()` will still return the same loop even after changing the event loop policy to use uvloop until the running loop is no longer running:
```py
import asyncio
import uvloop

async def test():
        loop_type = type(asyncio.get_event_loop())
        new_loop_type = type(asyncio.new_event_loop())
        print(f"In function, before change: {loop_type}")
        print(f"In function, before change (new): {new_loop_type}")

        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

        loop_type = type(asyncio.get_event_loop())
        new_loop_type = type(asyncio.new_event_loop())
        print(f"In function, after change: {loop_type}")
        print(f"In function, after change (new): {new_loop_type}")


loop = asyncio.get_event_loop()
new_loop = asyncio.new_event_loop()
print(f"Out of function, before change: {type(loop)}")
print(f"Out of function, before change (new): {type(new_loop)}")

loop.run_until_complete(test())

loop = asyncio.get_event_loop()
new_loop = asyncio.new_event_loop()
print(f"Out of function, after change: {type(loop)}")
print(f"Out of function, after change (new): {type(new_loop)}")
```
![Results](https://i.imgur.com/5GrQ52J.png)

Therefore, I would like to suggest **NOT** changing the event loop policy within async methods such as `create_server` (as these use an already-running loop for the worker that we can't change), and instead only change the loop policy within non-async methods such as `run` (where each worker uses a new event loop created via `asyncio.new_event_loop()` which would be affected by changing the policy).

Please let me know your thoughts!
prryplatypus: Ok, just missing tests for logging and should be good to go!
Might also move the `use_uvloop` method out of `__init__.py`.
ahopkins: Agreed, we should not be changing the loop inside of a coroutine.
prryplatypus: > Looks good. When you are ready, LMK and I will pull locally. But, this looks like the right direction.

I think I am ready if you want to pull locally!

Wondering if in `create_server` maybe we should check if the config value is not default and if it isn't, check if the running loop matches the config value (e.g: not from `uvloop` is `USE_UVLOOP` is `False`, or viceversa) and raise a warning if not. For now I haven't because -up until now- it's just been failing silently.
prryplatypus: Things worth noting:

- If running more than one app, do we want to raise a warning if their configs are different? Is there any way we can globally know the state of the loop policy?
- When running the app with `create_server`, do we maybe want to check if the config loop policy matches the currently running event loop and raise a warning only if it doesn't match? For example, if `config.USE_UVLOOP` is explicitly set to `True` but `asyncio.get_event_loop()` doesn't return a `uvloop.Loop` instance.
- When running Sanic with Gunicorn (`worker.py`), uvloop is always configured to be used. Do we want to add a warning about this or just put it in the docs?
prryplatypus: > Hmm... Yes, but if one is set to _default then I suppose it does not matter. The issue would be if one is True and one False. Correct?
Correct! Or if one is _default and the other one if False and the _default one is started first (as it will then override the event loop config).

> `create_server` is a pretty DIY. If anything, I would say that we could do a check in the `AsyncioServer.startup` to see if it is not `_default`, in which case we throw a warning. But no need to check the types.
We already do this inside the `create_server` function, so I guess that works then!

> Docs
💯 
prryplatypus: This is as far as I can get. I've tried adding a test for the asgi warning, but have not succeeded at it. Any help is welcome. That should be the last thing left to do!

Patch:
diff --git a/sanic/app.py b/sanic/app.py
index 751d1800..27e0c650 100644
--- a/sanic/app.py
+++ b/sanic/app.py
@@ -69,6 +69,7 @@ from sanic.exceptions import (
     URLBuildError,
 )
 from sanic.handlers import ErrorHandler
+from sanic.helpers import _default
 from sanic.http import Stage
 from sanic.log import LOGGING_CONFIG_DEFAULTS, Colors, error_logger, logger
 from sanic.mixins.listeners import ListenerEvent
@@ -88,7 +89,7 @@ from sanic.response import BaseHTTPResponse, HTTPResponse
 from sanic.router import Router
 from sanic.server import AsyncioServer, HttpProtocol
 from sanic.server import Signal as ServerSignal
-from sanic.server import serve, serve_multiple, serve_single
+from sanic.server import serve, serve_multiple, serve_single, try_use_uvloop
 from sanic.server.protocols.websocket_protocol import WebSocketProtocol
 from sanic.server.websockets.impl import ConnectionClosed
 from sanic.signals import Signal, SignalRouter
@@ -130,6 +131,7 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta):
         "_task_registry",
         "_test_client",
         "_test_manager",
+        "_uvloop_setting",  # TODO: Remove in v22.6
         "asgi",
         "auto_reload",
         "auto_reload",
@@ -159,6 +161,7 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta):
     )
 
     _app_registry: Dict[str, "Sanic"] = {}
+    _uvloop_setting = None
     test_mode = False
 
     def __init__(
@@ -1142,6 +1145,11 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta):
             register_sys_signals=register_sys_signals,
         )
 
+        if self.config.USE_UVLOOP is True or (
+            self.config.USE_UVLOOP is _default and not OS_IS_WINDOWS
+        ):
+            try_use_uvloop()
+
         try:
             self.is_running = True
             self.is_stopping = False
@@ -1239,12 +1247,13 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta):
                 WebSocketProtocol if self.websocket_enabled else HttpProtocol
             )
 
-        # if access_log is passed explicitly change config.ACCESS_LOG
-        if access_log is not None:
-            self.config.ACCESS_LOG = access_log
-
-        if noisy_exceptions is not None:
-            self.config.NOISY_EXCEPTIONS = noisy_exceptions
+        # Set explicitly passed configuration values
+        for attribute, value in {
+            "ACCESS_LOG": access_log,
+            "NOISY_EXCEPTIONS": noisy_exceptions,
+        }.items():
+            if value is not None:
+                setattr(self.config, attribute, value)
 
         server_settings = self._helper(
             host=host,
@@ -1259,6 +1268,14 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta):
             run_async=return_asyncio_server,
         )
 
+        if self.config.USE_UVLOOP is not _default:
+            error_logger.warning(
+                "You are trying to change the uvloop configuration, but "
+                "this is only effective when using the run(...) method. "
+                "When using the create_server(...) method Sanic will use "
+                "the already existing loop."
+            )
+
         main_start = server_settings.pop("main_start", None)
         main_stop = server_settings.pop("main_stop", None)
         if main_start or main_stop:
@@ -1833,6 +1850,19 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta):
         self._future_registry.clear()
         self.signalize()
         self.finalize()
+
+        # TODO: Replace in v22.6 to check against apps in app registry
+        if (
+            self.__class__._uvloop_setting is not None
+            and self.__class__._uvloop_setting != self.config.USE_UVLOOP
+        ):
+            error_logger.warning(
+                "It looks like you're running several apps with different "
+                "uvloop settings. This is not supported and may lead to "
+                "unintended behaviour."
+            )
+        self.__class__._uvloop_setting = self.config.USE_UVLOOP
+
         ErrorHandler.finalize(self.error_handler, config=self.config)
         TouchUp.run(self)
         self.state.is_started = True
diff --git a/sanic/asgi.py b/sanic/asgi.py
index 00b181dc..5ef15a91 100644
--- a/sanic/asgi.py
+++ b/sanic/asgi.py
@@ -7,6 +7,7 @@ import sanic.app  # noqa
 
 from sanic.compat import Header
 from sanic.exceptions import ServerError
+from sanic.helpers import _default
 from sanic.http import Stage
 from sanic.models.asgi import ASGIReceive, ASGIScope, ASGISend, MockTransport
 from sanic.request import Request
@@ -53,6 +54,13 @@ class Lifespan:
         await self.asgi_app.sanic_app._server_event("init", "before")
         await self.asgi_app.sanic_app._server_event("init", "after")
 
+        if self.asgi_app.sanic_app.config.USE_UVLOOP is not _default:
+            warnings.warn(
+                "You have set the USE_UVLOOP configuration option, but Sanic "
+                "cannot control the event loop when running in ASGI mode."
+                "This option will be ignored."
+            )
+
     async def shutdown(self) -> None:
         """
         Gather the listeners to fire on server stop.
diff --git a/sanic/compat.py b/sanic/compat.py
index 87278267..e28e64a1 100644
--- a/sanic/compat.py
+++ b/sanic/compat.py
@@ -8,6 +8,14 @@ from multidict import CIMultiDict  # type: ignore
 
 
 OS_IS_WINDOWS = os.name == "nt"
+UVLOOP_INSTALLED = False
+
+try:
+    import uvloop  # type: ignore # noqa
+
+    UVLOOP_INSTALLED = True
+except ImportError:
+    pass
 
 
 def enable_windows_color_support():
diff --git a/sanic/config.py b/sanic/config.py
index d4057a71..1e770f67 100644
--- a/sanic/config.py
+++ b/sanic/config.py
@@ -7,7 +7,7 @@ from typing import Any, Callable, Dict, Optional, Sequence, Union
 from warnings import warn
 
 from sanic.errorpages import DEFAULT_FORMAT, check_error_format
-from sanic.helpers import _default
+from sanic.helpers import Default, _default
 from sanic.http import Http
 from sanic.log import error_logger
 from sanic.utils import load_module_from_file_location, str_to_bool
@@ -38,6 +38,7 @@ DEFAULT_CONFIG = {
     "REQUEST_MAX_SIZE": 100000000,  # 100 megabytes
     "REQUEST_TIMEOUT": 60,  # 60 seconds
     "RESPONSE_TIMEOUT": 60,  # 60 seconds
+    "USE_UVLOOP": _default,
     "WEBSOCKET_MAX_SIZE": 2 ** 20,  # 1 megabyte
     "WEBSOCKET_PING_INTERVAL": 20,
     "WEBSOCKET_PING_TIMEOUT": 20,
@@ -79,6 +80,7 @@ class Config(dict, metaclass=DescriptorMeta):
     REQUEST_TIMEOUT: int
     RESPONSE_TIMEOUT: int
     SERVER_NAME: str
+    USE_UVLOOP: Union[Default, bool]
     WEBSOCKET_MAX_SIZE: int
     WEBSOCKET_PING_INTERVAL: int
     WEBSOCKET_PING_TIMEOUT: int
diff --git a/sanic/server/__init__.py b/sanic/server/__init__.py
index 8e26dcd0..116bd05c 100644
--- a/sanic/server/__init__.py
+++ b/sanic/server/__init__.py
@@ -1,20 +1,10 @@
-import asyncio
-
 from sanic.models.server_types import ConnInfo, Signal
 from sanic.server.async_server import AsyncioServer
+from sanic.server.loop import try_use_uvloop
 from sanic.server.protocols.http_protocol import HttpProtocol
 from sanic.server.runners import serve, serve_multiple, serve_single
 
 
-try:
-    import uvloop  # type: ignore
-
-    if not isinstance(asyncio.get_event_loop_policy(), uvloop.EventLoopPolicy):
-        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-except ImportError:
-    pass
-
-
 __all__ = (
     "AsyncioServer",
     "ConnInfo",
@@ -23,4 +13,5 @@ __all__ = (
     "serve",
     "serve_multiple",
     "serve_single",
+    "try_use_uvloop",
 )
diff --git a/sanic/server/loop.py b/sanic/server/loop.py
new file mode 100644
index 00000000..5613f709
--- /dev/null
+++ b/sanic/server/loop.py
@@ -0,0 +1,49 @@
+import asyncio
+
+from distutils.util import strtobool
+from os import getenv
+
+from sanic.compat import OS_IS_WINDOWS
+from sanic.log import error_logger
+
+
+def try_use_uvloop() -> None:
+    """
+    Use uvloop instead of the default asyncio loop.
+    """
+    if OS_IS_WINDOWS:
+        error_logger.warning(
+            "You are trying to use uvloop, but uvloop is not compatible "
+            "with your system. You can disable uvloop completely by setting "
+            "the 'USE_UVLOOP' configuration value to false, or simply not "
+            "defining it and letting Sanic handle it for you. Sanic will now "
+            "continue to run using the default event loop."
+        )
+        return
+
+    try:
+        import uvloop  # type: ignore
+    except ImportError:
+        error_logger.warning(
+            "You are trying to use uvloop, but uvloop is not "
+            "installed in your system. In order to use uvloop "
+            "you must first install it. Otherwise, you can disable "
+            "uvloop completely by setting the 'USE_UVLOOP' "
+            "configuration value to false. Sanic will now continue "
+            "to run with the default event loop."
+        )
+        return
+
+    uvloop_install_removed = strtobool(getenv("SANIC_NO_UVLOOP", "no"))
+    if uvloop_install_removed:
+        error_logger.info(
+            "You are requesting to run Sanic using uvloop, but the "
+            "install-time 'SANIC_NO_UVLOOP' environment variable (used to "
+            "opt-out of installing uvloop with Sanic) is set to true. If "
+            "you want to prevent Sanic from overriding the event loop policy "
+            "during runtime, set the 'USE_UVLOOP' configuration value to "
+            "false."
+        )
+
+    if not isinstance(asyncio.get_event_loop_policy(), uvloop.EventLoopPolicy):
+        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
diff --git a/sanic/worker.py b/sanic/worker.py
index a3bc29b8..cdc238e2 100644
--- a/sanic/worker.py
+++ b/sanic/worker.py
@@ -7,8 +7,9 @@ import traceback
 
 from gunicorn.workers import base  # type: ignore
 
+from sanic.compat import UVLOOP_INSTALLED
 from sanic.log import logger
-from sanic.server import HttpProtocol, Signal, serve
+from sanic.server import HttpProtocol, Signal, serve, try_use_uvloop
 from sanic.server.protocols.websocket_protocol import WebSocketProtocol
 
 
@@ -17,12 +18,8 @@ try:
 except ImportError:
     ssl = None  # type: ignore
 
-try:
-    import uvloop  # type: ignore
-
-    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-except ImportError:
-    pass
+if UVLOOP_INSTALLED:
+    try_use_uvloop()
 
 
 class GunicornWorker(base.Worker):


Test patch:
diff --git a/tests/test_app.py b/tests/test_app.py
index 9bcc87e7..6ce8ae65 100644
--- a/tests/test_app.py
+++ b/tests/test_app.py
@@ -2,16 +2,20 @@ import asyncio
 import logging
 import re
 
+from collections import Counter
 from inspect import isawaitable
 from os import environ
 from unittest.mock import Mock, patch
 
 import pytest
+import sanic
 
 from sanic import Sanic
+from sanic.compat import OS_IS_WINDOWS, UVLOOP_INSTALLED
 from sanic.config import Config
 from sanic.exceptions import SanicException
 from sanic.response import text
+from sanic.helpers import _default
 
 
 @pytest.fixture(autouse=True)
@@ -19,15 +23,6 @@ def clear_app_registry():
     Sanic._app_registry = {}
 
 
-def uvloop_installed():
-    try:
-        import uvloop  # noqa
-
-        return True
-    except ImportError:
-        return False
-
-
 def test_app_loop_running(app):
     @app.get("/test")
     async def handler(request):
@@ -472,6 +467,102 @@ def test_custom_context():
     assert app.ctx == ctx
 
 
+def test_uvloop_config(app, monkeypatch):
+    @app.get("/test")
+    def handler(request):
+        return text("ok")
+
+    try_use_uvloop = Mock()
+    monkeypatch.setattr(sanic.app, "try_use_uvloop", try_use_uvloop)
+
+    # Default config
+    app.test_client.get("/test")
+    if OS_IS_WINDOWS:
+        try_use_uvloop.assert_not_called()
+    else:
+        try_use_uvloop.assert_called_once()
+
+    try_use_uvloop.reset_mock()
+    app.config["USE_UVLOOP"] = False
+    app.test_client.get("/test")
+    try_use_uvloop.assert_not_called()
+
+    try_use_uvloop.reset_mock()
+    app.config["USE_UVLOOP"] = True
+    app.test_client.get("/test")
+    try_use_uvloop.assert_called_once()
+
+
+def test_uvloop_cannot_never_called_with_create_server(caplog, monkeypatch):
+    apps = (
+        Sanic("default-uvloop"),
+        Sanic("no-uvloop"),
+        Sanic("yes-uvloop")
+    )
+
+    apps[1].config.USE_UVLOOP = False
+    apps[2].config.USE_UVLOOP = True
+
+    try_use_uvloop = Mock()
+    monkeypatch.setattr(sanic.app, "try_use_uvloop", try_use_uvloop)
+
+    loop = asyncio.get_event_loop()
+
+    with caplog.at_level(logging.WARNING):
+        for app in apps:
+            srv_coro = app.create_server(
+                return_asyncio_server=True,
+                asyncio_server_kwargs=dict(start_serving=False)
+            )
+            loop.run_until_complete(srv_coro)
+
+    try_use_uvloop.assert_not_called()  # Check it didn't try to change policy
+
+    message = (
+        "You are trying to change the uvloop configuration, but "
+        "this is only effective when using the run(...) method. "
+        "When using the create_server(...) method Sanic will use "
+        "the already existing loop."
+    )
+
+    counter = Counter([(r[1], r[2]) for r in caplog.record_tuples])
+    modified = sum(1 for app in apps if app.config.USE_UVLOOP is not _default)
+
+    assert counter[(logging.WARNING, message)] == modified
+
+
+def test_multiple_uvloop_configs_display_warning(caplog):
+    Sanic._uvloop_setting = None  # Reset the setting (changed in prev tests)
+
+    default_uvloop = Sanic("default-uvloop")
+    no_uvloop = Sanic("no-uvloop")
+    yes_uvloop = Sanic("yes-uvloop")
+
+    no_uvloop.config.USE_UVLOOP = False
+    yes_uvloop.config.USE_UVLOOP = True
+
+    loop = asyncio.get_event_loop()
+
+    with caplog.at_level(logging.WARNING):
+        for app in (default_uvloop, no_uvloop, yes_uvloop):
+            srv_coro = app.create_server(
+                return_asyncio_server=True,
+                asyncio_server_kwargs=dict(start_serving=False)
+            )
+            srv = loop.run_until_complete(srv_coro)
+            loop.run_until_complete(srv.startup())
+
+    message = (
+        "It looks like you're running several apps with different "
+        "uvloop settings. This is not supported and may lead to "
+        "unintended behaviour."
+    )
+
+    counter = Counter([(r[1], r[2]) for r in caplog.record_tuples])
+
+    assert counter[(logging.WARNING, message)] == 2
+
+
 def test_cannot_run_fast_and_workers(app):
     message = "You cannot use both fast=True and workers=X"
     with pytest.raises(RuntimeError, match=message):
diff --git a/tests/test_asgi.py b/tests/test_asgi.py
index 3d464a4f..d00a70bd 100644
--- a/tests/test_asgi.py
+++ b/tests/test_asgi.py
@@ -145,6 +145,37 @@ def test_listeners_triggered_async(app):
     assert after_server_stop
 
 
+def test_non_default_uvloop_config_raises_warning(app):
+    app.config.USE_UVLOOP = True
+
+    class CustomServer(uvicorn.Server):
+        def install_signal_handlers(self):
+            pass
+
+    config = uvicorn.Config(app=app, loop="asyncio", limit_max_requests=0)
+    server = CustomServer(config=config)
+
+    with pytest.warns(UserWarning) as records:
+        server.run()
+
+    all_tasks = asyncio.all_tasks(asyncio.get_event_loop())
+    for task in all_tasks:
+        task.cancel()
+
+    msg = ""
+    for record in records:
+        _msg = str(record.message)
+        if _msg.startswith("You have set the USE_UVLOOP configuration"):
+            msg = _msg
+            break
+
+    assert msg == (
+        "You have set the USE_UVLOOP configuration option, but Sanic "
+        "cannot control the event loop when running in ASGI mode."
+        "This option will be ignored."
+    )
+
+
 @pytest.mark.asyncio
 async def test_mockprotocol_events(protocol):
     assert protocol._not_paused.is_set()
diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py
index eea97935..cfc0f832 100644
--- a/tests/test_exceptions.py
+++ b/tests/test_exceptions.py
@@ -271,9 +271,13 @@ def test_exception_in_ws_logged(caplog):
     with caplog.at_level(logging.INFO):
         app.test_client.websocket("/feed")
 
-    error_logs = [r for r in caplog.record_tuples if r[0] == "sanic.error"]
-    assert error_logs[1][1] == logging.ERROR
-    assert "Exception occurred while handling uri:" in error_logs[1][2]
+    for record in caplog.record_tuples:
+        if record[2].startswith("Exception occurred"):
+            break
+
+    assert record[0] == "sanic.error"
+    assert record[1] == logging.ERROR
+    assert "Exception occurred while handling uri:" in record[2]
 
 
 @pytest.mark.parametrize("debug", (True, False))
diff --git a/tests/test_graceful_shutdown.py b/tests/test_graceful_shutdown.py
index 1733ffd1..54ba92d8 100644
--- a/tests/test_graceful_shutdown.py
+++ b/tests/test_graceful_shutdown.py
@@ -2,7 +2,6 @@ import asyncio
 import logging
 import time
 
-from collections import Counter
 from multiprocessing import Process
 
 import httpx
@@ -36,11 +35,14 @@ def test_no_exceptions_when_cancel_pending_request(app, caplog):
 
     p.kill()
 
-    counter = Counter([r[1] for r in caplog.record_tuples])
-
-    assert counter[logging.INFO] == 11
-    assert logging.ERROR not in counter
-    assert (
-        caplog.record_tuples[9][2]
-        == "Request: GET http://127.0.0.1:8000/ stopped. Transport is closed."
-    )
+    info = 0
+    for record in caplog.record_tuples:
+        assert record[1] != logging.ERROR
+        if record[1] == logging.INFO:
+            info += 1
+        if record[2].startswith("Request:"):
+            assert record[2] == (
+                "Request: GET http://127.0.0.1:8000/ stopped. "
+                "Transport is closed."
+            )
+    assert info == 11
diff --git a/tests/test_server_loop.py b/tests/test_server_loop.py
new file mode 100644
index 00000000..fbd5cc2b
--- /dev/null
+++ b/tests/test_server_loop.py
@@ -0,0 +1,116 @@
+import logging
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from sanic.server import loop
+from sanic.compat import OS_IS_WINDOWS, UVLOOP_INSTALLED
+
+
+@pytest.mark.skipif(
+    not OS_IS_WINDOWS,
+    reason="Not testable with current client",
+)
+def test_raises_warning_if_os_is_windows(caplog):
+    with caplog.at_level(logging.WARNING):
+        loop.try_use_uvloop()
+
+    for record in caplog.records:
+        if record.message.startswith("You are trying to use"):
+            break
+
+    assert record.message == (
+        "You are trying to use uvloop, but uvloop is not compatible "
+        "with your system. You can disable uvloop completely by setting "
+        "the 'USE_UVLOOP' configuration value to false, or simply not "
+        "defining it and letting Sanic handle it for you. Sanic will now "
+        "continue to run using the default event loop."
+    )
+
+
+@pytest.mark.skipif(
+    OS_IS_WINDOWS or UVLOOP_INSTALLED,
+    reason="Not testable with current client",
+)
+def test_raises_warning_if_uvloop_not_installed(caplog):
+    with caplog.at_level(logging.WARNING):
+        loop.try_use_uvloop()
+
+    for record in caplog.records:
+        if record.message.startswith("You are trying to use"):
+            break
+
+    assert record.message == (
+        "You are trying to use uvloop, but uvloop is not "
+        "installed in your system. In order to use uvloop "
+        "you must first install it. Otherwise, you can disable "
+        "uvloop completely by setting the 'USE_UVLOOP' "
+        "configuration value to false. Sanic will now continue "
+        "to run with the default event loop."
+    )
+
+
+@pytest.mark.skipif(
+    OS_IS_WINDOWS or not UVLOOP_INSTALLED,
+    reason="Not testable with current client",
+)
+def test_logs_when_install_and_runtime_config_mismatch(caplog, monkeypatch):
+    getenv = Mock(return_value="no")
+    monkeypatch.setattr(loop, "getenv", getenv)
+
+    with caplog.at_level(logging.INFO):
+        loop.try_use_uvloop()
+
+    getenv.assert_called_once_with("SANIC_NO_UVLOOP", "no")
+    assert caplog.record_tuples == []
+
+    getenv = Mock(return_value="yes")
+    monkeypatch.setattr(loop, "getenv", getenv)
+    with caplog.at_level(logging.INFO):
+        loop.try_use_uvloop()
+
+    getenv.assert_called_once_with("SANIC_NO_UVLOOP", "no")
+    for record in caplog.records:
+        if record.message.startswith("You are requesting to run"):
+            break
+
+    assert record.message == (
+        "You are requesting to run Sanic using uvloop, but the "
+        "install-time 'SANIC_NO_UVLOOP' environment variable (used to "
+        "opt-out of installing uvloop with Sanic) is set to true. If "
+        "you want to prevent Sanic from overriding the event loop policy "
+        "during runtime, set the 'USE_UVLOOP' configuration value to "
+        "false."
+    )
+
+
+@pytest.mark.skipif(
+    OS_IS_WINDOWS or not UVLOOP_INSTALLED,
+    reason="Not testable with current client",
+)
+def test_sets_loop_policy_only_when_not_already_set(monkeypatch):
+    import uvloop  # type: ignore
+
+    # Existing policy is not uvloop.EventLoopPolicy
+    get_event_loop_policy = Mock(return_value=None)
+    monkeypatch.setattr(
+        loop.asyncio, "get_event_loop_policy", get_event_loop_policy
+    )
+
+    with patch("asyncio.set_event_loop_policy") as set_event_loop_policy:
+        loop.try_use_uvloop()
+        set_event_loop_policy.assert_called_once()
+        args, _ = set_event_loop_policy.call_args
+        policy = args[0]
+        assert isinstance(policy, uvloop.EventLoopPolicy)
+
+    # Existing policy is uvloop.EventLoopPolicy
+    get_event_loop_policy = Mock(return_value=policy)
+    monkeypatch.setattr(
+        loop.asyncio, "get_event_loop_policy", get_event_loop_policy
+    )
+
+    with patch("asyncio.set_event_loop_policy") as set_event_loop_policy:
+        loop.try_use_uvloop()
+        set_event_loop_policy.assert_not_called()