aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/mitmproxy/addons/test_core.py11
-rw-r--r--test/mitmproxy/addons/test_view.py113
-rw-r--r--test/mitmproxy/test_command.py6
-rw-r--r--test/mitmproxy/tools/web/test_app.py10
4 files changed, 99 insertions, 41 deletions
diff --git a/test/mitmproxy/addons/test_core.py b/test/mitmproxy/addons/test_core.py
index 6ebf4ba9..39aac935 100644
--- a/test/mitmproxy/addons/test_core.py
+++ b/test/mitmproxy/addons/test_core.py
@@ -1,5 +1,6 @@
from mitmproxy.addons import core
from mitmproxy.test import taddons
+from mitmproxy.test import tflow
from mitmproxy import exceptions
import pytest
@@ -15,3 +16,13 @@ def test_set():
with pytest.raises(exceptions.CommandError):
tctx.command(sa.set, "nonexistent")
+
+
+def test_resume():
+ sa = core.Core()
+ with taddons.context():
+ f = tflow.tflow()
+ assert not sa.resume([f])
+ f.intercept()
+ sa.resume([f])
+ assert not f.reply.state == "taken"
diff --git a/test/mitmproxy/addons/test_view.py b/test/mitmproxy/addons/test_view.py
index 05d4af30..979f0aa1 100644
--- a/test/mitmproxy/addons/test_view.py
+++ b/test/mitmproxy/addons/test_view.py
@@ -4,7 +4,6 @@ from mitmproxy.test import tflow
from mitmproxy.addons import view
from mitmproxy import flowfilter
-from mitmproxy import options
from mitmproxy import exceptions
from mitmproxy.test import taddons
@@ -26,12 +25,12 @@ def test_order_refresh():
v.sig_view_refresh.connect(save)
tf = tflow.tflow(resp=True)
- with taddons.context(options=options.Options()) as tctx:
+ with taddons.context() as tctx:
tctx.configure(v, console_order="time")
- v.add(tf)
+ v.add([tf])
tf.request.timestamp_start = 1
assert not sargs
- v.update(tf)
+ v.update([tf])
assert sargs
@@ -133,13 +132,14 @@ def test_filter():
def test_load():
v = view.View()
- with taddons.context(options=options.Options()) as tctx:
+ with taddons.context() as tctx:
tctx.master.addons.add(v)
def test_resolve():
v = view.View()
- with taddons.context(options=options.Options()) as tctx:
+ with taddons.context() as tctx:
+ assert tctx.command(v.resolve, "@all") == []
assert tctx.command(v.resolve, "@focus") == []
assert tctx.command(v.resolve, "@shown") == []
assert tctx.command(v.resolve, "@hidden") == []
@@ -149,6 +149,7 @@ def test_resolve():
v.request(tft(method="get"))
assert len(tctx.command(v.resolve, "~m get")) == 1
assert len(tctx.command(v.resolve, "@focus")) == 1
+ assert len(tctx.command(v.resolve, "@all")) == 1
assert len(tctx.command(v.resolve, "@shown")) == 1
assert len(tctx.command(v.resolve, "@unmarked")) == 1
assert tctx.command(v.resolve, "@hidden") == []
@@ -156,6 +157,7 @@ def test_resolve():
v.request(tft(method="put"))
assert len(tctx.command(v.resolve, "@focus")) == 1
assert len(tctx.command(v.resolve, "@shown")) == 2
+ assert len(tctx.command(v.resolve, "@all")) == 2
assert tctx.command(v.resolve, "@hidden") == []
assert tctx.command(v.resolve, "@marked") == []
@@ -175,14 +177,52 @@ def test_resolve():
assert m(tctx.command(v.resolve, "@hidden")) == ["PUT", "PUT"]
assert m(tctx.command(v.resolve, "@marked")) == ["GET"]
assert m(tctx.command(v.resolve, "@unmarked")) == ["PUT", "GET", "PUT"]
+ assert m(tctx.command(v.resolve, "@all")) == ["GET", "PUT", "GET", "PUT"]
with pytest.raises(exceptions.CommandError, match="Invalid flow filter"):
tctx.command(v.resolve, "~")
+def test_go():
+ v = view.View()
+ with taddons.context():
+ v.add([
+ tflow.tflow(),
+ tflow.tflow(),
+ tflow.tflow(),
+ tflow.tflow(),
+ tflow.tflow(),
+ ])
+ assert v.focus.index == 0
+ v.go(-1)
+ assert v.focus.index == 4
+ v.go(0)
+ assert v.focus.index == 0
+ v.go(1)
+ assert v.focus.index == 1
+ v.go(999)
+ assert v.focus.index == 4
+ v.go(-999)
+ assert v.focus.index == 0
+
+
+def test_duplicate():
+ v = view.View()
+ with taddons.context():
+ f = [
+ tflow.tflow(),
+ tflow.tflow(),
+ ]
+ v.add(f)
+ assert len(v) == 2
+ v.duplicate(f)
+ assert len(v) == 4
+ assert v.focus.index == 2
+
+
def test_order():
v = view.View()
- with taddons.context(options=options.Options()) as tctx:
+ with taddons.context() as tctx:
v.request(tft(method="get", start=1))
v.request(tft(method="put", start=2))
v.request(tft(method="get", start=3))
@@ -230,14 +270,14 @@ def test_update():
assert f in v
f.request.method = "put"
- v.update(f)
+ v.update([f])
assert f not in v
f.request.method = "get"
- v.update(f)
+ v.update([f])
assert f in v
- v.update(f)
+ v.update([f])
assert f in v
@@ -276,7 +316,7 @@ def test_signals():
assert not any([rec_add, rec_update, rec_remove, rec_refresh])
# Simple add
- v.add(tft())
+ v.add([tft()])
assert rec_add
assert not any([rec_update, rec_remove, rec_refresh])
@@ -291,14 +331,14 @@ def test_signals():
# An update that results in a flow being added to the view
clearrec()
v[0].request.method = "PUT"
- v.update(v[0])
+ v.update([v[0]])
assert rec_remove
assert not any([rec_update, rec_refresh, rec_add])
# An update that does not affect the view just sends update
v.set_filter(flowfilter.parse("~m put"))
clearrec()
- v.update(v[0])
+ v.update([v[0]])
assert rec_update
assert not any([rec_remove, rec_refresh, rec_add])
@@ -307,33 +347,33 @@ def test_signals():
v.set_filter(flowfilter.parse("~m get"))
assert not len(v)
clearrec()
- v.update(f)
+ v.update([f])
assert not any([rec_add, rec_update, rec_remove, rec_refresh])
def test_focus_follow():
v = view.View()
- with taddons.context(options=options.Options()) as tctx:
+ with taddons.context() as tctx:
tctx.configure(v, console_focus_follow=True, view_filter="~m get")
- v.add(tft(start=5))
+ v.add([tft(start=5)])
assert v.focus.index == 0
- v.add(tft(start=4))
+ v.add([tft(start=4)])
assert v.focus.index == 0
assert v.focus.flow.request.timestamp_start == 4
- v.add(tft(start=7))
+ v.add([tft(start=7)])
assert v.focus.index == 2
assert v.focus.flow.request.timestamp_start == 7
mod = tft(method="put", start=6)
- v.add(mod)
+ v.add([mod])
assert v.focus.index == 2
assert v.focus.flow.request.timestamp_start == 7
mod.request.method = "GET"
- v.update(mod)
+ v.update([mod])
assert v.focus.index == 2
assert v.focus.flow.request.timestamp_start == 6
@@ -341,7 +381,7 @@ def test_focus_follow():
def test_focus():
# Special case - initialising with a view that already contains data
v = view.View()
- v.add(tft())
+ v.add([tft()])
f = view.Focus(v)
assert f.index is 0
assert f.flow is v[0]
@@ -352,7 +392,7 @@ def test_focus():
assert f.index is None
assert f.flow is None
- v.add(tft(start=1))
+ v.add([tft(start=1)])
assert f.index == 0
assert f.flow is v[0]
@@ -362,11 +402,11 @@ def test_focus():
with pytest.raises(ValueError):
f.__setattr__("index", 99)
- v.add(tft(start=0))
+ v.add([tft(start=0)])
assert f.index == 1
assert f.flow is v[1]
- v.add(tft(start=2))
+ v.add([tft(start=2)])
assert f.index == 1
assert f.flow is v[1]
@@ -374,22 +414,25 @@ def test_focus():
assert f.index == 0
f.index = 1
- v.remove(v[1])
+ v.remove([v[1]])
+ v[1].intercept()
assert f.index == 1
assert f.flow is v[1]
- v.remove(v[1])
+ v.remove([v[1]])
assert f.index == 0
assert f.flow is v[0]
- v.remove(v[0])
+ v.remove([v[0]])
assert f.index is None
assert f.flow is None
- v.add(tft(method="get", start=0))
- v.add(tft(method="get", start=1))
- v.add(tft(method="put", start=2))
- v.add(tft(method="get", start=3))
+ v.add([
+ tft(method="get", start=0),
+ tft(method="get", start=1),
+ tft(method="put", start=2),
+ tft(method="get", start=3),
+ ])
f.flow = v[2]
assert f.flow.request.method == "PUT"
@@ -409,16 +452,16 @@ def test_settings():
with pytest.raises(KeyError):
v.settings[f]
- v.add(f)
+ v.add([f])
v.settings[f]["foo"] = "bar"
assert v.settings[f]["foo"] == "bar"
assert len(list(v.settings)) == 1
- v.remove(f)
+ v.remove([f])
with pytest.raises(KeyError):
v.settings[f]
assert not v.settings.keys()
- v.add(f)
+ v.add([f])
v.settings[f]["foo"] = "bar"
assert v.settings.keys()
v.clear()
@@ -427,7 +470,7 @@ def test_settings():
def test_configure():
v = view.View()
- with taddons.context(options=options.Options()) as tctx:
+ with taddons.context() as tctx:
tctx.configure(v, view_filter="~q")
with pytest.raises(Exception, match="Invalid interception filter"):
tctx.configure(v, view_filter="~~")
diff --git a/test/mitmproxy/test_command.py b/test/mitmproxy/test_command.py
index 64928dbf..306650c7 100644
--- a/test/mitmproxy/test_command.py
+++ b/test/mitmproxy/test_command.py
@@ -65,7 +65,7 @@ def test_typename():
class DummyConsole:
def load(self, l):
- l.add_command("console.resolve", self.resolve)
+ l.add_command("view.resolve", self.resolve)
def resolve(self, spec: str) -> typing.Sequence[flow.Flow]:
n = int(spec)
@@ -76,6 +76,10 @@ def test_parsearg():
with taddons.context() as tctx:
tctx.master.addons.add(DummyConsole())
assert command.parsearg(tctx.master.commands, "foo", str) == "foo"
+ assert command.parsearg(tctx.master.commands, "1", int) == 1
+ with pytest.raises(exceptions.CommandError):
+ command.parsearg(tctx.master.commands, "foo", int)
+
assert len(command.parsearg(
tctx.master.commands, "2", typing.Sequence[flow.Flow]
)) == 2
diff --git a/test/mitmproxy/tools/web/test_app.py b/test/mitmproxy/tools/web/test_app.py
index e3d5dc44..2b6181d3 100644
--- a/test/mitmproxy/tools/web/test_app.py
+++ b/test/mitmproxy/tools/web/test_app.py
@@ -23,8 +23,8 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
m = webmaster.WebMaster(o, proxy.DummyServer(), with_termlog=False)
f = tflow.tflow(resp=True)
f.id = "42"
- m.view.add(f)
- m.view.add(tflow.tflow(err=True))
+ m.view.add([f])
+ m.view.add([tflow.tflow(err=True)])
m.add_log("test log", "info")
self.master = m
self.view = m.view
@@ -78,7 +78,7 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
# restore
for f in flows:
- self.view.add(f)
+ self.view.add([f])
self.events.data = events
def test_resume(self):
@@ -110,7 +110,7 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
assert self.fetch("/flows/42", method="DELETE").code == 200
assert not self.view.get_by_id("42")
- self.view.add(f)
+ self.view.add([f])
assert self.fetch("/flows/1234", method="DELETE").code == 404
@@ -162,7 +162,7 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
f = self.view.get_by_id(resp.body.decode())
assert f
assert f.id != "42"
- self.view.remove(f)
+ self.view.remove([f])
def test_flow_revert(self):
f = self.view.get_by_id("42")
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-25 14:25:51 +0000