Get rid of tfile testing option

It's weird, it's ugly, it's getting in the way of my options refactoring, and
it must therefore die.
This commit is contained in:
Aldo Cortesi 2016-12-03 09:20:27 +13:00
parent 297493801d
commit b231836c70
6 changed files with 31 additions and 38 deletions

View File

@ -1,4 +1,5 @@
import itertools
import sys
import click
@ -25,10 +26,10 @@ def colorful(line, styles):
class Dumper:
def __init__(self):
def __init__(self, outfile=sys.stdout):
self.filter = None # type: flowfilter.TFilter
self.flow_detail = None # type: int
self.outfp = None # type: typing.io.TextIO
self.outfp = outfile # type: typing.io.TextIO
self.showhost = None # type: bool
self.default_contentview = "auto" # type: str
@ -43,7 +44,6 @@ class Dumper:
else:
self.filter = None
self.flow_detail = options.flow_detail
self.outfp = options.tfile
self.showhost = options.showhost
self.default_contentview = options.default_contentview

View File

@ -1,11 +1,13 @@
import sys
import click
from mitmproxy import log
class TermLog:
def __init__(self):
def __init__(self, outfile=sys.stdout):
self.options = None
self.outfile = outfile
def configure(self, options, updated):
self.options = options
@ -14,7 +16,7 @@ class TermLog:
if self.options.verbosity >= log.log_tier(e.level):
click.secho(
e.msg,
file=self.options.tfile,
file=self.outfile,
fg=dict(error="red", warn="yellow").get(e.level),
dim=(e.level == "debug"),
err=(e.level == "error")

View File

@ -1,4 +1,4 @@
from typing import Optional, IO
from typing import Optional
from mitmproxy import controller
from mitmproxy import exceptions
@ -20,13 +20,11 @@ class Options(options.Options):
keepserving: bool = False,
filtstr: Optional[str] = None,
flow_detail: int = 1,
tfile: Optional[IO[str]] = None,
**kwargs
) -> None:
self.filtstr = filtstr
self.flow_detail = flow_detail
self.keepserving = keepserving
self.tfile = tfile
super().__init__(**kwargs)

View File

@ -1,5 +1,5 @@
import webbrowser
from typing import Optional, IO
from typing import Optional
import tornado.httpserver
import tornado.ioloop
@ -20,7 +20,6 @@ class Options(options.Options):
self,
*, # all args are keyword-only.
intercept: Optional[str] = None,
tfile: Optional[IO[str]] = None,
open_browser: bool = True,
wdebug: bool = False,
wport: int = 8081,
@ -28,7 +27,6 @@ class Options(options.Options):
**kwargs
) -> None:
self.intercept = intercept
self.tfile = tfile
self.open_browser = open_browser
self.wdebug = wdebug
self.wport = wport

View File

