Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-23 00:01:36 +00:00
Code Issues Projects Releases Wiki Activity

console: convert add_event to a signal.

Browse Source
This commit is contained in:
Aldo Cortesi 2015-06-02 12:09:07 +12:00
parent 776f0a9669
commit 57a61ae8fd
7 changed files with 36 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
libmproxy/console/__init__.py
View File

@ -220,11 +220,29 @@ class ConsoleMaster(flow.FlowMaster):
signals.call_in.connect(self.sig_call_in)
signals.pop_view_state.connect(self.sig_pop_view_state)
signals.push_view_state.connect(self.sig_push_view_state)
signals.sig_add_event.connect(self.sig_add_event)
def __setattr__(self, name, value):
self.__dict__[name] = value
signals.update_settings.send(self)
def sig_add_event(self, sender, e, level):
needed = dict(error=0, info=1, debug=2).get(level, 1)
if self.options.verbosity < needed:
return
if level == "error":
e = urwid.Text(("error", str(e)))
else:
e = urwid.Text(str(e))
self.eventlist.append(e)
if len(self.eventlist) > EVENTLOG_SIZE:
self.eventlist.pop(0)
self.eventlist.set_focus(len(self.eventlist) - 1)
def add_event(self, e, level):
signals.add_event(e, level)
def sig_call_in(self, sender, seconds, callback, args=()):
def cb(*_):
return callback(*args)
@ -263,16 +281,16 @@ class ConsoleMaster(flow.FlowMaster):
status, val = s.run(method, f)
if val:
if status:
self.add_event("Method %s return: %s" % (method, val), "debug")
signals.add_event("Method %s return: %s" % (method, val), "debug")
else:
self.add_event(
signals.add_event(
"Method %s error: %s" %
(method, val[1]), "error")
def run_script_once(self, command, f):
if not command:
return
self.add_event("Running script on flow: %s" % command, "debug")
signals.add_event("Running script on flow: %s" % command, "debug")
try:
s = script.Script(command, self)
@ -280,7 +298,7 @@ class ConsoleMaster(flow.FlowMaster):
signals.status_message.send(
message = "Error loading script."
)
self.add_event("Error loading script:\n%s" % v.args[0], "error")
signals.add_event("Error loading script:\n%s" % v.args[0], "error")
return
if f.request:
@ -432,7 +450,7 @@ class ConsoleMaster(flow.FlowMaster):
if self.options.rfile:
ret = self.load_flows_path(self.options.rfile)
if ret and self.state.flow_count():
self.add_event(
signals.add_event(
"File truncated or corrupted. "
"Loaded as many flows as possible.",
"error"
@ -666,20 +684,6 @@ class ConsoleMaster(flow.FlowMaster):
def clear_events(self):
self.eventlist[:] = []
def add_event(self, e, level="info"):
needed = dict(error=0, info=1, debug=2).get(level, 1)
if self.options.verbosity < needed:
return
if level == "error":
e = urwid.Text(("error", str(e)))
else:
e = urwid.Text(str(e))
self.eventlist.append(e)
if len(self.eventlist) > EVENTLOG_SIZE:
self.eventlist.pop(0)
self.eventlist.set_focus(len(self.eventlist) - 1)
# Handlers
def handle_error(self, f):
f = flow.FlowMaster.handle_error(self, f)

1
libmproxy/console/common.py
View File

@ -285,7 +285,6 @@ def copy_flow(part, scope, flow, master, state):
return
try:
master.add_event(str(len(data)))
pyperclip.copy(data)
except (RuntimeError, UnicodeDecodeError, AttributeError):
def save(k):

6
libmproxy/console/contentview.py
View File

@ -13,7 +13,7 @@ import urwid
import netlib.utils
from netlib import odict
from . import common
from . import common, signals
from .. import utils, encoding
from ..contrib import jsbeautifier, html2text
from ..contrib.wbxml.ASCommandResponse import ASCommandResponse
@ -507,7 +507,7 @@ def get(name):
return i
def get_content_view(viewmode, hdrItems, content, limit, logfunc, is_request):
def get_content_view(viewmode, hdrItems, content, limit, is_request):
"""
Returns a (msg, body) tuple.
"""
@ -532,7 +532,7 @@ def get_content_view(viewmode, hdrItems, content, limit, logfunc, is_request):
except Exception:
s = traceback.format_exc()
s = "Content viewer failed: \n" + s
logfunc(s, "error")
signals.add_event(s, "error")
ret = None
if not ret:
ret = get("Raw")(hdrs, content, limit)

1
libmproxy/console/flowview.py
View File

@ -182,7 +182,6 @@ class FlowView(tabs.Tabs):
tuple(tuple(i) for i in conn.headers.lst),
conn.content,
limit,
self.master.add_event,
isinstance(conn, HTTPRequest)
)
return (description, text_objects)

9
libmproxy/console/signals.py
View File

@ -1,5 +1,14 @@
import blinker
# Show a status message in the action bar
sig_add_event = blinker.Signal()
def add_event(e, level):
sig_add_event.send(
None,
e=e,
level=level
)
# Show a status message in the action bar
status_message = blinker.Signal()

1
libmproxy/console/tabs.py
View File

@ -1,4 +1,5 @@
import urwid
import signals
class Tabs(urwid.WidgetWrap):

6
test/test_console_contentview.py
View File

@ -190,7 +190,6 @@ Larry
[["content-type", "application/json"]],
"[1, 2, 3]",
1000,
lambda x, l: None,
False
)
assert "Raw" in r[0]
@ -200,7 +199,6 @@ Larry
[["content-type", "application/json"]],
"[1, 2, 3]",
1000,
lambda x, l: None,
False
)
assert r[0] == "JSON"
@ -210,7 +208,6 @@ Larry
[["content-type", "application/json"]],
"[1, 2",
1000,
lambda x, l: None,
False
)
assert "Raw" in r[0]
@ -220,7 +217,6 @@ Larry
[],
"[1, 2",
1000,
lambda x, l: None,
False
)
assert "Raw" in r[0]
@ -233,7 +229,6 @@ Larry
],
encoding.encode('gzip', "[1, 2, 3]"),
1000,
lambda x, l: None,
False
)
assert "decoded gzip" in r[0]
@ -247,7 +242,6 @@ Larry
],
encoding.encode('gzip', "[1, 2, 3]"),
1000,
lambda x, l: None,
False
)
assert "decoded gzip" in r[0]