# is the lockfile, and maint/requirements.in is the input file containing only
# direct dependencies.
-black==19.10b0
+black
flake8
mypy==0.740
tox
# is the lockfile, and maint/requirements.in is the input file containing only
# direct dependencies.
-black==19.10b0
+black==20.8b1
flake8==3.8.4
mypy==0.740
tox==3.20.1
class HTTP1ConnectionParameters(object):
- """Parameters for `.HTTP1Connection` and `.HTTP1ServerConnection`.
- """
+ """Parameters for `.HTTP1Connection` and `.HTTP1ServerConnection`."""
def __init__(
self,
class _GzipMessageDelegate(httputil.HTTPMessageDelegate):
- """Wraps an `HTTPMessageDelegate` to decode ``Content-Encoding: gzip``.
- """
+ """Wraps an `HTTPMessageDelegate` to decode ``Content-Encoding: gzip``."""
def __init__(self, delegate: httputil.HTTPMessageDelegate, chunk_size: int) -> None:
self._delegate = delegate
raise NotImplementedError()
def finish(self) -> None:
- """Indicates that the last body data has been written.
- """
+ """Indicates that the last body data has been written."""
raise NotImplementedError()
class StreamBufferFullError(Exception):
- """Exception raised by `IOStream` methods when the buffer is full.
- """
+ """Exception raised by `IOStream` methods when the buffer is full."""
class _StreamBuffer(object):
class _ReleasingContextManager(object):
"""Releases a Lock or Semaphore at the end of a "with" statement.
- with (yield semaphore.acquire()):
- pass
+ with (yield semaphore.acquire()):
+ pass
- # Now semaphore.release() has been called.
+ # Now semaphore.release() has been called.
"""
def __init__(self, obj: Any) -> None:
) # type: Optional[Tuple[List[_FileDescriptorLike], List[_FileDescriptorLike]]]
self._closing_selector = False
self._thread = threading.Thread(
- name="Tornado selector", daemon=True, target=self._run_select,
+ name="Tornado selector",
+ daemon=True,
+ target=self._run_select,
)
self._thread.start()
# Start the select loop once the loop is started.
self._start_select()
def _handle_event(
- self, fd: "_FileDescriptorLike", cb_map: Dict["_FileDescriptorLike", Callable],
+ self,
+ fd: "_FileDescriptorLike",
+ cb_map: Dict["_FileDescriptorLike", Callable],
) -> None:
try:
callback = cb_map[fd]
class Loader(BaseLoader):
- """A template loader that loads from a single root directory.
- """
+ """A template loader that loads from a single root directory."""
def __init__(self, root_directory: str, **kwargs: Any) -> None:
super().__init__(**kwargs)
yield self.client.connect("127.0.0.1", port)
def test_source_ip_fail(self):
- """Fail when trying to use the source IP Address '8.8.8.8'.
- """
+ """Fail when trying to use the source IP Address '8.8.8.8'."""
self.assertRaises(
socket.error,
self.do_test_connect,
)
def test_source_ip_success(self):
- """Success when trying to use the source IP Address '127.0.0.1'.
- """
+ """Success when trying to use the source IP Address '127.0.0.1'."""
self.do_test_connect(socket.AF_INET, "127.0.0.1", source_ip="127.0.0.1")
@skipIfNonUnix
def test_source_port_fail(self):
- """Fail when trying to use source port 1.
- """
+ """Fail when trying to use source port 1."""
if getpass.getuser() == "root":
# Root can use any port so we can't easily force this to fail.
# This is mainly relevant for docker.
)
# If XSRF cookies are turned on, reject form submissions without
# the proper cookie
- if self.request.method not in (
- "GET",
- "HEAD",
- "OPTIONS",
- ) and self.application.settings.get("xsrf_cookies"):
+ if (
+ self.request.method
+ not in (
+ "GET",
+ "HEAD",
+ "OPTIONS",
+ )
+ and self.application.settings.get("xsrf_cookies")
+ ):
self.check_xsrf_cookie()
result = self.prepare()
class WebSocketProtocol(abc.ABC):
- """Base class for WebSocket protocol versions.
- """
+ """Base class for WebSocket protocol versions."""
def __init__(self, handler: "_WebSocketDelegate") -> None:
self.handler = handler
@staticmethod
def environ(request: httputil.HTTPServerRequest) -> Dict[Text, Any]:
- """Converts a `tornado.httputil.HTTPServerRequest` to a WSGI environment.
- """
+ """Converts a `tornado.httputil.HTTPServerRequest` to a WSGI environment."""
hostport = request.host.split(":")
if len(hostport) == 2:
host = hostport[0]