Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2025-01-30 14:58:38 +00:00
Code Issues Projects Releases Wiki Activity

Port mitmproxy test suite entirely to nose.

Browse Source
This commit is contained in:
Aldo Cortesi 2012-06-09 13:42:43 +12:00
parent a63240a848
commit b7b357528c
17 changed files with 388 additions and 480 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc Normal file
View File

@ -0,0 +1,2 @@
[report]
omit = *contrib*

1
.gitignore vendored
View File

@ -10,3 +10,4 @@ mitmproxyc
mitmdumpc mitmdumpc
mitmplaybackc mitmplaybackc
mitmrecordc mitmrecordc
.coverage

8
libmproxy/flow.py
View File

@ -903,9 +903,7 @@ class ClientPlaybackState:
n.request.client_conn = None n.request.client_conn = None
self.current = master.handle_request(n.request) self.current = master.handle_request(n.request)
if not testing and not self.current.response: if not testing and not self.current.response:
#begin nocover master.replay_request(self.current) # pragma: no cover
master.replay_request(self.current)
#end nocover
elif self.current.response: elif self.current.response:
master.handle_response(self.current.response) master.handle_response(self.current.response)
@ -1498,7 +1496,6 @@ class FlowMaster(controller.Master):
return "Can't replay while intercepting..." return "Can't replay while intercepting..."
if f.request.content == CONTENT_MISSING: if f.request.content == CONTENT_MISSING:
return "Can't replay request with missing content..." return "Can't replay request with missing content..."
#begin nocover
if f.request: if f.request:
f.request._set_replay() f.request._set_replay()
if f.request.content: if f.request.content:
@ -1511,8 +1508,7 @@ class FlowMaster(controller.Master):
f, f,
self.masterq, self.masterq,
) )
rt.start() rt.start() # pragma: no cover
#end nocover
def run_script_hook(self, name, *args, **kwargs): def run_script_hook(self, name, *args, **kwargs):
if self.script and not self.pause_scripts: if self.script and not self.pause_scripts:

6
libmproxy/wsgi.py
View File

@ -118,10 +118,8 @@ class WSGIAdaptor:
try: try:
s = traceback.format_exc() s = traceback.format_exc()
self.error_page(soc, state["headers_sent"], s) self.error_page(soc, state["headers_sent"], s)
# begin nocover except Exception, v: # pragma: no cover
except Exception, v: pass # pragma: no cover
pass
# end nocover
return errs.getvalue() return errs.getvalue()

74
test/test_certutils.py
View File

@ -1,11 +1,10 @@
import os import os
import libpry
from libmproxy import certutils from libmproxy import certutils
import tutils
class udummy_ca(libpry.AutoTree): def test_dummy_ca():
def test_all(self): with tutils.tmpdir() as d:
d = self.tmpdir()
path = os.path.join(d, "foo/cert.cnf") path = os.path.join(d, "foo/cert.cnf")
assert certutils.dummy_ca(path) assert certutils.dummy_ca(path)
assert os.path.exists(path) assert os.path.exists(path)
@ -17,45 +16,45 @@ class udummy_ca(libpry.AutoTree):
assert os.path.exists(os.path.join(d, "foo/cert2-cert.p12")) assert os.path.exists(os.path.join(d, "foo/cert2-cert.p12"))
class udummy_cert(libpry.AutoTree): class TestDummyCert:
def test_with_ca(self): def test_with_ca(self):
d = self.tmpdir() with tutils.tmpdir() as d:
cacert = os.path.join(d, "foo/cert.cnf") cacert = os.path.join(d, "foo/cert.cnf")
assert certutils.dummy_ca(cacert) assert certutils.dummy_ca(cacert)
p = certutils.dummy_cert( p = certutils.dummy_cert(
os.path.join(d, "foo"), os.path.join(d, "foo"),
cacert, cacert,
"foo.com", "foo.com",
["one.com", "two.com", "*.three.com"] ["one.com", "two.com", "*.three.com"]
) )
assert os.path.exists(p) assert os.path.exists(p)
# Short-circuit # Short-circuit
assert certutils.dummy_cert( assert certutils.dummy_cert(
os.path.join(d, "foo"), os.path.join(d, "foo"),
cacert, cacert,
"foo.com", "foo.com",
[] []
) )
def test_no_ca(self): def test_no_ca(self):
d = self.tmpdir() with tutils.tmpdir() as d:
p = certutils.dummy_cert( p = certutils.dummy_cert(
d, d,
None, None,
"foo.com", "foo.com",
[] []
) )
assert os.path.exists(p) assert os.path.exists(p)
class uSSLCert(libpry.AutoTree): class TestSSLCert:
def test_simple(self): def test_simple(self):
c = certutils.SSLCert(file("data/text_cert", "r").read()) c = certutils.SSLCert(file(tutils.test_data.path("data/text_cert"), "r").read())
assert c.cn == "google.com" assert c.cn == "google.com"
assert len(c.altnames) == 436 assert len(c.altnames) == 436
c = certutils.SSLCert(file("data/text_cert_2", "r").read()) c = certutils.SSLCert(file(tutils.test_data.path("data/text_cert_2"), "r").read())
assert c.cn == "www.inode.co.nz" assert c.cn == "www.inode.co.nz"
assert len(c.altnames) == 2 assert len(c.altnames) == 2
assert c.digest("sha1") assert c.digest("sha1")
@ -68,13 +67,6 @@ class uSSLCert(libpry.AutoTree):
c.has_expired c.has_expired
def test_der(self): def test_der(self):
d = file("data/dercert").read() d = file(tutils.test_data.path("data/dercert")).read()
s = certutils.SSLCert.from_der(d) s = certutils.SSLCert.from_der(d)
assert s.cn assert s.cn
tests = [
udummy_ca(),
udummy_cert(),
uSSLCert(),
]

150
test/test_cmdline.py
View File

