Add type hints to dummyserver
MODIFY
dummyserver/handlers.py
dummyserver/handlers.py
@@ -8,6 +8,7 @@ import zlib
from datetime import datetime, timedelta
from http.client import responses
from io import BytesIO
+from typing import Dict, Optional, Sequence, Tuple, Union
from urllib.parse import urlsplit
from tornado import httputil
@@ -19,33 +20,43 @@ log = logging.getLogger(__name__)
class Response:
- def __init__(self, body="", status="200 OK", headers=None):
+ def __init__(
+ self,
+ body: Union[str, bytes, Sequence[Union[str, bytes]]] = "",
+ status: str = "200 OK",
+ headers: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None,
+ ) -> None:
self.body = body
self.status = status
self.headers = headers or [("Content-type", "text/plain")]
- def __call__(self, request_handler):
+ def __call__(self, request_handler: RequestHandler) -> None:
status, reason = self.status.split(" ", 1)
request_handler.set_status(int(status), reason)
for header, value in self.headers:
request_handler.add_header(header, value)
+ if isinstance(self.body, str):
+ request_handler.write(self.body.encode())
+ elif isinstance(self.body, bytes):
+ request_handler.write(self.body)
# chunked
- if isinstance(self.body, list):
+ else:
for item in self.body:
if not isinstance(item, bytes):
item = item.encode("utf8")
request_handler.write(item)
request_handler.flush()
- else:
- body = self.body
- if not isinstance(body, bytes):
- body = body.encode("utf8")
- request_handler.write(body)
+
+RETRY_TEST_NAMES: Dict[str, int] = collections.defaultdict(int)
-RETRY_TEST_NAMES = collections.defaultdict(int)
+def request_params(request: httputil.HTTPServerRequest) -> Dict[str, bytes]:
+ params = {}
+ for k, v in request.arguments.items():
+ params[k] = next(iter(v))
+ return params
class TestingApp(RequestHandler):
@@ -58,32 +69,29 @@ class TestingApp(RequestHandler):
method has its own conditions for success/failure.
"""
- def get(self):
+ def get(self) -> None:
"""Handle GET requests"""
self._call_method()
- def post(self):
+ def post(self) -> None:
"""Handle POST requests"""
self._call_method()
- def put(self):
+ def put(self) -> None:
"""Handle PUT requests"""
self._call_method()
- def options(self):
+ def options(self) -> None:
"""Handle OPTIONS requests"""
self._call_method()
- def head(self):
+ def head(self) -> None:
"""Handle HEAD requests"""
self._call_method()
- def _call_method(self):
+ def _call_method(self) -> None:
"""Call the correct method in this class based on the incoming URI"""
req = self.request
- req.params = {}
- for k, v in req.arguments.items():
- req.params[k] = next(iter(v))
path = req.path[:]
if not path.startswith("/"):
@@ -100,56 +108,61 @@ class TestingApp(RequestHandler):
resp(self)
- def index(self, _request):
+ def index(self, _request: httputil.HTTPServerRequest) -> Response:
"Render simple message"
return Response("Dummy server!")
- def certificate(self, request):
+ def certificate(self, request: httputil.HTTPServerRequest) -> Response:
"""Return the requester's certificate."""
cert = request.get_ssl_certificate()
+ assert isinstance(cert, dict)
subject = {}
if cert is not None:
subject = {k: v for (k, v) in [y for z in cert["subject"] for y in z]}
return Response(json.dumps(subject))
- def alpn_protocol(self, request):
+ def alpn_protocol(self, request: httputil.HTTPServerRequest) -> Response:
"""Return the selected ALPN protocol."""
- proto = request.connection.stream.socket.selected_alpn_protocol()
+ assert request.connection is not None
+ proto = request.connection.stream.socket.selected_alpn_protocol() # type: ignore[attr-defined]
return Response(proto.encode("utf8") if proto is not None else "")
- def source_address(self, request):
+ def source_address(self, request: httputil.HTTPServerRequest) -> Response:
"""Return the requester's IP address."""
return Response(request.remote_ip)
- def set_up(self, request):
- test_type = request.params.get("test_type")
- test_id = request.params.get("test_id")
+ def set_up(self, request: httputil.HTTPServerRequest) -> Response:
+ params = request_params(request)
+ test_type = params.get("test_type")
+ test_id = params.get("test_id")
if test_id:
- print(f"\nNew test {test_type}: {test_id}")
+ print(f"\nNew test {test_type!r}: {test_id!r}")
else:
- print(f"\nNew test {test_type}")
+ print(f"\nNew test {test_type!r}")
return Response("Dummy server is ready!")
- def specific_method(self, request):
+ def specific_method(self, request: httputil.HTTPServerRequest) -> Response:
"Confirm that the request matches the desired method type"
- method = request.params.get("method")
- if method and not isinstance(method, str):
- method = method.decode("utf8")
+ params = request_params(request)
+ method = params.get("method")
+ method_str = method.decode() if method else None
- if request.method != method:
+ if request.method != method_str:
return Response(
- f"Wrong method: {method} != {request.method}",
+ f"Wrong method: {method_str} != {request.method}",
status="400 Bad Request",
)
return Response()
- def upload(self, request):
+ def upload(self, request: httputil.HTTPServerRequest) -> Response:
"Confirm that the uploaded file conforms to specification"
+ params = request_params(request)
# FIXME: This is a huge broken mess
- param = request.params.get("upload_param", b"myfile").decode("ascii")
- filename = request.params.get("upload_filename", b"").decode("utf-8")
- size = int(request.params.get("upload_size", "0"))
+ param = params.get("upload_param", b"myfile").decode("ascii")
+ filename = params.get("upload_filename", b"").decode("utf-8")
+ size = int(params.get("upload_size", "0"))
files_ = request.files.get(param)
+ assert files_ is not None
if len(files_) != 1:
return Response(
@@ -177,23 +190,26 @@ class TestingApp(RequestHandler):
return Response()
- def redirect(self, request):
+ def redirect(self, request: httputil.HTTPServerRequest) -> Response: # type: ignore[override]
"Perform a redirect to ``target``"
- target = request.params.get("target", "/")
- status = request.params.get("status", "303 See Other")
+ params = request_params(request)
+ target = params.get("target", "/")
+ status = params.get("status", b"303 See Other").decode("latin-1")
if len(status) == 3:
- status = f"{status.decode('latin-1')} Redirect"
+ status = f"{status} Redirect"
headers = [("Location", target)]
return Response(status=status, headers=headers)
- def not_found(self, request):
+ def not_found(self, request: httputil.HTTPServerRequest) -> Response:
return Response("Not found", status="404 Not Found")
- def multi_redirect(self, request):
+ def multi_redirect(self, request: httputil.HTTPServerRequest) -> Response:
"Performs a redirect chain based on ``redirect_codes``"
- codes = request.params.get("redirect_codes", b"200").decode("utf-8")
+ params = request_params(request)
+ codes = params.get("redirect_codes", b"200").decode("utf-8")
head, tail = codes.split(",", 1) if "," in codes else (codes, None)
+ assert head is not None
status = f"{head} {responses[int(head)]}"
if not tail:
return Response("Done redirecting", status=status)
@@ -201,30 +217,33 @@ class TestingApp(RequestHandler):
headers = [("Location", f"/multi_redirect?redirect_codes={tail}")]
return Response(status=status, headers=headers)
- def keepalive(self, request):
- if request.params.get("close", b"0") == b"1":
+ def keepalive(self, request: httputil.HTTPServerRequest) -> Response:
+ params = request_params(request)
+ if params.get("close", b"0") == b"1":
headers = [("Connection", "close")]
return Response("Closing", headers=headers)
headers = [("Connection", "keep-alive")]
return Response("Keeping alive", headers=headers)
- def echo_params(self, request):
- params = sorted([(to_str(k), to_str(v)) for k, v in request.params.items()])
- return Response(repr(params))
+ def echo_params(self, request: httputil.HTTPServerRequest) -> Response:
+ params = request_params(request)
+ echod = sorted((to_str(k), to_str(v)) for k, v in params.items())
+ return Response(repr(echod))
- def echo(self, request):
+ def echo(self, request: httputil.HTTPServerRequest) -> Response:
"Echo back the params"
if request.method == "GET":
return Response(request.query)
return Response(request.body)
- def echo_uri(self, request):
+ def echo_uri(self, request: httputil.HTTPServerRequest) -> Response:
"Echo back the requested URI"
+ assert request.uri is not None
return Response(request.uri)
- def encodingrequest(self, request):
+ def encodingrequest(self, request: httputil.HTTPServerRequest) -> Response:
"Check for UA accepting gzip/deflate encoding"
data = b"hello, world!"
encoding = request.headers.get("Accept-Encoding", "")
@@ -242,16 +261,16 @@ class TestingApp(RequestHandler):
data = zlib.compress(data)
elif encoding == "garbage-gzip":
headers = [("Content-Encoding", "gzip")]
- data = "garbage"
+ data = b"garbage"
elif encoding == "garbage-deflate":
headers = [("Content-Encoding", "deflate")]
- data = "garbage"
+ data = b"garbage"
return Response(data, headers=headers)
- def headers(self, request):
+ def headers(self, request: httputil.HTTPServerRequest) -> Response:
return Response(json.dumps(dict(request.headers)))
- def successful_retry(self, request):
+ def successful_retry(self, request: httputil.HTTPServerRequest) -> Response:
"""Handler which will return an error and then success
It's not currently very flexible as the number of retries is hard-coded.
@@ -267,10 +286,10 @@ class TestingApp(RequestHandler):
else:
return Response("need to keep retrying!", status="418 I'm A Teapot")
- def chunked(self, request):
+ def chunked(self, request: httputil.HTTPServerRequest) -> Response:
return Response(["123"] * 4)
- def chunked_gzip(self, request):
+ def chunked_gzip(self, request: httputil.HTTPServerRequest) -> Response:
chunks = []
compressor = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS)
@@ -281,39 +300,43 @@ class TestingApp(RequestHandler):
return Response(chunks, headers=[("Content-Encoding", "gzip")])
- def nbytes(self, request):
- length = int(request.params.get("length"))
+ def nbytes(self, request: httputil.HTTPServerRequest) -> Response:
+ params = request_params(request)
+ length = int(params["length"])
data = b"1" * length
return Response(data, headers=[("Content-Type", "application/octet-stream")])
- def status(self, request):
- status = request.params.get("status", "200 OK")
+ def status(self, request: httputil.HTTPServerRequest) -> Response:
+ params = request_params(request)
+ status = params.get("status", b"200 OK").decode("latin-1")
return Response(status=status)
- def retry_after(self, request):
- if datetime.now() - self.application.last_req < timedelta(seconds=1):
- status = request.params.get("status", b"429 Too Many Requests")
+ def retry_after(self, request: httputil.HTTPServerRequest) -> Response:
+ params = request_params(request)
+ if datetime.now() - self.application.last_req < timedelta(seconds=1): # type: ignore[attr-defined]
+ status = params.get("status", b"429 Too Many Requests")
return Response(
status=status.decode("utf-8"), headers=[("Retry-After", "1")]
)
- self.application.last_req = datetime.now()
+ self.application.last_req = datetime.now() # type: ignore[attr-defined]
return Response(status="200 OK")
- def redirect_after(self, request):
+ def redirect_after(self, request: httputil.HTTPServerRequest) -> Response:
"Perform a redirect to ``target``"
- date = request.params.get("date")
+ params = request_params(request)
+ date = params.get("date")
if date:
retry_after = str(
httputil.format_timestamp(datetime.utcfromtimestamp(float(date)))
)
else:
retry_after = "1"
- target = request.params.get("target", "/")
+ target = params.get("target", "/")
headers = [("Location", target), ("Retry-After", retry_after)]
return Response(status="303 See Other", headers=headers)
- def shutdown(self, request):
+ def shutdown(self, request: httputil.HTTPServerRequest) -> None:
sys.exit()

MODIFY
dummyserver/proxy.py
dummyserver/proxy.py
@@ -40,10 +40,10 @@ __all__ = ["ProxyHandler", "run_proxy"]
class ProxyHandler(tornado.web.RequestHandler):
- SUPPORTED_METHODS = ["GET", "POST", "CONNECT"]
+ SUPPORTED_METHODS = ["GET", "POST", "CONNECT"] # type: ignore[assignment]
- async def get(self):
- async def handle_response(response):
+ async def get(self) -> None:
+ async def handle_response(response: tornado.httpclient.HTTPResponse) -> None:
if response.error and not isinstance(
response.error, tornado.httpclient.HTTPError
):
@@ -72,6 +72,8 @@ class ProxyHandler(tornado.web.RequestHandler):
if upstream_ca_certs:
ssl_options = ssl.create_default_context(cafile=upstream_ca_certs)
+ assert self.request.uri is not None
+ assert self.request.method is not None
req = tornado.httpclient.HTTPRequest(
url=self.request.uri,
method=self.request.method,
@@ -94,14 +96,18 @@ class ProxyHandler(tornado.web.RequestHandler):
self.write("Internal server error:\n" + str(e))
self.finish()
- async def post(self):
+ async def post(self) -> None:
await self.get()
- async def connect(self):
+ async def connect(self) -> None:
+ assert self.request.uri is not None
host, port = self.request.uri.split(":")
- client = self.request.connection.stream
+ assert self.request.connection is not None
+ client: tornado.iostream.IOStream = self.request.connection.stream # type: ignore[attr-defined]
- async def start_forward(reader, writer):
+ async def start_forward(
+ reader: tornado.iostream.IOStream, writer: tornado.iostream.IOStream
+ ) -> None:
while True:
try:
data = await reader.read_bytes(4096, partial=True)
@@ -122,7 +128,7 @@ class ProxyHandler(tornado.web.RequestHandler):
await tornado.gen.multi([fu1, fu2])
-def run_proxy(port, start_ioloop=True):
+def run_proxy(port: int, start_ioloop: bool = True) -> None:
"""
Run proxy on the specified port. If start_ioloop is True (default),
the tornado IOLoop will be started immediately.

MODIFY
dummyserver/server.py
dummyserver/server.py
@@ -12,12 +12,13 @@ import sys
import threading
import warnings
from datetime import datetime
+from typing import Callable, Optional, Tuple, Union
import tornado.httpserver
import tornado.ioloop
import tornado.netutil
import tornado.web
-import trustme
+import trustme # type: ignore[import]
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
@@ -38,7 +39,7 @@ DEFAULT_CA = os.path.join(CERTS_PATH, "cacert.pem")
DEFAULT_CA_KEY = os.path.join(CERTS_PATH, "cacert.key")
-def _resolves_to_ipv6(host):
+def _resolves_to_ipv6(host: str) -> bool:
"""Returns True if the system resolves host to an IPv6 address by default."""
resolves_to_ipv6 = False
try:
@@ -52,7 +53,7 @@ def _resolves_to_ipv6(host):
return resolves_to_ipv6
-def _has_ipv6(host):
+def _has_ipv6(host: str) -> bool:
"""Returns True if the system can bind an IPv6 address."""
sock = None
has_ipv6 = False
@@ -101,7 +102,12 @@ class SocketServerThread(threading.Thread):
USE_IPV6 = HAS_IPV6_AND_DNS
- def __init__(self, socket_handler, host="localhost", ready_event=None):
+ def __init__(
+ self,
+ socket_handler: Callable[[socket.socket], None],
+ host: str = "localhost",
+ ready_event: Optional[threading.Event] = None,
+ ) -> None:
super().__init__()
self.daemon = True
@@ -109,7 +115,7 @@ class SocketServerThread(threading.Thread):
self.host = host
self.ready_event = ready_event
- def _start_server(self):
+ def _start_server(self) -> None:
if self.USE_IPV6:
sock = socket.socket(socket.AF_INET6)
else:
@@ -129,22 +135,22 @@ class SocketServerThread(threading.Thread):
self.socket_handler(sock)
sock.close()
- def run(self):
- self.server = self._start_server()
+ def run(self) -> None:
+ self._start_server()
-def ssl_options_to_context(
+def ssl_options_to_context( # type: ignore[no-untyped-def]
keyfile=None,
certfile=None,
server_side=None,
cert_reqs=None,
- ssl_version=None,
+ ssl_version: Optional[Union[str, int]] = None,
ca_certs=None,
do_handshake_on_connect=None,
suppress_ragged_eofs=None,
ciphers=None,
alpn_protocols=None,
-):
+) -> ssl.SSLContext:
"""Return an equivalent SSLContext based on ssl.wrap_socket args."""
ssl_version = resolve_ssl_version(ssl_version)
cert_none = resolve_cert_reqs("CERT_NONE")
@@ -166,12 +172,18 @@ def ssl_options_to_context(
return ctx
-def run_tornado_app(app, io_loop, certs, scheme, host):
+def run_tornado_app( # type: ignore[no-untyped-def]
+ app: tornado.web.Application,
+ io_loop: tornado.ioloop.IOLoop,
+ certs,
+ scheme: str,
+ host: str,
+) -> Tuple[tornado.httpserver.HTTPServer, int]:
assert io_loop == tornado.ioloop.IOLoop.current()
# We can't use fromtimestamp(0) because of CPython issue 29097, so we'll
# just construct the datetime object directly.
- app.last_req = datetime(1970, 1, 1)
+ app.last_req = datetime(1970, 1, 1) # type: ignore[attr-defined]
if scheme == "https":
ssl_opts = ssl_options_to_context(**certs)
@@ -179,26 +191,26 @@ def run_tornado_app(app, io_loop, certs, scheme, host):
else:
http_server = tornado.httpserver.HTTPServer(app)
- sockets = tornado.netutil.bind_sockets(None, address=host)
+ sockets = tornado.netutil.bind_sockets(None, address=host) # type: ignore[arg-type]
port = sockets[0].getsockname()[1]
http_server.add_sockets(sockets)
return http_server, port
-def run_loop_in_thread(io_loop):
+def run_loop_in_thread(io_loop: tornado.ioloop.IOLoop) -> threading.Thread:
t = threading.Thread(target=io_loop.start)
t.start()
return t
-def get_unreachable_address():
+def get_unreachable_address() -> Tuple[str, int]:
# reserved as per rfc2606
return ("something.invalid", 54321)
if __name__ == "__main__":
# For debugging dummyserver itself - python -m dummyserver.server
- from .testcase import TestingApp
+ from .handlers import TestingApp
host = "127.0.0.1"
@@ -210,9 +222,9 @@ if __name__ == "__main__":
print(f"Listening on http://{host}:{port}")
-def encrypt_key_pem(private_key_pem, password):
+def encrypt_key_pem(private_key_pem: trustme.Blob, password: bytes) -> trustme.Blob:
private_key = serialization.load_pem_private_key(
- private_key_pem.bytes(), password=None, backend=default_backend()
+ private_key_pem.bytes(), password=None, backend=default_backend() # type: ignore[no-untyped-call]
)
encrypted_key = private_key.private_bytes(
serialization.Encoding.PEM,

MODIFY
dummyserver/testcase.py
dummyserver/testcase.py
+import socket
import threading
from contextlib import contextmanager
+from typing import Any, Callable, ClassVar, Generator, Iterable, Optional
import pytest
-from tornado import ioloop, web
+from tornado import httpserver, ioloop, web
from dummyserver.handlers import TestingApp
from dummyserver.proxy import ProxyHandler
@@ -16,7 +18,7 @@ from dummyserver.server import (
from urllib3.connection import HTTPConnection
-def consume_socket(sock, chunks=65536):
+def consume_socket(sock: socket.socket, chunks: int = 65536) -> bytearray:
consumed = bytearray()
while True:
b = sock.recv(chunks)
@@ -35,8 +37,11 @@ class SocketDummyServerTestCase:
scheme = "http"
host = "localhost"
+ server_thread: ClassVar[SocketServerThread]
+ port: ClassVar[int]
+
@classmethod
- def _start_server(cls, socket_handler):
+ def _start_server(cls, socket_handler: Callable[[socket.socket], None]) -> None:
ready_event = threading.Event()
cls.server_thread = SocketServerThread(
socket_handler=socket_handler, ready_event=ready_event, host=cls.host
@@ -48,10 +53,12 @@ class SocketDummyServerTestCase:
cls.port = cls.server_thread.port
@classmethod
- def start_response_handler(cls, response, num=1, block_send=None):
+ def start_response_handler(
+ cls, response: bytes, num: int = 1, block_send: Optional[threading.Event] = None
+ ) -> threading.Event:
ready_event = threading.Event()
- def socket_handler(listener):
+ def socket_handler(listener: socket.socket) -> None:
for _ in range(num):
ready_event.set()
@@ -67,34 +74,43 @@ class SocketDummyServerTestCase:
return ready_event
@classmethod
- def start_basic_handler(cls, **kw):
+ def start_basic_handler(
+ cls, num: int = 1, block_send: Optional[threading.Event] = None
+ ) -> threading.Event:
return cls.start_response_handler(
- b"HTTP/1.1 200 OK\r\n" b"Content-Length: 0\r\n" b"\r\n", **kw
+ b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n",
+ num,
+ block_send,
)
@classmethod
- def teardown_class(cls):
+ def teardown_class(cls) -> None:
if hasattr(cls, "server_thread"):
cls.server_thread.join(0.1)
def assert_header_received(
- self, received_headers, header_name, expected_value=None
- ):
- header_name = header_name.encode("ascii")
- if expected_value is not None:
- expected_value = expected_value.encode("ascii")
+ self,
+ received_headers: Iterable[bytes],
+ header_name: str,
+ expected_value: Optional[str] = None,
+ ) -> None:
+ header_name_bytes = header_name.encode("ascii")
+ if expected_value is None:
+ expected_value_bytes = None
+ else:
+ expected_value_bytes = expected_value.encode("ascii")
header_titles = []
for header in received_headers:
key, value = header.split(b": ")
header_titles.append(key)
- if key == header_name and expected_value is not None:
- assert value == expected_value
- assert header_name in header_titles
+ if key == header_name_bytes and expected_value_bytes is not None:
+ assert value == expected_value_bytes
+ assert header_name_bytes in header_titles
class IPV4SocketDummyServerTestCase(SocketDummyServerTestCase):
@classmethod
- def _start_server(cls, socket_handler):
+ def _start_server(cls, socket_handler: Callable[[socket.socket], None]) -> None:
ready_event = threading.Event()
cls.server_thread = SocketServerThread(
socket_handler=socket_handler, ready_event=ready_event, host=cls.host
@@ -121,8 +137,13 @@ class HTTPDummyServerTestCase:
host_alt = "127.0.0.1" # Some tests need two hosts
certs = DEFAULT_CERTS
+ io_loop: ClassVar[ioloop.IOLoop]
+ server: ClassVar[httpserver.HTTPServer]
+ port: ClassVar[int]
+ server_thread: ClassVar[threading.Thread]
+
@classmethod
- def _start_server(cls):
+ def _start_server(cls) -> None:
cls.io_loop = ioloop.IOLoop.current()
app = web.Application([(r".*", TestingApp)])
cls.server, cls.port = run_tornado_app(
@@ -131,17 +152,17 @@ class HTTPDummyServerTestCase:
cls.server_thread = run_loop_in_thread(cls.io_loop)
@classmethod
- def _stop_server(cls):
+ def _stop_server(cls) -> None:
cls.io_loop.add_callback(cls.server.stop)
cls.io_loop.add_callback(cls.io_loop.stop)
cls.server_thread.join()
@classmethod
- def setup_class(cls):
+ def setup_class(cls) -> None:
cls._start_server()
@classmethod
- def teardown_class(cls):
+ def teardown_class(cls) -> None:
cls._stop_server()
@@ -152,19 +173,30 @@ class HTTPSDummyServerTestCase(HTTPDummyServerTestCase):
class HTTPDummyProxyTestCase:
+ io_loop: ClassVar[ioloop.IOLoop]
http_host = "localhost"
http_host_alt = "127.0.0.1"
+ http_server: ClassVar[httpserver.HTTPServer]
+ http_port: ClassVar[int]
https_host = "localhost"
https_host_alt = "127.0.0.1"
https_certs = DEFAULT_CERTS
+ https_server: ClassVar[httpserver.HTTPServer]
+ https_port: ClassVar[int]
proxy_host = "localhost"
proxy_host_alt = "127.0.0.1"
+ proxy_server: ClassVar[httpserver.HTTPServer]
+ proxy_port: ClassVar[int]
+ https_proxy_server: ClassVar[httpserver.HTTPServer]
+ https_proxy_port: ClassVar[int]
+
+ server_thread: ClassVar[threading.Thread]
@classmethod
- def setup_class(cls):
+ def setup_class(cls) -> None:
cls.io_loop = ioloop.IOLoop.current()
app = web.Application([(r".*", TestingApp)])
@@ -193,7 +225,7 @@ class HTTPDummyProxyTestCase:
cls.server_thread = run_loop_in_thread(cls.io_loop)
@classmethod
- def teardown_class(cls):
+ def teardown_class(cls) -> None:
cls.io_loop.add_callback(cls.http_server.stop)
cls.io_loop.add_callback(cls.https_server.stop)
cls.io_loop.add_callback(cls.proxy_server.stop)
@@ -233,7 +265,7 @@ class ConnectionMarker:
@classmethod
@contextmanager
- def mark(cls, monkeypatch):
+ def mark(cls, monkeypatch: Any) -> Generator[None, None, None]:
"""
Mark connections under in that context.
"""
@@ -241,11 +273,10 @@ class ConnectionMarker:
orig_request = HTTPConnection.request
orig_request_chunked = HTTPConnection.request_chunked
- def call_and_mark(target):
- def part(self, *args, **kwargs):
- result = target(self, *args, **kwargs)
+ def call_and_mark(target: Callable[..., None]) -> Callable[..., None]:
+ def part(self: HTTPConnection, *args: Any, **kwargs: Any) -> None:
+ target(self, *args, **kwargs)
self.sock.sendall(cls._get_socket_mark(self.sock, False))
- return result
return part
@@ -257,7 +288,7 @@ class ConnectionMarker:
yield
@classmethod
- def consume_request(cls, sock, chunks=65536):
+ def consume_request(cls, sock: socket.socket, chunks: int = 65536) -> bytearray:
"""
Consume a socket until after the HTTP request is sent.
"""
@@ -273,7 +304,7 @@ class ConnectionMarker:
return consumed
@classmethod
- def _get_socket_mark(cls, sock, server):
+ def _get_socket_mark(cls, sock: socket.socket, server: bool) -> bytes:
if server:
port = sock.getpeername()[1]
else:

MODIFY
noxfile.py
noxfile.py
@@ -155,8 +155,10 @@ def mypy(session):
session.install("mypy==0.910")
session.install("idna>=2.0.0")
session.install("cryptography>=1.3.4")
+ session.install("tornado>=6.1")
+ session.install("pytest>=6.2")
session.run("mypy", "--version")
- session.run("mypy", "src/urllib3")
+ session.run("mypy", "src/urllib3", "dummyserver")
@nox.session

Add type hints to dummyserver