addons.serverplayback: 100% test coverage

This commit is contained in:
Aldo Cortesi 2016-11-03 19:58:18 +13:00
parent e1fc80937d
commit d7d6edb3d1
2 changed files with 296 additions and 249 deletions

View File

@ -1,3 +1,5 @@
import os
from mitmproxy.test import tutils
from mitmproxy.test import tflow
from mitmproxy.test import taddons
@ -5,283 +7,326 @@ import mitmproxy.test.tutils
from mitmproxy.addons import serverplayback
from mitmproxy import options
from mitmproxy import exceptions
from mitmproxy import io
class TestServerPlayback:
def test_server_playback(self):
sp = serverplayback.ServerPlayback()
sp.configure(options.Options(), [])
f = tflow.tflow(resp=True)
def tdump(path, flows):
w = io.FlowWriter(open(path, "wb"))
for i in flows:
w.add(i)
assert not sp.flowmap
sp.load([f])
assert sp.flowmap
assert sp.next_flow(f)
assert not sp.flowmap
def test_ignore_host(self):
sp = serverplayback.ServerPlayback()
sp.configure(options.Options(server_replay_ignore_host=True), [])
r = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
r.request.host = "address"
r2.request.host = "address"
assert sp._hash(r) == sp._hash(r2)
r2.request.host = "wrong_address"
assert sp._hash(r) == sp._hash(r2)
def test_ignore_content(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(server_replay_ignore_content=False), [])
r = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
r.request.content = b"foo"
r2.request.content = b"foo"
assert s._hash(r) == s._hash(r2)
r2.request.content = b"bar"
assert not s._hash(r) == s._hash(r2)
s.configure(options.Options(server_replay_ignore_content=True), [])
r = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
r.request.content = b"foo"
r2.request.content = b"foo"
assert s._hash(r) == s._hash(r2)
r2.request.content = b"bar"
assert s._hash(r) == s._hash(r2)
r2.request.content = b""
assert s._hash(r) == s._hash(r2)
r2.request.content = None
assert s._hash(r) == s._hash(r2)
def test_ignore_content_wins_over_params(self):
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
server_replay_ignore_content=True,
server_replay_ignore_payload_params=[
"param1", "param2"
]
),
[]
)
# NOTE: parameters are mutually exclusive in options
r = tflow.tflow(resp=True)
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = b"paramx=y"
r2 = tflow.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = b"paramx=x"
# same parameters
assert s._hash(r) == s._hash(r2)
def test_ignore_payload_params_other_content_type(self):
s = serverplayback.ServerPlayback()
def test_config():
s = serverplayback.ServerPlayback()
with tutils.tmpdir() as p:
with taddons.context() as tctx:
tctx.configure(
s,
server_replay_ignore_content=False,
server_replay_ignore_payload_params=[
"param1", "param2"
]
)
fpath = os.path.join(p, "flows")
tdump(fpath, [tflow.tflow(resp=True)])
tctx.configure(s, server_replay = [fpath])
tutils.raises(exceptions.OptionsError, tctx.configure, s, server_replay = [p])
r = tflow.tflow(resp=True)
r.request.headers["Content-Type"] = "application/json"
r.request.content = b'{"param1":"1"}'
r2 = tflow.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/json"
r2.request.content = b'{"param1":"1"}'
# same content
assert s._hash(r) == s._hash(r2)
# distint content (note only x-www-form-urlencoded payload is analysed)
r2.request.content = b'{"param1":"2"}'
assert not s._hash(r) == s._hash(r2)
def test_hash(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(), [])
def test_tick():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
s.stop = True
s.final_flow = tflow.tflow()
s.final_flow.live = False
s.tick()
assert tctx.master.should_exit.is_set()
r = tflow.tflow()
r2 = tflow.tflow()
assert s._hash(r)
assert s._hash(r) == s._hash(r2)
r.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r.request.path = "voing"
assert s._hash(r) != s._hash(r2)
def test_server_playback():
sp = serverplayback.ServerPlayback()
sp.configure(options.Options(), [])
f = tflow.tflow(resp=True)
r.request.path = "path?blank_value"
r2.request.path = "path?"
assert s._hash(r) != s._hash(r2)
assert not sp.flowmap
def test_headers(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(server_replay_use_headers=["foo"]), [])
sp.load([f])
assert sp.flowmap
assert sp.next_flow(f)
assert not sp.flowmap
r = tflow.tflow(resp=True)
r.request.headers["foo"] = "bar"
r2 = tflow.tflow(resp=True)
assert not s._hash(r) == s._hash(r2)
r2.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r2.request.headers["oink"] = "bar"
assert s._hash(r) == s._hash(r2)
sp.load([f])
assert sp.flowmap
sp.clear()
assert not sp.flowmap
r = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
assert s._hash(r) == s._hash(r2)
def test_load(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(), [])
def test_ignore_host():
sp = serverplayback.ServerPlayback()
sp.configure(options.Options(server_replay_ignore_host=True), [])
r = tflow.tflow(resp=True)
r.request.headers["key"] = "one"
r = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
r2.request.headers["key"] = "two"
r.request.host = "address"
r2.request.host = "address"
assert sp._hash(r) == sp._hash(r2)
r2.request.host = "wrong_address"
assert sp._hash(r) == sp._hash(r2)
s.load([r, r2])
assert s.count() == 2
def test_ignore_content():
s = serverplayback.ServerPlayback()
s.configure(options.Options(server_replay_ignore_content=False), [])
n = s.next_flow(r)
assert n.request.headers["key"] == "one"
assert s.count() == 1
r = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
n = s.next_flow(r)
assert n.request.headers["key"] == "two"
assert not s.flowmap
assert s.count() == 0
r.request.content = b"foo"
r2.request.content = b"foo"
assert s._hash(r) == s._hash(r2)
r2.request.content = b"bar"
assert not s._hash(r) == s._hash(r2)
assert not s.next_flow(r)
s.configure(options.Options(server_replay_ignore_content=True), [])
r = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
r.request.content = b"foo"
r2.request.content = b"foo"
assert s._hash(r) == s._hash(r2)
r2.request.content = b"bar"
assert s._hash(r) == s._hash(r2)
r2.request.content = b""
assert s._hash(r) == s._hash(r2)
r2.request.content = None
assert s._hash(r) == s._hash(r2)
def test_load_with_server_replay_nopop(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(server_replay_nopop=True), [])
r = tflow.tflow(resp=True)
r.request.headers["key"] = "one"
def test_ignore_content_wins_over_params():
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
server_replay_ignore_content=True,
server_replay_ignore_payload_params=[
"param1", "param2"
]
),
[]
)
# NOTE: parameters are mutually exclusive in options
r2 = tflow.tflow(resp=True)
r2.request.headers["key"] = "two"
r = tflow.tflow(resp=True)
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = b"paramx=y"
s.load([r, r2])
r2 = tflow.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = b"paramx=x"
assert s.count() == 2
s.next_flow(r)
assert s.count() == 2
# same parameters
assert s._hash(r) == s._hash(r2)
def test_ignore_params(self):
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
server_replay_ignore_params=["param1", "param2"]
),
[]
def test_ignore_payload_params_other_content_type():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
tctx.configure(
s,
server_replay_ignore_content=False,
server_replay_ignore_payload_params=[
"param1", "param2"
]
)
r = tflow.tflow(resp=True)
r.request.path = "/test?param1=1"
r.request.headers["Content-Type"] = "application/json"
r.request.content = b'{"param1":"1"}'
r2 = tflow.tflow(resp=True)
r2.request.path = "/test"
r2.request.headers["Content-Type"] = "application/json"
r2.request.content = b'{"param1":"1"}'
# same content
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param1=2"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param2=1"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param3=2"
# distint content (note only x-www-form-urlencoded payload is analysed)
r2.request.content = b'{"param1":"2"}'
assert not s._hash(r) == s._hash(r2)
def test_ignore_payload_params(self):
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
server_replay_ignore_payload_params=["param1", "param2"]
),
[]
def test_hash():
s = serverplayback.ServerPlayback()
s.configure(options.Options(), [])
r = tflow.tflow()
r2 = tflow.tflow()
assert s._hash(r)
assert s._hash(r) == s._hash(r2)
r.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r.request.path = "voing"
assert s._hash(r) != s._hash(r2)
r.request.path = "path?blank_value"
r2.request.path = "path?"
assert s._hash(r) != s._hash(r2)
def test_headers():
s = serverplayback.ServerPlayback()
s.configure(options.Options(server_replay_use_headers=["foo"]), [])
r = tflow.tflow(resp=True)
r.request.headers["foo"] = "bar"
r2 = tflow.tflow(resp=True)
assert not s._hash(r) == s._hash(r2)
r2.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r2.request.headers["oink"] = "bar"
assert s._hash(r) == s._hash(r2)
r = tflow.tflow(resp=True)
r2 = tflow.tflow(resp=True)
assert s._hash(r) == s._hash(r2)
def test_load():
s = serverplayback.ServerPlayback()
s.configure(options.Options(), [])
r = tflow.tflow(resp=True)
r.request.headers["key"] = "one"
r2 = tflow.tflow(resp=True)
r2.request.headers["key"] = "two"
s.load([r, r2])
assert s.count() == 2
n = s.next_flow(r)
assert n.request.headers["key"] == "one"
assert s.count() == 1
n = s.next_flow(r)
assert n.request.headers["key"] == "two"
assert not s.flowmap
assert s.count() == 0
assert not s.next_flow(r)
def test_load_with_server_replay_nopop():
s = serverplayback.ServerPlayback()
s.configure(options.Options(server_replay_nopop=True), [])
r = tflow.tflow(resp=True)
r.request.headers["key"] = "one"
r2 = tflow.tflow(resp=True)
r2.request.headers["key"] = "two"
s.load([r, r2])
assert s.count() == 2
s.next_flow(r)
assert s.count() == 2
def test_ignore_params():
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
server_replay_ignore_params=["param1", "param2"]
),
[]
)
r = tflow.tflow(resp=True)
r.request.path = "/test?param1=1"
r2 = tflow.tflow(resp=True)
r2.request.path = "/test"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param1=2"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param2=1"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param3=2"
assert not s._hash(r) == s._hash(r2)
def test_ignore_payload_params():
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
server_replay_ignore_payload_params=["param1", "param2"]
),
[]
)
r = tflow.tflow(resp=True)
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = b"paramx=x&param1=1"
r2 = tflow.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = b"paramx=x&param1=1"
# same parameters
assert s._hash(r) == s._hash(r2)
# ignored parameters !=
r2.request.content = b"paramx=x&param1=2"
assert s._hash(r) == s._hash(r2)
# missing parameter
r2.request.content = b"paramx=x"
assert s._hash(r) == s._hash(r2)
# ignorable parameter added
r2.request.content = b"paramx=x&param1=2"
assert s._hash(r) == s._hash(r2)
# not ignorable parameter changed
r2.request.content = b"paramx=y&param1=1"
assert not s._hash(r) == s._hash(r2)
# not ignorable parameter missing
r2.request.content = b"param1=1"
assert not s._hash(r) == s._hash(r2)
def test_server_playback_full():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
tctx.configure(
s,
refresh_server_playback = True,
keepserving=False
)
r = tflow.tflow(resp=True)
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = b"paramx=x&param1=1"
r2 = tflow.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = b"paramx=x&param1=1"
# same parameters
assert s._hash(r) == s._hash(r2)
# ignored parameters !=
r2.request.content = b"paramx=x&param1=2"
assert s._hash(r) == s._hash(r2)
# missing parameter
r2.request.content = b"paramx=x"
assert s._hash(r) == s._hash(r2)
# ignorable parameter added
r2.request.content = b"paramx=x&param1=2"
assert s._hash(r) == s._hash(r2)
# not ignorable parameter changed
r2.request.content = b"paramx=y&param1=1"
assert not s._hash(r) == s._hash(r2)
# not ignorable parameter missing
r2.request.content = b"param1=1"
assert not s._hash(r) == s._hash(r2)
f = tflow.tflow()
f.response = mitmproxy.test.tutils.tresp(content=f.request.content)
s.load([f, f])
def test_server_playback_full(self):
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
tctx.configure(
s,
refresh_server_playback = True,
keepserving=False
)
tf = tflow.tflow()
assert not tf.response
s.request(tf)
assert tf.response == f.response
f = tflow.tflow()
f.response = mitmproxy.test.tutils.tresp(content=f.request.content)
s.load([f, f])
tf = tflow.tflow()
tf.request.content = b"gibble"
assert not tf.response
s.request(tf)
assert not tf.response
tf = tflow.tflow()
assert not tf.response
s.request(tf)
assert tf.response == f.response
assert not s.stop
s.tick()
assert not s.stop
tf = tflow.tflow()
tf.request.content = b"gibble"
assert not tf.response
s.request(tf)
assert not tf.response
tf = tflow.tflow()
s.request(tflow.tflow())
assert s.stop
assert not s.stop
s.tick()
assert not s.stop
tf = tflow.tflow()
s.request(tflow.tflow())
assert s.stop
def test_server_playback_kill():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
tctx.configure(
s,
refresh_server_playback = True,
replay_kill_extra=True
)
def test_server_playback_kill(self):
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
tctx.configure(
s,
refresh_server_playback = True,
replay_kill_extra=True
)
f = tflow.tflow()
f.response = mitmproxy.test.tutils.tresp(content=f.request.content)
s.load([f])
f = tflow.tflow()
f.response = mitmproxy.test.tutils.tresp(content=f.request.content)
s.load([f])
f = tflow.tflow()
f.request.host = "nonexistent"
tctx.cycle(s, f)
assert f.reply.value == exceptions.Kill
f = tflow.tflow()
f.request.host = "nonexistent"
tctx.cycle(s, f)
assert f.reply.value == exceptions.Kill

View File

@ -1,25 +1,27 @@
from mitmproxy.test import tflow
from mitmproxy.test import taddons
from mitmproxy.test import tutils
from .. import mastertest
from mitmproxy.addons import stickyauth
from mitmproxy import master
from mitmproxy import options
from mitmproxy import proxy
from mitmproxy import exceptions
class TestStickyAuth(mastertest.MasterTest):
def test_simple(self):
o = options.Options(stickyauth = ".*")
m = master.Master(o, proxy.DummyServer())
sa = stickyauth.StickyAuth()
m.addons.add(sa)
def test_configure():
r = stickyauth.StickyAuth()
with taddons.context() as tctx:
tctx.configure(r, stickyauth="~s")
tutils.raises(exceptions.OptionsError, tctx.configure, r, stickyauth="~~")
def test_simple():
r = stickyauth.StickyAuth()
with taddons.context():
f = tflow.tflow(resp=True)
f.request.headers["authorization"] = "foo"
m.request(f)
r.request(f)
assert "address" in sa.hosts
assert "address" in r.hosts
f = tflow.tflow(resp=True)
m.request(f)
r.request(f)
assert f.request.headers["authorization"] == "foo"