diff --git a/mitmproxy/addons/core.py b/mitmproxy/addons/core.py index b0d8ee275..33d672790 100644 --- a/mitmproxy/addons/core.py +++ b/mitmproxy/addons/core.py @@ -112,7 +112,7 @@ class Core: val = sval # type: typing.Union[int, str] if spec == "status_code": try: - val = int(val) + val = int(val) # type: ignore except ValueError as v: raise exceptions.CommandError( "Status code is not an integer: %s" % val @@ -145,7 +145,7 @@ class Core: if spec == "status_code": resp.status_code = val if val in status_codes.RESPONSES: - resp.reason = status_codes.RESPONSES[int(val)] + resp.reason = status_codes.RESPONSES[val] # type: ignore elif spec == "reason": resp.reason = val else: diff --git a/mitmproxy/eventsequence.py b/mitmproxy/eventsequence.py index 68d97a23b..d263727b7 100644 --- a/mitmproxy/eventsequence.py +++ b/mitmproxy/eventsequence.py @@ -1,8 +1,8 @@ import typing from mitmproxy import controller -from mitmproxy import http from mitmproxy import flow +from mitmproxy import http from mitmproxy import tcp from mitmproxy import websocket @@ -40,8 +40,10 @@ Events = frozenset([ "update", ]) +TEventGenerator = typing.Iterator[typing.Tuple[str, typing.Any]] -def _iterate_http(f: http.HTTPFlow): + +def _iterate_http(f: http.HTTPFlow) -> TEventGenerator: if f.request: yield "requestheaders", f yield "request", f @@ -52,7 +54,7 @@ def _iterate_http(f: http.HTTPFlow): yield "error", f -def _iterate_websocket(f: websocket.WebSocketFlow): +def _iterate_websocket(f: websocket.WebSocketFlow) -> TEventGenerator: messages = f.messages f.messages = [] f.reply = controller.DummyReply() @@ -65,7 +67,7 @@ def _iterate_websocket(f: websocket.WebSocketFlow): yield "websocket_end", f -def _iterate_tcp(f: tcp.TCPFlow): +def _iterate_tcp(f: tcp.TCPFlow) -> TEventGenerator: messages = f.messages f.messages = [] f.reply = controller.DummyReply() @@ -78,19 +80,17 @@ def _iterate_tcp(f: tcp.TCPFlow): yield "tcp_end", f -TEventGenerator = typing.Iterator[typing.Tuple[str, typing.Any]] - _iterate_map = { http.HTTPFlow: _iterate_http, websocket.WebSocketFlow: _iterate_websocket, - tcp.TCPFlow: _iterate_tcp -} # type: typing.Dict[typing.Type[flow.Flow], typing.Callable[[flow.Flow], TEventGenerator]] + tcp.TCPFlow: _iterate_tcp, +} # type: typing.Dict[typing.Type[flow.Flow], typing.Callable[[typing.Any], TEventGenerator]] def iterate(f: flow.Flow) -> TEventGenerator: try: e = _iterate_map[type(f)] - except KeyError as e: - raise TypeError("Unknown flow type: {}".format(f)) from e + except KeyError as err: + raise TypeError("Unknown flow type: {}".format(f)) from err else: yield from e(f) diff --git a/mitmproxy/utils/human.py b/mitmproxy/utils/human.py index b3934846c..d67fb3104 100644 --- a/mitmproxy/utils/human.py +++ b/mitmproxy/utils/human.py @@ -71,7 +71,7 @@ def format_address(address: tuple) -> str: """ try: host = ipaddress.ip_address(address[0]) - if host.version == 4: + if isinstance(host, ipaddress.IPv4Address): return "{}:{}".format(str(host), address[1]) # If IPv6 is mapped to IPv4 elif host.ipv4_mapped: diff --git a/setup.py b/setup.py index 212ad95ea..441a1d8f9 100644 --- a/setup.py +++ b/setup.py @@ -90,7 +90,7 @@ setup( 'dev': [ "flake8>=3.2.1, <3.4", "Flask>=0.10.1, <0.13", - "mypy>=0.501, <0.512", + "mypy>=0.501, <0.521", "pytest-cov>=2.2.1, <3", "pytest-faulthandler>=1.3.0, <2", "pytest-timeout>=1.0.0, <2",