jingrow/gunicorn
1
0
Fork 0
mirror of https://github.com/frappe/gunicorn.git synced 2026-07-01 10:11:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

lint: use dict literal and hoist mock import

Browse Source
This commit is contained in:
Benoit Chesneau 2026-05-03 18:23:45 +02:00
parent e90b1c2c1e
commit 6f9ed30d23
2 changed files with 12 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
gunicorn/asgi/protocol.py
View File

@ -488,16 +488,16 @@ class ASGIProtocol(asyncio.Protocol):
# Create parser with callbacks and limit parameters (both parsers support them).
# Only the Python parser implements PROXY protocol framing; pass the option there.
parser_kwargs = dict(
on_headers_complete=self._on_headers_complete,
on_body=self._on_body,
on_message_complete=self._on_message_complete,
limit_request_line=limit_request_line,
limit_request_fields=self.cfg.limit_request_fields,
limit_request_field_size=self.cfg.limit_request_field_size,
permit_unconventional_http_method=self.cfg.permit_unconventional_http_method,
permit_unconventional_http_version=self.cfg.permit_unconventional_http_version,
)
parser_kwargs = {
'on_headers_complete': self._on_headers_complete,
'on_body': self._on_body,
'on_message_complete': self._on_message_complete,
'limit_request_line': limit_request_line,
'limit_request_fields': self.cfg.limit_request_fields,
'limit_request_field_size': self.cfg.limit_request_field_size,
'permit_unconventional_http_method': self.cfg.permit_unconventional_http_method,
'permit_unconventional_http_version': self.cfg.permit_unconventional_http_version,
}
if parser_class is PythonProtocol:
parser_kwargs['proxy_protocol'] = getattr(self.cfg, 'proxy_protocol', 'off')
self._callback_parser = parser_class(**parser_kwargs)

4
tests/test_http.py
View File

@ -297,13 +297,13 @@ def test_finish_body_returns_false_on_expired_deadline():
b"\r\n"
b"only-partial"
)
import time as _time
parser = _build_request_parser(payload)
# Force an already-elapsed deadline; the drain must abandon immediately.
import time as _time
expired = _time.monotonic() - 1.0
# IterUnreader has no socket; deadline path is exercised only when sock
# is present. Stub a sock with gettimeout/settimeout to drive the branch.
from unittest import mock
sock = mock.Mock()
sock.gettimeout.return_value = None
parser.unreader.sock = sock