Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

improve logging test and increase coverage on server.py #1743

Merged
merged 24 commits into from Nov 2, 2022
Merged
Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
2 changes: 1 addition & 1 deletion setup.cfg
Expand Up @@ -82,7 +82,7 @@ plugins =

[coverage:report]
precision = 2
fail_under = 97.92
fail_under = 98.06
show_missing = true
skip_covered = true
exclude_lines =
Expand Down
56 changes: 56 additions & 0 deletions tests/middleware/test_logging.py
@@ -1,5 +1,8 @@
import asyncio
import contextlib
import logging
import socket
import sys

import httpx
import pytest
Expand Down Expand Up @@ -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()


iudeen marked this conversation as resolved.
Show resolved Hide resolved
@pytest.mark.anyio
async def test_unknown_status_code(caplog):
async def app(scope, receive, send):
Expand Down
8 changes: 4 additions & 4 deletions uvicorn/server.py
Expand Up @@ -120,7 +120,7 @@ def _share_socket(sock: socket.SocketType) -> socket.SocketType:
self.servers.append(server)
listeners = sockets

elif config.fd is not None:
elif config.fd is not None: # pragma: py-win32
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add # pragma: py-win32 because we use socket.AF_UNIX in all instances.

Copy link
Sponsor Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, you mean in the line after... Got it. Yep!

# Use an existing socket, from a file descriptor.
sock = socket.fromfd(config.fd, socket.AF_UNIX, socket.SOCK_STREAM)
server = await loop.create_server(
Expand All @@ -130,7 +130,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):
Expand Down Expand Up @@ -174,14 +174,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
)
Expand Down