aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/mitmproxy/builtins/test_serverplayback.py284
-rw-r--r--test/mitmproxy/mastertest.py8
-rw-r--r--test/mitmproxy/test_dump.py2
-rw-r--r--test/mitmproxy/test_flow.py299
4 files changed, 292 insertions, 301 deletions
diff --git a/test/mitmproxy/builtins/test_serverplayback.py b/test/mitmproxy/builtins/test_serverplayback.py
new file mode 100644
index 00000000..72070c7a
--- /dev/null
+++ b/test/mitmproxy/builtins/test_serverplayback.py
@@ -0,0 +1,284 @@
+from .. import tutils, mastertest
+
+import netlib.tutils
+from mitmproxy.builtins import serverplayback
+from mitmproxy import options
+from mitmproxy import exceptions
+from mitmproxy import flow
+
+
+class TestServerPlayback:
+ def test_server_playback(self):
+ sp = serverplayback.ServerPlayback()
+ sp.configure(options.Options(), [])
+ f = tutils.tflow(resp=True)
+
+ 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(replay_ignore_host=True), [])
+
+ r = tutils.tflow(resp=True)
+ r2 = tutils.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(replay_ignore_content=False), [])
+
+ r = tutils.tflow(resp=True)
+ r2 = tutils.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(replay_ignore_content=True), [])
+ r = tutils.tflow(resp=True)
+ r2 = tutils.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(
+ replay_ignore_content=True,
+ replay_ignore_payload_params=[
+ "param1", "param2"
+ ]
+ ),
+ []
+ )
+ # NOTE: parameters are mutually exclusive in options
+
+ r = tutils.tflow(resp=True)
+ r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
+ r.request.content = b"paramx=y"
+
+ r2 = tutils.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()
+ s.configure(
+ options.Options(
+ replay_ignore_content=False,
+ replay_ignore_payload_params=[
+ "param1", "param2"
+ ]
+ ),
+ []
+
+ )
+ r = tutils.tflow(resp=True)
+ r.request.headers["Content-Type"] = "application/json"
+ r.request.content = b'{"param1":"1"}'
+ r2 = tutils.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(), [])
+
+ r = tutils.tflow()
+ r2 = tutils.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(self):
+ s = serverplayback.ServerPlayback()
+ s.configure(options.Options(rheaders=["foo"]), [])
+
+ r = tutils.tflow(resp=True)
+ r.request.headers["foo"] = "bar"
+ r2 = tutils.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 = tutils.tflow(resp=True)
+ r2 = tutils.tflow(resp=True)
+ assert s._hash(r) == s._hash(r2)
+
+ def test_load(self):
+ s = serverplayback.ServerPlayback()
+ s.configure(options.Options(), [])
+
+ r = tutils.tflow(resp=True)
+ r.request.headers["key"] = "one"
+
+ r2 = tutils.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_nopop(self):
+ s = serverplayback.ServerPlayback()
+ s.configure(options.Options(nopop=True), [])
+
+ r = tutils.tflow(resp=True)
+ r.request.headers["key"] = "one"
+
+ r2 = tutils.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(self):
+ s = serverplayback.ServerPlayback()
+ s.configure(
+ options.Options(
+ replay_ignore_params=["param1", "param2"]
+ ),
+ []
+ )
+
+ r = tutils.tflow(resp=True)
+ r.request.path = "/test?param1=1"
+ r2 = tutils.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(self):
+ s = serverplayback.ServerPlayback()
+ s.configure(
+ options.Options(
+ replay_ignore_payload_params=["param1", "param2"]
+ ),
+ []
+ )
+
+ r = tutils.tflow(resp=True)
+ r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
+ r.request.content = b"paramx=x&param1=1"
+ r2 = tutils.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(self):
+ state = flow.State()
+ s = serverplayback.ServerPlayback()
+ o = options.Options(refresh_server_playback = True, keepserving=False)
+ m = mastertest.RecordingMaster(o, None, state)
+ m.addons.add(o, s)
+
+ f = tutils.tflow()
+ f.response = netlib.tutils.tresp(content=f.request.content)
+ s.load([f, f])
+
+ tf = tutils.tflow()
+ assert not tf.response
+ m.request(tf)
+ assert tf.response == f.response
+
+ tf = tutils.tflow()
+ tf.request.content = b"gibble"
+ assert not tf.response
+ m.request(tf)
+ assert not tf.response
+
+ assert not s.stop
+ s.tick()
+ assert not s.stop
+
+ tf = tutils.tflow()
+ m.request(tutils.tflow())
+ assert s.stop
+
+ def test_server_playback_kill(self):
+ state = flow.State()
+ s = serverplayback.ServerPlayback()
+ o = options.Options(refresh_server_playback = True, kill=True)
+ m = mastertest.RecordingMaster(o, None, state)
+ m.addons.add(o, s)
+
+ f = tutils.tflow()
+ f.response = netlib.tutils.tresp(content=f.request.content)
+ s.load([f])
+
+ f = tutils.tflow()
+ f.request.host = "nonexistent"
+ m.request(f)
+ assert f.reply.value == exceptions.Kill
diff --git a/test/mitmproxy/mastertest.py b/test/mitmproxy/mastertest.py
index 08659d19..68d88ea1 100644
--- a/test/mitmproxy/mastertest.py
+++ b/test/mitmproxy/mastertest.py
@@ -5,6 +5,10 @@ from mitmproxy.flow import master
from mitmproxy import flow, proxy, models, controller
+class TestMaster:
+ pass
+
+
class MasterTest:
def cycle(self, master, content):
@@ -16,7 +20,9 @@ class MasterTest:
master.serverconnect(f.server_conn)
master.request(f)
if not f.error:
- f.response = models.HTTPResponse.wrap(netlib.tutils.tresp(content=content))
+ f.response = models.HTTPResponse.wrap(
+ netlib.tutils.tresp(content=content)
+ )
master.response(f)
master.clientdisconnect(f)
return f
diff --git a/test/mitmproxy/test_dump.py b/test/mitmproxy/test_dump.py
index 90f33264..40beeb0d 100644
--- a/test/mitmproxy/test_dump.py
+++ b/test/mitmproxy/test_dump.py
@@ -50,7 +50,7 @@ class TestDumpMaster(mastertest.MasterTest):
def test_replay(self):
o = dump.Options(server_replay=["nonexistent"], kill=True)
- tutils.raises(dump.DumpError, dump.DumpMaster, None, o)
+ tutils.raises(exceptions.OptionsError, dump.DumpMaster, None, o)
with tutils.tmpdir() as t:
p = os.path.join(t, "rep")
diff --git a/test/mitmproxy/test_flow.py b/test/mitmproxy/test_flow.py
index 1caeb100..91013efc 100644
--- a/test/mitmproxy/test_flow.py
+++ b/test/mitmproxy/test_flow.py
@@ -70,228 +70,6 @@ class TestClientPlaybackState:
assert not fm.client_playback
-class TestServerPlaybackState:
-
- def test_hash(self):
- s = flow.ServerPlaybackState(
- None,
- [],
- False,
- False,
- None,
- False,
- None,
- False)
- r = tutils.tflow()
- r2 = tutils.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(self):
- s = flow.ServerPlaybackState(
- ["foo"],
- [],
- False,
- False,
- None,
- False,
- None,
- False)
- r = tutils.tflow(resp=True)
- r.request.headers["foo"] = "bar"
- r2 = tutils.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 = tutils.tflow(resp=True)
- r2 = tutils.tflow(resp=True)
- assert s._hash(r) == s._hash(r2)
-
- def test_load(self):
- r = tutils.tflow(resp=True)
- r.request.headers["key"] = "one"
-
- r2 = tutils.tflow(resp=True)
- r2.request.headers["key"] = "two"
-
- s = flow.ServerPlaybackState(
- None, [
- r, r2], False, False, None, False, None, False)
- assert s.count() == 2
- assert len(s.fmap.keys()) == 1
-
- 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 s.count() == 0
-
- assert not s.next_flow(r)
-
- def test_load_with_nopop(self):
- r = tutils.tflow(resp=True)
- r.request.headers["key"] = "one"
-
- r2 = tutils.tflow(resp=True)
- r2.request.headers["key"] = "two"
-
- s = flow.ServerPlaybackState(
- None, [
- r, r2], False, True, None, False, None, False)
-
- assert s.count() == 2
- s.next_flow(r)
- assert s.count() == 2
-
- def test_ignore_params(self):
- s = flow.ServerPlaybackState(
- None, [], False, False, [
- "param1", "param2"], False, None, False)
- r = tutils.tflow(resp=True)
- r.request.path = "/test?param1=1"
- r2 = tutils.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(self):
- s = flow.ServerPlaybackState(
- None, [], False, False, None, False, [
- "param1", "param2"], False)
- r = tutils.tflow(resp=True)
- r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
- r.request.content = b"paramx=x&param1=1"
- r2 = tutils.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_ignore_payload_params_other_content_type(self):
- s = flow.ServerPlaybackState(
- None, [], False, False, None, False, [
- "param1", "param2"], False)
- r = tutils.tflow(resp=True)
- r.request.headers["Content-Type"] = "application/json"
- r.request.content = b'{"param1":"1"}'
- r2 = tutils.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_ignore_payload_wins_over_params(self):
- # NOTE: parameters are mutually exclusive in options
- s = flow.ServerPlaybackState(
- None, [], False, False, None, True, [
- "param1", "param2"], False)
- r = tutils.tflow(resp=True)
- r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
- r.request.content = b"paramx=y"
- r2 = tutils.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_content(self):
- s = flow.ServerPlaybackState(
- None,
- [],
- False,
- False,
- None,
- False,
- None,
- False)
- r = tutils.tflow(resp=True)
- r2 = tutils.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)
-
- # now ignoring content
- s = flow.ServerPlaybackState(
- None,
- [],
- False,
- False,
- None,
- True,
- None,
- False)
- r = tutils.tflow(resp=True)
- r2 = tutils.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_host(self):
- s = flow.ServerPlaybackState(
- None,
- [],
- False,
- False,
- None,
- False,
- None,
- True)
- r = tutils.tflow(resp=True)
- r2 = tutils.tflow(resp=True)
-
- r.request.host = "address"
- r2.request.host = "address"
- assert s._hash(r) == s._hash(r2)
- r2.request.host = "wrong_address"
- assert s._hash(r) == s._hash(r2)
-
-
class TestHTTPFlow(object):
def test_copy(self):
@@ -753,16 +531,6 @@ class TestFlowMaster:
DummyServer(ProxyConfig(options.Options())),
s
)
- assert not fm.start_server_playback(
- pb,
- False,
- [],
- False,
- False,
- None,
- False,
- None,
- False)
assert not fm.start_client_playback(pb, False)
fm.client_playback.testing = True
@@ -773,73 +541,6 @@ class TestFlowMaster:
f.error = Error("error")
fm.error(f)
- def test_server_playback(self):
- s = flow.State()
-
- f = tutils.tflow()
- f.response = HTTPResponse.wrap(netlib.tutils.tresp(content=f.request))
- pb = [f]
-
- fm = flow.FlowMaster(options.Options(), None, s)
- fm.refresh_server_playback = True
- assert not fm.do_server_playback(tutils.tflow())
-
- fm.start_server_playback(
- pb,
- False,
- [],
- False,
- False,
- None,
- False,
- None,
- False)
- assert fm.do_server_playback(tutils.tflow())
-
- fm.start_server_playback(
- pb,
- False,
- [],
- True,
- False,
- None,
- False,
- None,
- False)
- r = tutils.tflow()
- r.request.content = b"gibble"
- assert not fm.do_server_playback(r)
- assert fm.do_server_playback(tutils.tflow())
-
- fm.tick(0)
- assert fm.should_exit.is_set()
-
- fm.stop_server_playback()
- assert not fm.server_playback
-
- def test_server_playback_kill(self):
- s = flow.State()
- f = tutils.tflow()
- f.response = HTTPResponse.wrap(netlib.tutils.tresp(content=f.request))
- pb = [f]
- fm = flow.FlowMaster(None, None, s)
- fm.refresh_server_playback = True
- fm.start_server_playback(
- pb,
- True,
- [],
- False,
- False,
- None,
- False,
- None,
- False)
-
- f = tutils.tflow()
- f.request.host = "nonexistent"
- fm.request(f)
- assert f.reply.value == Kill
-
class TestRequest:
generated by cgit v1.2.3 (git 2.25.1) at 2025-11-04 03:53:36 +0000