@ -1,95 +1,89 @@
import optparse import optparse
import libpry
from libmproxy import cmdline from libmproxy import cmdline
import tutils
class uAll(libpry.AutoTree): def test_parse_replace_hook():
def test_parse_replace_hook(self): x = cmdline.parse_replace_hook("/foo/bar/voing")
x = cmdline.parse_replace_hook("/foo/bar/voing") assert x == ("foo", "bar", "voing")
assert x == ("foo", "bar", "voing")
x = cmdline.parse_replace_hook("/foo/bar/vo/ing/") x = cmdline.parse_replace_hook("/foo/bar/vo/ing/")
assert x == ("foo", "bar", "vo/ing/") assert x == ("foo", "bar", "vo/ing/")
x = cmdline.parse_replace_hook("/bar/voing") x = cmdline.parse_replace_hook("/bar/voing")
assert x == (".*", "bar", "voing") assert x == (".*", "bar", "voing")
libpry.raises( tutils.raises(
cmdline.ParseReplaceException, cmdline.ParseReplaceException,
cmdline.parse_replace_hook, cmdline.parse_replace_hook,
"/foo" "/foo"
) )
libpry.raises( tutils.raises(
"replacement regex", "replacement regex",
cmdline.parse_replace_hook, cmdline.parse_replace_hook,
"patt/[/rep" "patt/[/rep"
) )
libpry.raises( tutils.raises(
"filter pattern", "filter pattern",
cmdline.parse_replace_hook, cmdline.parse_replace_hook,
"/~/foo/rep" "/~/foo/rep"
) )
libpry.raises( tutils.raises(
"empty replacement regex", "empty replacement regex",
cmdline.parse_replace_hook, cmdline.parse_replace_hook,
"//" "//"
) )
def test_common(self): def test_common():
parser = optparse.OptionParser() parser = optparse.OptionParser()
cmdline.common_options(parser) cmdline.common_options(parser)
opts, args = parser.parse_args(args=[]) opts, args = parser.parse_args(args=[])
assert cmdline.get_common_options(opts) assert cmdline.get_common_options(opts)
opts.stickycookie_all = True opts.stickycookie_all = True
opts.stickyauth_all = True opts.stickyauth_all = True
v = cmdline.get_common_options(opts) v = cmdline.get_common_options(opts)
assert v["stickycookie"] == ".*" assert v["stickycookie"] == ".*"
assert v["stickyauth"] == ".*" assert v["stickyauth"] == ".*"
opts.stickycookie_all = False opts.stickycookie_all = False
opts.stickyauth_all = False opts.stickyauth_all = False
opts.stickycookie_filt = "foo" opts.stickycookie_filt = "foo"
opts.stickyauth_filt = "foo" opts.stickyauth_filt = "foo"
v = cmdline.get_common_options(opts) v = cmdline.get_common_options(opts)
assert v["stickycookie"] == "foo" assert v["stickycookie"] == "foo"
assert v["stickyauth"] == "foo" assert v["stickyauth"] == "foo"
opts.replace = ["/foo/bar/voing"] opts.replace = ["/foo/bar/voing"]
v = cmdline.get_common_options(opts) v = cmdline.get_common_options(opts)
assert v["replacements"] == [("foo", "bar", "voing")] assert v["replacements"] == [("foo", "bar", "voing")]
opts.replace = ["//"] opts.replace = ["//"]
libpry.raises( tutils.raises(
"empty replacement regex", "empty replacement regex",
cmdline.get_common_options, cmdline.get_common_options,
opts opts
) )
opts.replace = [] opts.replace = []
opts.replace_file = [("/foo/bar/nonexistent")] opts.replace_file = [("/foo/bar/nonexistent")]
libpry.raises( tutils.raises(
"could not read replace file", "could not read replace file",
cmdline.get_common_options, cmdline.get_common_options,
opts opts
) )
opts.replace_file = [("/~/bar/nonexistent")] opts.replace_file = [("/~/bar/nonexistent")]
libpry.raises( tutils.raises(
"filter pattern", "filter pattern",
cmdline.get_common_options, cmdline.get_common_options,
opts opts
) )
opts.replace_file = [("/foo/bar/./data/replace")] p = tutils.test_data.path("data/replace")
v = cmdline.get_common_options(opts)["replacements"] opts.replace_file = [("/foo/bar/%s"%p)]
assert len(v) == 1 v = cmdline.get_common_options(opts)["replacements"]
assert v[0][2].strip() == "replacecontents" assert len(v) == 1
assert v[0][2].strip() == "replacecontents"
tests = [
uAll()
]

55
test/test_console.py
View File

@ -1,10 +1,9 @@
import os
from libmproxy import console from libmproxy import console
from libmproxy.console import common from libmproxy.console import common
import tutils import tutils
import libpry
class TestConsoleState:
class uConsoleState(libpry.AutoTree):
def test_flow(self): def test_flow(self):
""" """
normal flow: normal flow:
@ -89,32 +88,34 @@ class uConsoleState(libpry.AutoTree):
assert len(c.flowsettings) == 0 assert len(c.flowsettings) == 0
class uformat_keyvals(libpry.AutoTree): def test_format_keyvals():
def test_simple(self): assert common.format_keyvals(
assert common.format_keyvals( [
[ ("aa", "bb"),
("aa", "bb"), None,
None, ("cc", "dd"),
("cc", "dd"), (None, "dd"),
(None, "dd"), (None, "dd"),
(None, "dd"), ]
] )
)
class uPathCompleter(libpry.AutoTree): class TestPathCompleter:
def test_lookup_construction(self): def test_lookup_construction(self):
c = console._PathCompleter() c = console._PathCompleter()
assert c.complete("/tm") == "/tmp/" assert c.complete("/tm") == "/tmp/"
c.reset() c.reset()
assert c.complete("./completion/a") == "./completion/aaa" cd = tutils.test_data.path("completion")
assert c.complete("./completion/a") == "./completion/aab" ca = os.path.join(cd, "a")
assert c.complete(ca).endswith("/completion/aaa")
assert c.complete(ca).endswith("/completion/aab")
c.reset() c.reset()
assert c.complete("./completion/aaa") == "./completion/aaa" ca = os.path.join(cd, "aaa")
assert c.complete("./completion/aaa") == "./completion/aaa" assert c.complete(ca).endswith("/completion/aaa")
assert c.complete(ca).endswith("/completion/aaa")
c.reset() c.reset()
assert c.complete("./completion") == "./completion/aaa" assert c.complete(cd).endswith("/completion/aaa")
def test_completion(self): def test_completion(self):
c = console._PathCompleter(True) c = console._PathCompleter(True)
@ -144,15 +145,5 @@ class uPathCompleter(libpry.AutoTree):
assert c.final == s assert c.final == s
class uOptions(libpry.AutoTree): def test_options():
def test_all(self): assert console.Options(kill=True)
assert console.Options(kill=True)
tests = [
uformat_keyvals(),
uConsoleState(),
uPathCompleter(),
uOptions()
]