@ -28,43 +28,40 @@ def test_configure():
def test_simple():
d = dumper.Dumper()
sio = io.StringIO()
d = dumper.Dumper(sio)
with taddons.context(options=dump.Options()) as ctx:
sio = io.StringIO()
ctx.configure(d, tfile = sio, flow_detail = 0)
ctx.configure(d, flow_detail = 0)
d.response(tflow.tflow(resp=True))
assert not sio.getvalue()
sio.truncate(0)
ctx.configure(d, tfile = sio, flow_detail = 1)
ctx.configure(d, flow_detail = 1)
d.response(tflow.tflow(resp=True))
assert sio.getvalue()
sio.truncate(0)
ctx.configure(d, tfile = sio, flow_detail = 1)
ctx.configure(d, flow_detail = 1)
d.error(tflow.tflow(err=True))
assert sio.getvalue()
sio.truncate(0)
ctx.configure(d, tfile = sio, flow_detail = 4)
ctx.configure(d, flow_detail = 4)
d.response(tflow.tflow(resp=True))
assert sio.getvalue()
sio.truncate(0)
sio = io.StringIO()
ctx.configure(d, tfile = sio, flow_detail = 4)
ctx.configure(d, flow_detail = 4)
d.response(tflow.tflow(resp=True))
assert "<<" in sio.getvalue()
sio.truncate(0)
sio = io.StringIO()
ctx.configure(d, tfile = sio, flow_detail = 4)
ctx.configure(d, flow_detail = 4)
d.response(tflow.tflow(err=True))
assert "<<" in sio.getvalue()
sio.truncate(0)
sio = io.StringIO()
ctx.configure(d, tfile = sio, flow_detail = 4)
ctx.configure(d, flow_detail = 4)
flow = tflow.tflow()
flow.request = tutils.treq()
flow.request.stickycookie = True
@ -77,8 +74,7 @@ def test_simple():
assert sio.getvalue()
sio.truncate(0)
sio = io.StringIO()
ctx.configure(d, tfile = sio, flow_detail = 4)
ctx.configure(d, flow_detail = 4)
flow = tflow.tflow(resp=tutils.tresp(content=b"{"))
flow.response.headers["content-type"] = "application/json"
flow.response.status_code = 400
@ -86,8 +82,7 @@ def test_simple():
assert sio.getvalue()
sio.truncate(0)
sio = io.StringIO()
ctx.configure(d, tfile = sio, flow_detail = 4)
ctx.configure(d, flow_detail = 4)
flow = tflow.tflow()
flow.request.content = None
flow.response = http.HTTPResponse.wrap(tutils.tresp())
@ -102,20 +97,20 @@ def test_echo_body():
f.response.headers["content-type"] = "text/html"
f.response.content = b"foo bar voing\n" * 100
d = dumper.Dumper()
sio = io.StringIO()
d = dumper.Dumper(sio)
with taddons.context(options=dump.Options()) as ctx:
ctx.configure(d, tfile=sio, flow_detail = 3)
ctx.configure(d, flow_detail = 3)
d._echo_message(f.response)
t = sio.getvalue()
assert "cut off" in t
def test_echo_request_line():
d = dumper.Dumper()
sio = io.StringIO()
d = dumper.Dumper(sio)
with taddons.context(options=dump.Options()) as ctx:
ctx.configure(d, tfile=sio, flow_detail = 3, showhost = True)
ctx.configure(d, flow_detail = 3, showhost = True)
f = tflow.tflow(client_conn=None, server_conn=True, resp=True)
f.request.is_replay = True
d._echo_request_line(f)
@ -139,19 +134,19 @@ class TestContentView:
@mock.patch("mitmproxy.contentviews.ViewAuto.__call__")
def test_contentview(self, view_auto):
view_auto.side_effect = exceptions.ContentViewException("")
d = dumper.Dumper()
sio = io.StringIO()
d = dumper.Dumper(sio)
with taddons.context(options=dump.Options()) as ctx:
sio = io.StringIO()
ctx.configure(d, flow_detail=4, verbosity=3, tfile=sio)
ctx.configure(d, flow_detail=4, verbosity=3)
d.response(tflow.tflow())
assert "Content viewer failed" in ctx.master.event_log[0][1]
def test_tcp():
d = dumper.Dumper()
sio = io.StringIO()
d = dumper.Dumper(sio)
with taddons.context(options=dump.Options()) as ctx:
ctx.configure(d, tfile=sio, flow_detail = 3, showhost = True)
ctx.configure(d, flow_detail = 3, showhost = True)
f = tflow.ttcpflow(client_conn=True, server_conn=True)
d.tcp_message(f)
assert "it's me" in sio.getvalue()

View File

@ -7,9 +7,9 @@ from mitmproxy.tools import dump
class TestTermLog:
def test_simple(self):
t = termlog.TermLog()
sio = io.StringIO()
t.configure(dump.Options(tfile = sio, verbosity = 2), set([]))
t = termlog.TermLog(outfile=sio)
t.configure(dump.Options(verbosity = 2), set([]))
t.log(log.LogEntry("one", "info"))
assert "one" in sio.getvalue()
t.log(log.LogEntry("two", "debug"))