mirror of
https://github.com/frappe/gunicorn.git
synced 2026-07-01 18:21:30 +08:00
95b7ffeeaa
- Bump version to 25.0.2 - Update copyright year to 2026 in LICENSE and NOTICE - Add license headers to all Python source files - Add changelog entry for 25.0.2
96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
#
|
|
# This file is part of gunicorn released under the MIT license.
|
|
# See the NOTICE for more information.
|
|
|
|
import datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from gunicorn.config import Config
|
|
from gunicorn.glogging import Logger
|
|
|
|
|
|
def test_atoms_defaults():
|
|
response = SimpleNamespace(
|
|
status='200', response_length=1024,
|
|
headers=(('Content-Type', 'application/json'),), sent=1024,
|
|
)
|
|
request = SimpleNamespace(headers=(('Accept', 'application/json'),))
|
|
environ = {
|
|
'REQUEST_METHOD': 'GET', 'RAW_URI': '/my/path?foo=bar',
|
|
'PATH_INFO': '/my/path', 'QUERY_STRING': 'foo=bar',
|
|
'SERVER_PROTOCOL': 'HTTP/1.1',
|
|
}
|
|
logger = Logger(Config())
|
|
atoms = logger.atoms(response, request, environ, datetime.timedelta(seconds=1))
|
|
assert isinstance(atoms, dict)
|
|
assert atoms['r'] == 'GET /my/path?foo=bar HTTP/1.1'
|
|
assert atoms['m'] == 'GET'
|
|
assert atoms['U'] == '/my/path'
|
|
assert atoms['q'] == 'foo=bar'
|
|
assert atoms['H'] == 'HTTP/1.1'
|
|
assert atoms['b'] == '1024'
|
|
assert atoms['B'] == 1024
|
|
assert atoms['{accept}i'] == 'application/json'
|
|
assert atoms['{content-type}o'] == 'application/json'
|
|
|
|
|
|
def test_atoms_zero_bytes():
|
|
response = SimpleNamespace(
|
|
status='200', response_length=0,
|
|
headers=(('Content-Type', 'application/json'),), sent=0,
|
|
)
|
|
request = SimpleNamespace(headers=(('Accept', 'application/json'),))
|
|
environ = {
|
|
'REQUEST_METHOD': 'GET', 'RAW_URI': '/my/path?foo=bar',
|
|
'PATH_INFO': '/my/path', 'QUERY_STRING': 'foo=bar',
|
|
'SERVER_PROTOCOL': 'HTTP/1.1',
|
|
}
|
|
logger = Logger(Config())
|
|
atoms = logger.atoms(response, request, environ, datetime.timedelta(seconds=1))
|
|
assert atoms['b'] == '0'
|
|
assert atoms['B'] == 0
|
|
|
|
|
|
@pytest.mark.parametrize('auth', [
|
|
# auth type is case in-sensitive
|
|
'Basic YnJrMHY6',
|
|
'basic YnJrMHY6',
|
|
'BASIC YnJrMHY6',
|
|
])
|
|
def test_get_username_from_basic_auth_header(auth):
|
|
request = SimpleNamespace(headers=())
|
|
response = SimpleNamespace(
|
|
status='200', response_length=1024, sent=1024,
|
|
headers=(('Content-Type', 'text/plain'),),
|
|
)
|
|
environ = {
|
|
'REQUEST_METHOD': 'GET', 'RAW_URI': '/my/path?foo=bar',
|
|
'PATH_INFO': '/my/path', 'QUERY_STRING': 'foo=bar',
|
|
'SERVER_PROTOCOL': 'HTTP/1.1',
|
|
'HTTP_AUTHORIZATION': auth,
|
|
}
|
|
logger = Logger(Config())
|
|
atoms = logger.atoms(response, request, environ, datetime.timedelta(seconds=1))
|
|
assert atoms['u'] == 'brk0v'
|
|
|
|
|
|
def test_get_username_handles_malformed_basic_auth_header():
|
|
"""Should catch a malformed auth header"""
|
|
request = SimpleNamespace(headers=())
|
|
response = SimpleNamespace(
|
|
status='200', response_length=1024, sent=1024,
|
|
headers=(('Content-Type', 'text/plain'),),
|
|
)
|
|
environ = {
|
|
'REQUEST_METHOD': 'GET', 'RAW_URI': '/my/path?foo=bar',
|
|
'PATH_INFO': '/my/path', 'QUERY_STRING': 'foo=bar',
|
|
'SERVER_PROTOCOL': 'HTTP/1.1',
|
|
'HTTP_AUTHORIZATION': 'Basic ixsTtkKzIpVTncfQjbBcnoRNoDfbnaXG',
|
|
}
|
|
logger = Logger(Config())
|
|
|
|
atoms = logger.atoms(response, request, environ, datetime.timedelta(seconds=1))
|
|
assert atoms['u'] == '-'
|