24
test/test_console_contentview.py
View File

@ -1,9 +1,9 @@
import sys import sys
import libpry
import libmproxy.console.contentview as cv import libmproxy.console.contentview as cv
from libmproxy import utils, flow, encoding from libmproxy import utils, flow, encoding
import tutils
class uContentView(libpry.AutoTree): class TestContentView:
def test_trailer(self): def test_trailer(self):
txt = [] txt = []
cv.trailer(5, txt, 1000) cv.trailer(5, txt, 1000)
@ -97,10 +97,18 @@ class uContentView(libpry.AutoTree):
assert cv.view_hex([], "foo", 1000) assert cv.view_hex([], "foo", 1000)
def test_view_image(self): def test_view_image(self):
assert cv.view_image([], file("data/image.png").read(), sys.maxint) p = tutils.test_data.path("data/image.png")
assert cv.view_image([], file("data/image.gif").read(), sys.maxint) assert cv.view_image([], file(p).read(), sys.maxint)
assert cv.view_image([], file("data/image-err1.jpg").read(), sys.maxint)
assert cv.view_image([], file("data/image.ico").read(), sys.maxint) p = tutils.test_data.path("data/image.gif")
assert cv.view_image([], file(p).read(), sys.maxint)
p = tutils.test_data.path("data/image-err1.jpg")
assert cv.view_image([], file(p).read(), sys.maxint)
p = tutils.test_data.path("data/image.ico")
assert cv.view_image([], file(p).read(), sys.maxint)
assert not cv.view_image([], "flibble", sys.maxint) assert not cv.view_image([], "flibble", sys.maxint)
def test_view_multipart(self): def test_view_multipart(self):
@ -178,7 +186,3 @@ Larry
assert "decoded gzip" in r[0] assert "decoded gzip" in r[0]
assert "Raw" in r[0] assert "Raw" in r[0]
tests = [
uContentView()
]

8
test/test_console_help.py
View File

@ -1,5 +1,4 @@
import sys import sys
import libpry
import libmproxy.console.help as help import libmproxy.console.help as help
from libmproxy import utils, flow, encoding from libmproxy import utils, flow, encoding
@ -8,7 +7,7 @@ class DummyMaster:
pass pass
class uHelp(libpry.AutoTree): class TestHelp:
def test_helptext(self): def test_helptext(self):
h = help.HelpView(None, "foo", None) h = help.HelpView(None, "foo", None)
assert h.helptext() assert h.helptext()
@ -18,8 +17,3 @@ class uHelp(libpry.AutoTree):
assert not h.keypress((0, 0), "q") assert not h.keypress((0, 0), "q")
assert not h.keypress((0, 0), "?") assert not h.keypress((0, 0), "?")
assert h.keypress((0, 0), "o") == "o" assert h.keypress((0, 0), "o") == "o"
tests = [
uHelp()
]

87
test/test_dump.py
View File

