mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-22 15:37:45 +00:00
update mypy
This commit is contained in:
parent
9176626dab
commit
7b0485d6d6
@ -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:
|
||||
|
@ -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)
|
||||
|
@ -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:
|
||||
|
Loading…
Reference in New Issue
Block a user