improve logging test and increase coverage on server.py

setup.cfg

@@ -82,7 +82,7 @@ plugins =
 
 [coverage:report]
 precision = 2
-fail_under = 97.92
+fail_under = 98.56
 show_missing = true
 skip_covered = true
 exclude_lines =

tests/middleware/test_logging.py

@@ -1,5 +1,8 @@
+import asyncio
 import contextlib
 import logging
+import socket
+import sys
 
 import httpx
 import pytest
@@ -155,6 +158,59 @@ async def test_default_logging(use_colors, caplog, logging_config):
         assert "Shutting down" in messages.pop(0)
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("use_colors", [(True), (False)])
+@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
+async def test_default_logging_with_uds(
+    use_colors, caplog, logging_config, short_socket_name
+):  # pragma: py-win32
+    config = Config(
+        app=app, use_colors=use_colors, log_config=logging_config, uds=short_socket_name
+    )
+    with caplog_for_logger(caplog, "uvicorn.access"):
+        async with run_server(config):
+            transport = httpx.AsyncHTTPTransport(uds=short_socket_name)
+            async with httpx.AsyncClient(transport=transport) as client:
+                response = await client.get("http://my")
+        assert response.status_code == 204
+
+        messages = [
+            record.message for record in caplog.records if "uvicorn" in record.name
+        ]
+        assert "Started server process" in messages.pop(0)
+        assert "Waiting for application startup" in messages.pop(0)
+        assert "ASGI 'lifespan' protocol appears unsupported" in messages.pop(0)
+        assert "Application startup complete" in messages.pop(0)
+        assert "Uvicorn running on unix socket " + short_socket_name in messages.pop(0)
+        assert '"GET / HTTP/1.1" 204' in messages.pop(0)
+        assert "Shutting down" in messages.pop(0)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("use_colors", [(True), (False)])
+@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
+async def test_default_logging_with_fd(
+    use_colors, caplog, logging_config
+):  # pragma: py-win32
+    fdsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    fd = fdsock.fileno()
+    config = Config(app=app, use_colors=use_colors, log_config=logging_config, fd=fd)
+    with caplog_for_logger(caplog, "uvicorn.access"):
+        async with run_server(config):
+            await asyncio.sleep(0.1)
+            fdsock.listen(9)
+        messages = [
+            record.message for record in caplog.records if "uvicorn" in record.name
+        ]
+        assert "Started server process" in messages.pop(0)
+        assert "Waiting for application startup" in messages.pop(0)
+        assert "ASGI 'lifespan' protocol appears unsupported" in messages.pop(0)
+        assert "Application startup complete" in messages.pop(0)
+        assert f"Uvicorn running on socket {fdsock.getsockname()}" in messages.pop(0)
+        assert "Shutting down" in messages.pop(0)
+    fdsock.close()
+
+
 @pytest.mark.anyio
 async def test_unknown_status_code(caplog):
     async def app(scope, receive, send):
@@ -175,3 +231,25 @@ async def app(scope, receive, send):
             if record.name == "uvicorn.access"
         ]
         assert '"GET / HTTP/1.1" 599' in messages.pop()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("use_colors", [True, False])
+async def test_default_logging_with_port_zero(use_colors, caplog, logging_config):
+    config = Config(app=app, use_colors=use_colors, log_config=logging_config, port=0)
+    with caplog_for_logger(caplog, "uvicorn.access"):
+        async with run_server(config) as server:
+            while not server.started:
+                await asyncio.sleep(0.1)
+            for s in server.servers:
+                for sock in s.sockets:
+                    host, port = sock.getsockname()
+        messages = [
+            record.message for record in caplog.records if "uvicorn" in record.name
+        ]
+        assert "Started server process" in messages.pop(0)
+        assert "Waiting for application startup" in messages.pop(0)
+        assert "ASGI 'lifespan' protocol appears unsupported" in messages.pop(0)
+        assert "Application startup complete" in messages.pop(0)
+        assert f"Uvicorn running on http://{host}:{port}" in messages.pop(0)
+        assert "Shutting down" in messages.pop(0)

tests/test_main.py

@@ -1,3 +1,4 @@
+import asyncio
 import inspect
 import socket
 from logging import WARNING
@@ -6,6 +7,7 @@
 import pytest
 
 from tests.utils import run_server
+from uvicorn import Server
 from uvicorn.config import Config
 from uvicorn.main import run
 
@@ -122,3 +124,24 @@ def test_run_match_config_params() -> None:
         if key not in ("app_dir",)
     }
     assert config_params == run_params
+
+
+@pytest.mark.anyio
+async def test_run_multiprocess_with_sockets():
+    config = Config(app=app, workers=2, limit_max_requests=1)
+    with socket.socket() as sock:
+        sock.bind(("localhost", 0))
+        async with run_server(config, sockets=[sock]) as server:
+            while not server.started:
+                await asyncio.sleep(0.1)
+            await asyncio.sleep(0.1)
+
+
+@pytest.mark.anyio
+async def test_run_invalid_host() -> None:
+    with pytest.raises(SystemExit) as e:
+        config = Config(app=app, host="illegal_host")
+        server = Server(config=config)
+        await server.serve()
+    assert e.type == SystemExit
+    assert e.value.code == 1

uvicorn/server.py

@@ -102,7 +102,9 @@ async def startup(self, sockets: list = None) -> None:
             # Explicitly passed a list of open sockets.
             # We use this when the server is run from a Gunicorn worker.
 
-            def _share_socket(sock: socket.SocketType) -> socket.SocketType:
+            def _share_socket(
+                sock: socket.SocketType,
+            ) -> socket.SocketType:  # pragma py-linux pragma: py-darwin
                 # Windows requires the socket be explicitly shared across
                 # multiple workers (processes).
                 from socket import fromshare  # type: ignore
@@ -113,14 +115,14 @@ def _share_socket(sock: socket.SocketType) -> socket.SocketType:
             self.servers = []
             for sock in sockets:
                 if config.workers > 1 and platform.system() == "Windows":
-                    sock = _share_socket(sock)
+                    sock = _share_socket(sock)  # pragma py-linux pragma: py-darwin
                 server = await loop.create_server(
                     create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog
                 )
                 self.servers.append(server)
             listeners = sockets
 
-        elif config.fd is not None:
+        elif config.fd is not None:  # pragma: py-win32
             # Use an existing socket, from a file descriptor.
             sock = socket.fromfd(config.fd, socket.AF_UNIX, socket.SOCK_STREAM)
             server = await loop.create_server(
@@ -130,7 +132,7 @@ def _share_socket(sock: socket.SocketType) -> socket.SocketType:
             listeners = server.sockets
             self.servers = [server]
 
-        elif config.uds is not None:
+        elif config.uds is not None:  # pragma: py-win32
             # Create a socket using UNIX domain socket.
             uds_perms = 0o666
             if os.path.exists(config.uds):
@@ -174,14 +176,14 @@ def _share_socket(sock: socket.SocketType) -> socket.SocketType:
     def _log_started_message(self, listeners: Sequence[socket.SocketType]) -> None:
         config = self.config
 
-        if config.fd is not None:
+        if config.fd is not None:  # pragma: py-win32
             sock = listeners[0]
             logger.info(
                 "Uvicorn running on socket %s (Press CTRL+C to quit)",
                 sock.getsockname(),
             )
 
-        elif config.uds is not None:
+        elif config.uds is not None:  # pragma: py-win32
             logger.info(
                 "Uvicorn running on unix socket %s (Press CTRL+C to quit)", config.uds
             )