@ -4,20 +4,19 @@ import libpry
from libmproxy import dump, flow from libmproxy import dump, flow
import tutils import tutils
class uStrFuncs(libpry.AutoTree): def test_strfuncs():
def test_all(self): t = tutils.tresp()
t = tutils.tresp() t._set_replay()
t._set_replay() dump.str_response(t)
dump.str_response(t)
t = tutils.treq() t = tutils.treq()
t.client_conn = None t.client_conn = None
t.stickycookie = True t.stickycookie = True
assert "stickycookie" in dump.str_request(t) assert "stickycookie" in dump.str_request(t)
assert "replay" in dump.str_request(t) assert "replay" in dump.str_request(t)
class uDumpMaster(libpry.AutoTree): class TestDumpMaster:
def _cycle(self, m, content): def _cycle(self, m, content):
req = tutils.treq() req = tutils.treq()
req.content = content req.content = content
@ -53,38 +52,38 @@ class uDumpMaster(libpry.AutoTree):
o = dump.Options(server_replay="nonexistent", kill=True) o = dump.Options(server_replay="nonexistent", kill=True)
libpry.raises(dump.DumpError, dump.DumpMaster, None, o, None, outfile=cs) libpry.raises(dump.DumpError, dump.DumpMaster, None, o, None, outfile=cs)
t = self.tmpdir() with tutils.tmpdir() as t:
p = os.path.join(t, "rep") p = os.path.join(t, "rep")
self._flowfile(p) self._flowfile(p)
o = dump.Options(server_replay=p, kill=True) o = dump.Options(server_replay=p, kill=True)
m = dump.DumpMaster(None, o, None, outfile=cs) m = dump.DumpMaster(None, o, None, outfile=cs)
self._cycle(m, "content") self._cycle(m, "content")
self._cycle(m, "content") self._cycle(m, "content")
o = dump.Options(server_replay=p, kill=False) o = dump.Options(server_replay=p, kill=False)
m = dump.DumpMaster(None, o, None, outfile=cs) m = dump.DumpMaster(None, o, None, outfile=cs)
self._cycle(m, "nonexistent") self._cycle(m, "nonexistent")
o = dump.Options(client_replay=p, kill=False) o = dump.Options(client_replay=p, kill=False)
m = dump.DumpMaster(None, o, None, outfile=cs) m = dump.DumpMaster(None, o, None, outfile=cs)
def test_read(self): def test_read(self):
t = self.tmpdir() with tutils.tmpdir() as t:
p = os.path.join(t, "read") p = os.path.join(t, "read")
self._flowfile(p) self._flowfile(p)
assert "GET" in self._dummy_cycle(0, None, "", verbosity=1, rfile=p) assert "GET" in self._dummy_cycle(0, None, "", verbosity=1, rfile=p)
libpry.raises( libpry.raises(
dump.DumpError, self._dummy_cycle, dump.DumpError, self._dummy_cycle,
0, None, "", verbosity=1, rfile="/nonexistent" 0, None, "", verbosity=1, rfile="/nonexistent"
) )
libpry.raises( libpry.raises(
dump.DumpError, self._dummy_cycle, dump.DumpError, self._dummy_cycle,
0, None, "", verbosity=1, rfile="test_dump.py" 0, None, "", verbosity=1, rfile="test_dump.py"
) )
def test_options(self): def test_options(self):
o = dump.Options(verbosity = 2) o = dump.Options(verbosity = 2)
@ -107,10 +106,10 @@ class uDumpMaster(libpry.AutoTree):
assert "GET" in self._dummy_cycle(1, "~s", "ascii", verbosity=i) assert "GET" in self._dummy_cycle(1, "~s", "ascii", verbosity=i)
def test_write(self): def test_write(self):
d = self.tmpdir() with tutils.tmpdir() as d:
p = os.path.join(d, "a") p = os.path.join(d, "a")
self._dummy_cycle(1, None, "", wfile=p, verbosity=0) self._dummy_cycle(1, None, "", wfile=p, verbosity=0)
assert len(list(flow.FlowReader(open(p)).stream())) == 1 assert len(list(flow.FlowReader(open(p)).stream())) == 1
def test_write_err(self): def test_write_err(self):
libpry.raises( libpry.raises(
@ -125,7 +124,7 @@ class uDumpMaster(libpry.AutoTree):
def test_script(self): def test_script(self):
ret = self._dummy_cycle( ret = self._dummy_cycle(
1, None, "", 1, None, "",
script="scripts/all.py", verbosity=0, eventlog=True script=tutils.test_data.path("scripts/all.py"), verbosity=0, eventlog=True
) )
assert "XCLIENTCONNECT" in ret assert "XCLIENTCONNECT" in ret
assert "XREQUEST" in ret assert "XREQUEST" in ret
@ -146,11 +145,3 @@ class uDumpMaster(libpry.AutoTree):
def test_stickyauth(self): def test_stickyauth(self):
self._dummy_cycle(1, None, "", stickyauth = ".*") self._dummy_cycle(1, None, "", stickyauth = ".*")
tests = [
uStrFuncs(),
uDumpMaster()
]

35
test/test_encoding.py
View File

@ -1,28 +1,19 @@
from libmproxy import encoding from libmproxy import encoding
import libpry
class uidentity(libpry.AutoTree): def test_identity():
def test_simple(self): assert "string" == encoding.decode("identity", "string")
assert "string" == encoding.decode("identity", "string") assert "string" == encoding.encode("identity", "string")
assert "string" == encoding.encode("identity", "string") assert not encoding.encode("nonexistent", "string")
assert not encoding.encode("nonexistent", "string") assert None == encoding.decode("nonexistent encoding", "string")
def test_fallthrough(self):
assert None == encoding.decode("nonexistent encoding", "string")
class ugzip(libpry.AutoTree): def test_gzip():
def test_simple(self): assert "string" == encoding.decode("gzip", encoding.encode("gzip", "string"))
assert "string" == encoding.decode("gzip", encoding.encode("gzip", "string")) assert None == encoding.decode("gzip", "bogus")
assert None == encoding.decode("gzip", "bogus")
class udeflate(libpry.AutoTree):
def test_simple(self):
assert "string" == encoding.decode("deflate", encoding.encode("deflate", "string"))
assert "string" == encoding.decode("deflate", encoding.encode("deflate", "string")[2:-4])
assert None == encoding.decode("deflate", "bogus")
tests = [ def test_deflate():
uidentity(), assert "string" == encoding.decode("deflate", encoding.encode("deflate", "string"))
ugzip(), assert "string" == encoding.decode("deflate", encoding.encode("deflate", "string")[2:-4])
udeflate() assert None == encoding.decode("deflate", "bogus")
]

13
test/test_filt.py
View File

@ -1,9 +1,8 @@
import cStringIO import cStringIO
from libmproxy import filt, flow from libmproxy import filt, flow
import libpry
class uParsing(libpry.AutoTree): class TestParsing:
def _dump(self, x): def _dump(self, x):
c = cStringIO.StringIO() c = cStringIO.StringIO()
x.dump(fp=c) x.dump(fp=c)
@ -71,7 +70,7 @@ class uParsing(libpry.AutoTree):
self._dump(a) self._dump(a)
class uMatching(libpry.AutoTree): class TestMatching:
def req(self): def req(self):
conn = flow.ClientConnect(("one", 2222)) conn = flow.ClientConnect(("one", 2222))
headers = flow.ODictCaseless() headers = flow.ODictCaseless()
@ -235,11 +234,3 @@ class uMatching(libpry.AutoTree):
assert self.q("!~c 201 !~c 202", s) assert self.q("!~c 201 !~c 202", s)
assert not self.q("!~c 201 !~c 200", s) assert not self.q("!~c 201 !~c 200", s)
tests = [
uMatching(),
uParsing()
]

176
test/test_flow.py
View File

@ -6,7 +6,7 @@ import tutils
import libpry import libpry
class uStickyCookieState(libpry.AutoTree): class TestStickyCookieState:
def _response(self, cookie, host): def _response(self, cookie, host):
s = flow.StickyCookieState(filt.parse(".*")) s = flow.StickyCookieState(filt.parse(".*"))
f = tutils.tflow_full() f = tutils.tflow_full()
@ -40,7 +40,7 @@ class uStickyCookieState(libpry.AutoTree):
assert "cookie" in f.request.headers assert "cookie" in f.request.headers
class uStickyAuthState(libpry.AutoTree): class TestStickyAuthState:
def test_handle_response(self): def test_handle_response(self):
s = flow.StickyAuthState(filt.parse(".*")) s = flow.StickyAuthState(filt.parse(".*"))
f = tutils.tflow_full() f = tutils.tflow_full()
@ -53,7 +53,7 @@ class uStickyAuthState(libpry.AutoTree):
assert f.request.headers["authorization"] == ["foo"] assert f.request.headers["authorization"] == ["foo"]
class uClientPlaybackState(libpry.AutoTree): class TestClientPlaybackState:
def test_tick(self): def test_tick(self):
first = tutils.tflow() first = tutils.tflow()
s = flow.State() s = flow.State()
@ -85,7 +85,7 @@ class uClientPlaybackState(libpry.AutoTree):
assert not fm.client_playback assert not fm.client_playback
class uServerPlaybackState(libpry.AutoTree): class TestServerPlaybackState:
def test_hash(self): def test_hash(self):
s = flow.ServerPlaybackState(None, [], False, False) s = flow.ServerPlaybackState(None, [], False, False)
r = tutils.tflow() r = tutils.tflow()
@ -147,7 +147,8 @@ class uServerPlaybackState(libpry.AutoTree):
n = s.next_flow(r) n = s.next_flow(r)
assert s.count() == 2 assert s.count() == 2
class uFlow(libpry.AutoTree):
class TestFlow:
def test_copy(self): def test_copy(self):
f = tutils.tflow_full() f = tutils.tflow_full()
f2 = f.copy() f2 = f.copy()
@ -302,7 +303,7 @@ class uFlow(libpry.AutoTree):
class uState(libpry.AutoTree): class TestState:
def test_backup(self): def test_backup(self):
c = flow.State() c = flow.State()
req = tutils.treq() req = tutils.treq()
@ -451,7 +452,7 @@ class uState(libpry.AutoTree):
c.accept_all() c.accept_all()
class uSerialize(libpry.AutoTree): class TestSerialize:
def _treader(self): def _treader(self):
sio = StringIO() sio = StringIO()
w = flow.FlowWriter(sio) w = flow.FlowWriter(sio)
@ -494,7 +495,7 @@ class uSerialize(libpry.AutoTree):
sio.write("bogus") sio.write("bogus")
sio.seek(0) sio.seek(0)
r = flow.FlowReader(sio) r = flow.FlowReader(sio)
libpry.raises(flow.FlowReadError, list, r.stream()) tutils.raises(flow.FlowReadError, list, r.stream())
f = flow.FlowReadError("foo") f = flow.FlowReadError("foo")
assert f.strerror == "foo" assert f.strerror == "foo"
@ -508,18 +509,18 @@ class uSerialize(libpry.AutoTree):
sio.seek(0) sio.seek(0)
r = flow.FlowReader(sio) r = flow.FlowReader(sio)
libpry.raises("version", list, r.stream()) tutils.raises("version", list, r.stream())
class uFlowMaster(libpry.AutoTree): class TestFlowMaster:
def test_load_script(self): def test_load_script(self):
s = flow.State() s = flow.State()
fm = flow.FlowMaster(None, s) fm = flow.FlowMaster(None, s)
assert not fm.load_script("scripts/a.py") assert not fm.load_script(tutils.test_data.path("scripts/a.py"))
assert not fm.load_script("scripts/a.py") assert not fm.load_script(tutils.test_data.path("scripts/a.py"))
assert not fm.load_script(None) assert not fm.load_script(None)
assert fm.load_script("nonexistent") assert fm.load_script("nonexistent")
assert "ValueError" in fm.load_script("scripts/starterr.py") assert "ValueError" in fm.load_script(tutils.test_data.path("scripts/starterr.py"))
def test_replay(self): def test_replay(self):
s = flow.State() s = flow.State()
@ -534,7 +535,7 @@ class uFlowMaster(libpry.AutoTree):
def test_script_reqerr(self): def test_script_reqerr(self):
s = flow.State() s = flow.State()
fm = flow.FlowMaster(None, s) fm = flow.FlowMaster(None, s)
assert not fm.load_script("scripts/reqerr.py") assert not fm.load_script(tutils.test_data.path("scripts/reqerr.py"))
req = tutils.treq() req = tutils.treq()
fm.handle_clientconnect(req.client_conn) fm.handle_clientconnect(req.client_conn)
assert fm.handle_request(req) assert fm.handle_request(req)
@ -542,7 +543,7 @@ class uFlowMaster(libpry.AutoTree):
def test_script(self): def test_script(self):
s = flow.State() s = flow.State()
fm = flow.FlowMaster(None, s) fm = flow.FlowMaster(None, s)
assert not fm.load_script("scripts/all.py") assert not fm.load_script(tutils.test_data.path("scripts/all.py"))
req = tutils.treq() req = tutils.treq()
fm.handle_clientconnect(req.client_conn) fm.handle_clientconnect(req.client_conn)
assert fm.script.ns["log"][-1] == "clientconnect" assert fm.script.ns["log"][-1] == "clientconnect"
@ -680,7 +681,7 @@ class uFlowMaster(libpry.AutoTree):
fm.handle_request(f.request) fm.handle_request(f.request)
assert f.request.headers["authorization"] == ["foo"] assert f.request.headers["authorization"] == ["foo"]
class uRequest(libpry.AutoTree): class TestRequest:
def test_simple(self): def test_simple(self):
h = flow.ODictCaseless() h = flow.ODictCaseless()
h["test"] = ["test"] h["test"] = ["test"]
@ -819,7 +820,7 @@ class uRequest(libpry.AutoTree):
assert r.content == "falafel" assert r.content == "falafel"
class uResponse(libpry.AutoTree): class TestResponse:
def test_simple(self): def test_simple(self):
h = flow.ODictCaseless() h = flow.ODictCaseless()
h["test"] = ["test"] h["test"] = ["test"]
@ -869,7 +870,10 @@ class uResponse(libpry.AutoTree):
def test_get_cert(self): def test_get_cert(self):
req = tutils.treq() req = tutils.treq()
resp = flow.Response(req, 200, "msg", flow.ODictCaseless(), "content", file("data/dercert").read()) resp = flow.Response(
req, 200, "msg", flow.ODictCaseless(), "content",
file(tutils.test_data.path("data/dercert")).read()
)
assert resp.get_cert() assert resp.get_cert()
resp = tutils.tresp() resp = tutils.tresp()
@ -924,7 +928,7 @@ class uResponse(libpry.AutoTree):
assert r.content == "falafel" assert r.content == "falafel"
class uError(libpry.AutoTree): class TestError:
def test_getset_state(self): def test_getset_state(self):
e = flow.Error(None, "Error") e = flow.Error(None, "Error")
state = e._get_state() state = e._get_state()
@ -947,7 +951,7 @@ class uError(libpry.AutoTree):
assert e.msg == "abarp" assert e.msg == "abarp"
class uClientConnect(libpry.AutoTree): class TestClientConnect:
def test_state(self): def test_state(self):
c = flow.ClientConnect(("a", 22)) c = flow.ClientConnect(("a", 22))
assert flow.ClientConnect._from_state(c._get_state()) == c assert flow.ClientConnect._from_state(c._get_state()) == c
@ -963,13 +967,13 @@ class uClientConnect(libpry.AutoTree):
assert c3 == c assert c3 == c
class uODict(libpry.AutoTree): class TestODict:
def setUp(self): def setUp(self):
self.od = flow.ODict() self.od = flow.ODict()
def test_str_err(self): def test_str_err(self):
h = flow.ODict() h = flow.ODict()
libpry.raises(ValueError, h.__setitem__, "key", "foo") tutils.raises(ValueError, h.__setitem__, "key", "foo")
def test_dictToHeader1(self): def test_dictToHeader1(self):
self.od.add("one", "uno") self.od.add("one", "uno")
@ -1047,7 +1051,7 @@ class uODict(libpry.AutoTree):
assert self.od.get("two") == None assert self.od.get("two") == None
class uODictCaseless(libpry.AutoTree): class TestODictCaseless:
def setUp(self): def setUp(self):
self.od = flow.ODictCaseless() self.od = flow.ODictCaseless()
@ -1068,89 +1072,67 @@ class uODictCaseless(libpry.AutoTree):
assert len(self.od) == 1 assert len(self.od) == 1
class udecoded(libpry.AutoTree): def test_decoded():
def test_del(self): r = tutils.treq()
r = tutils.treq() assert r.content == "content"
assert r.content == "content" assert not r.headers["content-encoding"]
r.encode("gzip")
assert r.headers["content-encoding"]
assert r.content != "content"
with flow.decoded(r):
assert not r.headers["content-encoding"] assert not r.headers["content-encoding"]
r.encode("gzip") assert r.content == "content"
assert r.headers["content-encoding"] assert r.headers["content-encoding"]
assert r.content != "content" assert r.content != "content"
with flow.decoded(r):
assert not r.headers["content-encoding"]
assert r.content == "content"
assert r.headers["content-encoding"]
assert r.content != "content"
with flow.decoded(r): with flow.decoded(r):
r.content = "foo" r.content = "foo"
assert r.content != "foo" assert r.content != "foo"
r.decode() r.decode()
assert r.content == "foo" assert r.content == "foo"
class uReplaceHooks(libpry.AutoTree): def test_replacehooks():
def test_add_remove(self): h = flow.ReplaceHooks()
h = flow.ReplaceHooks() h.add("~q", "foo", "bar")
h.add("~q", "foo", "bar") assert h.lst
assert h.lst h.remove("~q", "foo", "bar")
h.remove("~q", "foo", "bar") assert not h.lst
assert not h.lst
h.add("~q", "foo", "bar") h.add("~q", "foo", "bar")
h.add("~s", "foo", "bar") h.add("~s", "foo", "bar")
v = h.get_specs() v = h.get_specs()
assert v == [('~q', 'foo', 'bar'), ('~s', 'foo', 'bar')] assert v == [('~q', 'foo', 'bar'), ('~s', 'foo', 'bar')]
assert h.count() == 2 assert h.count() == 2
h.remove("~q", "foo", "bar") h.remove("~q", "foo", "bar")
assert h.count() == 1 assert h.count() == 1
h.remove("~q", "foo", "bar") h.remove("~q", "foo", "bar")
assert h.count() == 1 assert h.count() == 1
h.clear() h.clear()
assert h.count() == 0 assert h.count() == 0
f = tutils.tflow() f = tutils.tflow()
f.request.content = "foo" f.request.content = "foo"
h.add("~s", "foo", "bar") h.add("~s", "foo", "bar")
h.run(f) h.run(f)
assert f.request.content == "foo" assert f.request.content == "foo"
f = tutils.tflow_full() f = tutils.tflow_full()
f.request.content = "foo" f.request.content = "foo"
f.response.content = "foo" f.response.content = "foo"
h.run(f) h.run(f)
assert f.response.content == "bar" assert f.response.content == "bar"
assert f.request.content == "foo" assert f.request.content == "foo"
f = tutils.tflow() f = tutils.tflow()
h.clear() h.clear()
h.add("~q", "foo", "bar") h.add("~q", "foo", "bar")
f.request.content = "foo" f.request.content = "foo"
h.run(f) h.run(f)
assert f.request.content == "bar" assert f.request.content == "bar"
assert not h.add("~", "foo", "bar") assert not h.add("~", "foo", "bar")
assert not h.add("foo", "*", "bar") assert not h.add("foo", "*", "bar")
tests = [
uReplaceHooks(),
uStickyCookieState(),
uStickyAuthState(),
uServerPlaybackState(),
uClientPlaybackState(),
uFlow(),
uState(),
uSerialize(),
uFlowMaster(),
uRequest(),
uResponse(),
uError(),
uClientConnect(),
uODict(),
uODictCaseless(),
udecoded()
]

170
test/test_proxy.py
View File

@ -2,60 +2,57 @@ import cStringIO, textwrap
from cStringIO import StringIO from cStringIO import StringIO
import libpry import libpry
from libmproxy import proxy, flow from libmproxy import proxy, flow
import tutils
class u_read_chunked(libpry.AutoTree):
def test_all(self):
s = cStringIO.StringIO("1\r\na\r\n0\r\n")
libpry.raises(IOError, proxy.read_chunked, s, None)
s = cStringIO.StringIO("1\r\na\r\n0\r\n\r\n")
assert proxy.read_chunked(s, None) == "a"
s = cStringIO.StringIO("\r\n")
libpry.raises(IOError, proxy.read_chunked, s, None)
s = cStringIO.StringIO("1\r\nfoo")
libpry.raises(IOError, proxy.read_chunked, s, None)
s = cStringIO.StringIO("foo\r\nfoo")
libpry.raises(proxy.ProxyError, proxy.read_chunked, s, None)
class Dummy: pass class Dummy: pass
class u_read_http_body(libpry.AutoTree): def test_read_chunked():
def test_all(self): s = cStringIO.StringIO("1\r\na\r\n0\r\n")
tutils.raises(IOError, proxy.read_chunked, s, None)
d = Dummy() s = cStringIO.StringIO("1\r\na\r\n0\r\n\r\n")
h = flow.ODict() assert proxy.read_chunked(s, None) == "a"
s = cStringIO.StringIO("testing")
assert proxy.read_http_body(s, d, h, False, None) == ""
h["content-length"] = ["foo"] s = cStringIO.StringIO("\r\n")
s = cStringIO.StringIO("testing") tutils.raises(IOError, proxy.read_chunked, s, None)
libpry.raises(proxy.ProxyError, proxy.read_http_body, s, d, h, False, None)
h["content-length"] = [5] s = cStringIO.StringIO("1\r\nfoo")
s = cStringIO.StringIO("testing") tutils.raises(IOError, proxy.read_chunked, s, None)
assert len(proxy.read_http_body(s, d, h, False, None)) == 5
s = cStringIO.StringIO("testing")
libpry.raises(proxy.ProxyError, proxy.read_http_body, s, d, h, False, 4)
h = flow.ODict() s = cStringIO.StringIO("foo\r\nfoo")
s = cStringIO.StringIO("testing") tutils.raises(proxy.ProxyError, proxy.read_chunked, s, None)
assert len(proxy.read_http_body(s, d, h, True, 4)) == 4
s = cStringIO.StringIO("testing")
assert len(proxy.read_http_body(s, d, h, True, 100)) == 7
class u_parse_request_line(libpry.AutoTree): def test_read_http_body():
d = Dummy()
h = flow.ODict()
s = cStringIO.StringIO("testing")
assert proxy.read_http_body(s, d, h, False, None) == ""
h["content-length"] = ["foo"]
s = cStringIO.StringIO("testing")
tutils.raises(proxy.ProxyError, proxy.read_http_body, s, d, h, False, None)
h["content-length"] = [5]
s = cStringIO.StringIO("testing")
assert len(proxy.read_http_body(s, d, h, False, None)) == 5
s = cStringIO.StringIO("testing")
tutils.raises(proxy.ProxyError, proxy.read_http_body, s, d, h, False, 4)
h = flow.ODict()
s = cStringIO.StringIO("testing")
assert len(proxy.read_http_body(s, d, h, True, 4)) == 4
s = cStringIO.StringIO("testing")
assert len(proxy.read_http_body(s, d, h, True, 100)) == 7
class TestParseRequestLine:
def test_simple(self): def test_simple(self):
libpry.raises(proxy.ProxyError, proxy.parse_request_line, "") tutils.raises(proxy.ProxyError, proxy.parse_request_line, "")
u = "GET ... HTTP/1.1" u = "GET ... HTTP/1.1"
libpry.raises("invalid url", proxy.parse_request_line, u) tutils.raises("invalid url", proxy.parse_request_line, u)
u = "GET http://foo.com:8888/test HTTP/1.1" u = "GET http://foo.com:8888/test HTTP/1.1"
m, s, h, po, pa, minor = proxy.parse_request_line(u) m, s, h, po, pa, minor = proxy.parse_request_line(u)
@ -77,7 +74,7 @@ class u_parse_request_line(libpry.AutoTree):
assert proxy.parse_request_line(u) == ('GET', None, None, None, '/', 1) assert proxy.parse_request_line(u) == ('GET', None, None, None, '/', 1)
class uFileLike(libpry.AutoTree): class TestFileLike:
def test_wrap(self): def test_wrap(self):
s = cStringIO.StringIO("foobar\nfoobar") s = cStringIO.StringIO("foobar\nfoobar")
s = proxy.FileLike(s) s = proxy.FileLike(s)
@ -88,14 +85,13 @@ class uFileLike(libpry.AutoTree):
assert s.isatty assert s.isatty
class TestProxyError:
class uProxyError(libpry.AutoTree):
def test_simple(self): def test_simple(self):
p = proxy.ProxyError(111, "msg") p = proxy.ProxyError(111, "msg")
assert repr(p) assert repr(p)
class u_read_headers(libpry.AutoTree): class TestReadHeaders:
def test_read_simple(self): def test_read_simple(self):
data = """ data = """
Header: one Header: one
@ -135,63 +131,45 @@ class u_read_headers(libpry.AutoTree):
assert headers["header"] == ['one\r\n two'] assert headers["header"] == ['one\r\n two']
class u_parse_http_protocol(libpry.AutoTree): def test_parse_http_protocol():
def test_simple(self): assert proxy.parse_http_protocol("HTTP/1.1") == (1, 1)
assert proxy.parse_http_protocol("HTTP/1.1") == (1, 1) assert proxy.parse_http_protocol("HTTP/0.0") == (0, 0)
assert proxy.parse_http_protocol("HTTP/0.0") == (0, 0) assert not proxy.parse_http_protocol("foo/0.0")
assert not proxy.parse_http_protocol("foo/0.0")
class u_parse_init_connect(libpry.AutoTree): def test_parse_init_connect():
def test_simple(self): assert proxy.parse_init_connect("CONNECT host.com:443 HTTP/1.0")
assert proxy.parse_init_connect("CONNECT host.com:443 HTTP/1.0") assert not proxy.parse_init_connect("bogus")
assert not proxy.parse_init_connect("bogus") assert not proxy.parse_init_connect("GET host.com:443 HTTP/1.0")
assert not proxy.parse_init_connect("GET host.com:443 HTTP/1.0") assert not proxy.parse_init_connect("CONNECT host.com443 HTTP/1.0")
assert not proxy.parse_init_connect("CONNECT host.com443 HTTP/1.0") assert not proxy.parse_init_connect("CONNECT host.com:443 foo/1.0")
assert not proxy.parse_init_connect("CONNECT host.com:443 foo/1.0")
class u_parse_init_proxy(libpry.AutoTree): def test_prase_init_proxy():
def test_simple(self): u = "GET http://foo.com:8888/test HTTP/1.1"
u = "GET http://foo.com:8888/test HTTP/1.1" m, s, h, po, pa, major, minor = proxy.parse_init_proxy(u)
m, s, h, po, pa, major, minor = proxy.parse_init_proxy(u) assert m == "GET"
assert m == "GET" assert s == "http"
assert s == "http" assert h == "foo.com"
assert h == "foo.com" assert po == 8888
assert po == 8888 assert pa == "/test"
assert pa == "/test" assert major == 1
assert major == 1 assert minor == 1
assert minor == 1
assert not proxy.parse_init_proxy("invalid") assert not proxy.parse_init_proxy("invalid")
assert not proxy.parse_init_proxy("GET invalid HTTP/1.1") assert not proxy.parse_init_proxy("GET invalid HTTP/1.1")
assert not proxy.parse_init_proxy("GET http://foo.com:8888/test foo/1.1") assert not proxy.parse_init_proxy("GET http://foo.com:8888/test foo/1.1")
class u_parse_init_http(libpry.AutoTree): def test_parse_init_http():
def test_simple(self): u = "GET /test HTTP/1.1"
u = "GET /test HTTP/1.1" m, u, major, minor = proxy.parse_init_http(u)
m, u, major, minor = proxy.parse_init_http(u) assert m == "GET"
assert m == "GET" assert u == "/test"
assert u == "/test" assert major == 1
assert major == 1 assert minor == 1
assert minor == 1
assert not proxy.parse_init_http("invalid") assert not proxy.parse_init_http("invalid")
assert not proxy.parse_init_http("GET invalid HTTP/1.1") assert not proxy.parse_init_http("GET invalid HTTP/1.1")
assert not proxy.parse_init_http("GET /test foo/1.1") assert not proxy.parse_init_http("GET /test foo/1.1")
tests = [
u_parse_http_protocol(),
u_parse_init_connect(),
u_parse_init_proxy(),
u_parse_init_http(),
uProxyError(),
uFileLike(),
u_parse_request_line(),
u_read_chunked(),
u_read_http_body(),
u_read_headers()
]

27
test/test_script.py
View File

@ -1,15 +1,14 @@
import os import os
from libmproxy import script, flow from libmproxy import script, flow
import libpry
import tutils import tutils
class uScript(libpry.AutoTree): class TestScript:
def test_simple(self): def test_simple(self):
s = flow.State() s = flow.State()
fm = flow.FlowMaster(None, s) fm = flow.FlowMaster(None, s)
ctx = flow.ScriptContext(fm) ctx = flow.ScriptContext(fm)
p = script.Script(os.path.join("scripts", "a.py"), ctx) p = script.Script(tutils.test_data.path("scripts/a.py"), ctx)
p.load() p.load()
assert "here" in p.ns assert "here" in p.ns
assert p.run("here") == (True, 1) assert p.run("here") == (True, 1)
@ -26,7 +25,7 @@ class uScript(libpry.AutoTree):
def test_duplicate_flow(self): def test_duplicate_flow(self):
s = flow.State() s = flow.State()
fm = flow.FlowMaster(None, s) fm = flow.FlowMaster(None, s)
fm.load_script(os.path.join("scripts", "duplicate_flow.py")) fm.load_script(tutils.test_data.path("scripts/duplicate_flow.py"))
r = tutils.treq() r = tutils.treq()
fm.handle_request(r) fm.handle_request(r)
assert fm.state.flow_count() == 2 assert fm.state.flow_count() == 2
@ -40,32 +39,26 @@ class uScript(libpry.AutoTree):
s = script.Script("nonexistent", ctx) s = script.Script("nonexistent", ctx)
libpry.raises( tutils.raises(
"no such file", "no such file",
s.load s.load
) )
s = script.Script("scripts", ctx) s = script.Script(tutils.test_data.path("scripts"), ctx)
libpry.raises( tutils.raises(
"not a file", "not a file",
s.load s.load
) )
s = script.Script("scripts/syntaxerr.py", ctx) s = script.Script(tutils.test_data.path("scripts/syntaxerr.py"), ctx)
libpry.raises( tutils.raises(
script.ScriptError, script.ScriptError,
s.load s.load
) )
s = script.Script("scripts/loaderr.py", ctx) s = script.Script(tutils.test_data.path("scripts/loaderr.py"), ctx)
libpry.raises( tutils.raises(
script.ScriptError, script.ScriptError,
s.load s.load
) )
tests = [
uScript(),
]

11
test/test_wsgi.py
View File

@ -1,5 +1,4 @@
import cStringIO, sys import cStringIO, sys
import libpry
from libmproxy import wsgi from libmproxy import wsgi
import tutils import tutils
@ -16,7 +15,7 @@ class TestApp:
return ['Hello', ' world!\n'] return ['Hello', ' world!\n']
class uWSGIAdaptor(libpry.AutoTree): class TestWSGIAdaptor:
def test_make_environ(self): def test_make_environ(self):
w = wsgi.WSGIAdaptor(None, "foo", 80) w = wsgi.WSGIAdaptor(None, "foo", 80)
tr = tutils.treq() tr = tutils.treq()
@ -97,7 +96,7 @@ class uWSGIAdaptor(libpry.AutoTree):
assert "Internal Server Error" in self._serve(app) assert "Internal Server Error" in self._serve(app)
class uAppRegistry(libpry.AutoTree): class TestAppRegistry:
def test_add_get(self): def test_add_get(self):
ar = wsgi.AppRegistry() ar = wsgi.AppRegistry()
ar.add("foo", "domain", 80) ar.add("foo", "domain", 80)
@ -109,9 +108,3 @@ class uAppRegistry(libpry.AutoTree):
r.port = 81 r.port = 81
assert not ar.get(r) assert not ar.get(r)
tests = [
uWSGIAdaptor(),
uAppRegistry()
]

21
test/tutils.py
View File

@ -1,6 +1,8 @@
import threading, Queue, unittest import threading, Queue
import os, shutil,tempfile
from contextlib import contextmanager
import libpry import libpry
from libmproxy import proxy, flow, controller from libmproxy import proxy, flow, controller, utils
import requests import requests
import libpathod.test import libpathod.test
import random import random
@ -128,6 +130,19 @@ class ProxTest:
return pthread.tmaster.log return pthread.tmaster.log
@contextmanager
def tmpdir(*args, **kwargs):
orig_workdir = os.getcwd()
temp_workdir = tempfile.mkdtemp(*args, **kwargs)
os.chdir(temp_workdir)
yield temp_workdir
os.chdir(orig_workdir)
shutil.rmtree(temp_workdir)
def raises(exc, obj, *args, **kwargs): def raises(exc, obj, *args, **kwargs):
""" """
Assert that a callable raises a specified exception. Assert that a callable raises a specified exception.
@ -165,3 +180,5 @@ def raises(exc, obj, *args, **kwargs):
) )
) )
raise AssertionError("No exception raised.") raise AssertionError("No exception raised.")
test_data = utils.Data(__name__)