aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/examples/_test_har_dump.py114
-rw-r--r--test/examples/test_har_dump.py86
-rw-r--r--test/mitmproxy/addons/test_dumper.py1
-rw-r--r--test/mitmproxy/addons/test_readfile.py142
-rw-r--r--test/mitmproxy/addons/test_readstdin.py53
-rw-r--r--test/mitmproxy/addons/test_stickycookie.py4
-rw-r--r--test/mitmproxy/data/test_flow_export/locust_task_post.py2
-rw-r--r--test/mitmproxy/data/test_flow_export/python_post.py13
-rw-r--r--test/mitmproxy/net/http/test_cookies.py4
-rw-r--r--test/mitmproxy/test_addonmanager.py7
-rw-r--r--test/mitmproxy/test_export.py145
-rw-r--r--test/mitmproxy/test_optmanager.py6
-rw-r--r--test/mitmproxy/utils/test_typecheck.py6
13 files changed, 295 insertions, 288 deletions
diff --git a/test/examples/_test_har_dump.py b/test/examples/_test_har_dump.py
deleted file mode 100644
index e5cfd2e1..00000000
--- a/test/examples/_test_har_dump.py
+++ /dev/null
@@ -1,114 +0,0 @@
-import json
-import shlex
-import pytest
-
-from mitmproxy import options
-from mitmproxy import proxy
-from mitmproxy import master
-from mitmproxy.addons import script
-
-from mitmproxy.test import tflow
-from mitmproxy.test import tutils
-from mitmproxy.net.http import cookies
-
-example_dir = tutils.test_data.push("../examples")
-
-
-class ScriptError(Exception):
- pass
-
-
-class RaiseMaster(master.Master):
- def add_log(self, e, level):
- if level in ("warn", "error"):
- raise ScriptError(e)
-
-
-def tscript(cmd, args=""):
- o = options.Options()
- cmd = example_dir.path(cmd) + " " + args
- m = RaiseMaster(o, proxy.DummyServer())
- sc = script.Script(cmd)
- m.addons.add(sc)
- return m, sc
-
-
-class TestHARDump:
-
- def flow(self, resp_content=b'message'):
- times = dict(
- timestamp_start=746203272,
- timestamp_end=746203272,
- )
-
- # Create a dummy flow for testing
- return tflow.tflow(
- req=tutils.treq(method=b'GET', **times),
- resp=tutils.tresp(content=resp_content, **times)
- )
-
- def test_no_file_arg(self):
- with pytest.raises(ScriptError):
- tscript("complex/har_dump.py")
-
- def test_simple(self, tmpdir):
- path = str(tmpdir.join("somefile"))
-
- m, sc = tscript("complex/har_dump.py", shlex.quote(path))
- m.addons.trigger("response", self.flow())
- m.addons.remove(sc)
-
- with open(path, "r") as inp:
- har = json.load(inp)
- assert len(har["log"]["entries"]) == 1
-
- def test_base64(self, tmpdir):
- path = str(tmpdir.join("somefile"))
-
- m, sc = tscript("complex/har_dump.py", shlex.quote(path))
- m.addons.trigger(
- "response", self.flow(resp_content=b"foo" + b"\xFF" * 10)
- )
- m.addons.remove(sc)
-
- with open(path, "r") as inp:
- har = json.load(inp)
- assert har["log"]["entries"][0]["response"]["content"]["encoding"] == "base64"
-
- def test_format_cookies(self):
- m, sc = tscript("complex/har_dump.py", "-")
- format_cookies = sc.ns.format_cookies
-
- CA = cookies.CookieAttrs
-
- f = format_cookies([("n", "v", CA([("k", "v")]))])[0]
- assert f['name'] == "n"
- assert f['value'] == "v"
- assert not f['httpOnly']
- assert not f['secure']
-
- f = format_cookies([("n", "v", CA([("httponly", None), ("secure", None)]))])[0]
- assert f['httpOnly']
- assert f['secure']
-
- f = format_cookies([("n", "v", CA([("expires", "Mon, 24-Aug-2037 00:00:00 GMT")]))])[0]
- assert f['expires']
-
- def test_binary(self, tmpdir):
-
- f = self.flow()
- f.request.method = "POST"
- f.request.headers["content-type"] = "application/x-www-form-urlencoded"
- f.request.content = b"foo=bar&baz=s%c3%bc%c3%9f"
- f.response.headers["random-junk"] = bytes(range(256))
- f.response.content = bytes(range(256))
-
- path = str(tmpdir.join("somefile"))
-
- m, sc = tscript("complex/har_dump.py", shlex.quote(path))
- m.addons.trigger("response", f)
- m.addons.remove(sc)
-
- with open(path, "r") as inp:
- har = json.load(inp)
- assert len(har["log"]["entries"]) == 1
diff --git a/test/examples/test_har_dump.py b/test/examples/test_har_dump.py
new file mode 100644
index 00000000..11cd5c29
--- /dev/null
+++ b/test/examples/test_har_dump.py
@@ -0,0 +1,86 @@
+import json
+
+from mitmproxy.test import tflow
+from mitmproxy.test import tutils
+from mitmproxy.test import taddons
+from mitmproxy.net.http import cookies
+
+example_dir = tutils.test_data.push("../examples")
+
+
+class TestHARDump:
+ def flow(self, resp_content=b'message'):
+ times = dict(
+ timestamp_start=746203272,
+ timestamp_end=746203272,
+ )
+
+ # Create a dummy flow for testing
+ return tflow.tflow(
+ req=tutils.treq(method=b'GET', **times),
+ resp=tutils.tresp(content=resp_content, **times)
+ )
+
+ def test_simple(self, tmpdir):
+ with taddons.context() as tctx:
+ a = tctx.script(example_dir.path("complex/har_dump.py"))
+ path = str(tmpdir.join("somefile"))
+ tctx.configure(a, hardump=path)
+ tctx.invoke(a, "response", self.flow())
+ tctx.invoke(a, "done")
+ with open(path, "r") as inp:
+ har = json.load(inp)
+ assert len(har["log"]["entries"]) == 1
+
+ def test_base64(self, tmpdir):
+ with taddons.context() as tctx:
+ a = tctx.script(example_dir.path("complex/har_dump.py"))
+ path = str(tmpdir.join("somefile"))
+ tctx.configure(a, hardump=path)
+
+ tctx.invoke(
+ a, "response", self.flow(resp_content=b"foo" + b"\xFF" * 10)
+ )
+ tctx.invoke(a, "done")
+ with open(path, "r") as inp:
+ har = json.load(inp)
+ assert har["log"]["entries"][0]["response"]["content"]["encoding"] == "base64"
+
+ def test_format_cookies(self):
+ with taddons.context() as tctx:
+ a = tctx.script(example_dir.path("complex/har_dump.py"))
+
+ CA = cookies.CookieAttrs
+
+ f = a.format_cookies([("n", "v", CA([("k", "v")]))])[0]
+ assert f['name'] == "n"
+ assert f['value'] == "v"
+ assert not f['httpOnly']
+ assert not f['secure']
+
+ f = a.format_cookies([("n", "v", CA([("httponly", None), ("secure", None)]))])[0]
+ assert f['httpOnly']
+ assert f['secure']
+
+ f = a.format_cookies([("n", "v", CA([("expires", "Mon, 24-Aug-2037 00:00:00 GMT")]))])[0]
+ assert f['expires']
+
+ def test_binary(self, tmpdir):
+ with taddons.context() as tctx:
+ a = tctx.script(example_dir.path("complex/har_dump.py"))
+ path = str(tmpdir.join("somefile"))
+ tctx.configure(a, hardump=path)
+
+ f = self.flow()
+ f.request.method = "POST"
+ f.request.headers["content-type"] = "application/x-www-form-urlencoded"
+ f.request.content = b"foo=bar&baz=s%c3%bc%c3%9f"
+ f.response.headers["random-junk"] = bytes(range(256))
+ f.response.content = bytes(range(256))
+
+ tctx.invoke(a, "response", f)
+ tctx.invoke(a, "done")
+
+ with open(path, "r") as inp:
+ har = json.load(inp)
+ assert len(har["log"]["entries"]) == 1
diff --git a/test/mitmproxy/addons/test_dumper.py b/test/mitmproxy/addons/test_dumper.py
index d2cefe79..d8aa593b 100644
--- a/test/mitmproxy/addons/test_dumper.py
+++ b/test/mitmproxy/addons/test_dumper.py
@@ -68,7 +68,6 @@ def test_simple():
ctx.configure(d, flow_detail=4)
flow = tflow.tflow()
flow.request = tutils.treq()
- flow.request.stickycookie = True
flow.client_conn = mock.MagicMock()
flow.client_conn.address[0] = "foo"
flow.response = tutils.tresp(content=None)
diff --git a/test/mitmproxy/addons/test_readfile.py b/test/mitmproxy/addons/test_readfile.py
index b30c147b..813aa10e 100644
--- a/test/mitmproxy/addons/test_readfile.py
+++ b/test/mitmproxy/addons/test_readfile.py
@@ -1,62 +1,104 @@
+import io
+from unittest import mock
+
+import pytest
+
+import mitmproxy.io
+from mitmproxy import exceptions
from mitmproxy.addons import readfile
from mitmproxy.test import taddons
from mitmproxy.test import tflow
-from mitmproxy import io
-from mitmproxy import exceptions
-from unittest import mock
-import pytest
+@pytest.fixture
+def data():
+ f = io.BytesIO()
+
+ w = mitmproxy.io.FlowWriter(f)
+ flows = [
+ tflow.tflow(resp=True),
+ tflow.tflow(err=True),
+ tflow.ttcpflow(),
+ tflow.ttcpflow(err=True)
+ ]
+ for flow in flows:
+ w.add(flow)
-def write_data(path, corrupt=False):
- with open(path, "wb") as tf:
- w = io.FlowWriter(tf)
- for i in range(3):
- f = tflow.tflow(resp=True)
- w.add(f)
- for i in range(3):
- f = tflow.tflow(err=True)
- w.add(f)
- f = tflow.ttcpflow()
- w.add(f)
- f = tflow.ttcpflow(err=True)
- w.add(f)
- if corrupt:
- tf.write(b"flibble")
-
-
-@mock.patch('mitmproxy.master.Master.load_flow')
-def test_configure(mck, tmpdir):
-
- rf = readfile.ReadFile()
- with taddons.context() as tctx:
- tf = str(tmpdir.join("tfile"))
- write_data(tf)
- tctx.configure(rf, rfile=str(tf))
- assert not mck.called
- rf.running()
- assert mck.called
-
- write_data(tf, corrupt=True)
- tctx.configure(rf, rfile=str(tf))
- with pytest.raises(exceptions.OptionsError):
+ f.seek(0)
+ return f
+
+
+@pytest.fixture
+def corrupt_data():
+ f = data()
+ f.seek(0, io.SEEK_END)
+ f.write(b"qibble")
+ f.seek(0)
+ return f
+
+
+class TestReadFile:
+ @mock.patch('mitmproxy.master.Master.load_flow')
+ def test_configure(self, mck, tmpdir, data, corrupt_data):
+ rf = readfile.ReadFile()
+ with taddons.context() as tctx:
+ tf = tmpdir.join("tfile")
+
+ tf.write(data.getvalue())
+ tctx.configure(rf, rfile=str(tf))
+ assert not mck.called
rf.running()
+ assert mck.called
+ tf.write(corrupt_data.getvalue())
+ tctx.configure(rf, rfile=str(tf))
+ with pytest.raises(exceptions.OptionsError):
+ rf.running()
-@mock.patch('mitmproxy.master.Master.load_flow')
-def test_corruption(mck, tmpdir):
+ @mock.patch('mitmproxy.master.Master.load_flow')
+ def test_corrupt(self, mck, corrupt_data):
+ rf = readfile.ReadFile()
+ with taddons.context() as tctx:
+ with pytest.raises(exceptions.FlowReadException):
+ rf.load_flows(io.BytesIO(b"qibble"))
+ assert not mck.called
+ assert len(tctx.master.logs) == 1
- rf = readfile.ReadFile()
- with taddons.context() as tctx:
- with pytest.raises(exceptions.FlowReadException):
- rf.load_flows_file("nonexistent")
- assert not mck.called
- assert len(tctx.master.logs) == 1
+ with pytest.raises(exceptions.FlowReadException):
+ rf.load_flows(corrupt_data)
+ assert mck.called
+ assert len(tctx.master.logs) == 2
+
+ def test_nonexisting_file(self):
+ rf = readfile.ReadFile()
+ with taddons.context() as tctx:
+ with pytest.raises(exceptions.FlowReadException):
+ rf.load_flows_from_path("nonexistent")
+ assert len(tctx.master.logs) == 1
+
+
+class TestReadFileStdin:
+ @mock.patch('mitmproxy.master.Master.load_flow')
+ @mock.patch('sys.stdin')
+ def test_stdin(self, stdin, load_flow, data, corrupt_data):
+ rf = readfile.ReadFileStdin()
+ with taddons.context() as tctx:
+ stdin.buffer = data
+ tctx.configure(rf, rfile="-")
+ assert not load_flow.called
+ rf.running()
+ assert load_flow.called
- tfc = str(tmpdir.join("tfile"))
- write_data(tfc, corrupt=True)
+ stdin.buffer = corrupt_data
+ tctx.configure(rf, rfile="-")
+ with pytest.raises(exceptions.OptionsError):
+ rf.running()
- with pytest.raises(exceptions.FlowReadException):
- rf.load_flows_file(tfc)
- assert mck.called
- assert len(tctx.master.logs) == 2
+ @mock.patch('mitmproxy.master.Master.load_flow')
+ def test_normal(self, load_flow, tmpdir, data):
+ rf = readfile.ReadFileStdin()
+ with taddons.context():
+ tfile = tmpdir.join("tfile")
+ tfile.write(data.getvalue())
+ rf.load_flows_from_path(str(tfile))
+ assert load_flow.called
diff --git a/test/mitmproxy/addons/test_readstdin.py b/test/mitmproxy/addons/test_readstdin.py
deleted file mode 100644
index 76b01f4f..00000000
--- a/test/mitmproxy/addons/test_readstdin.py
+++ /dev/null
@@ -1,53 +0,0 @@
-
-import io
-from mitmproxy.addons import readstdin
-from mitmproxy.test import taddons
-from mitmproxy.test import tflow
-import mitmproxy.io
-from unittest import mock
-
-
-def gen_data(corrupt=False):
- tf = io.BytesIO()
- w = mitmproxy.io.FlowWriter(tf)
- for i in range(3):
- f = tflow.tflow(resp=True)
- w.add(f)
- for i in range(3):
- f = tflow.tflow(err=True)
- w.add(f)
- f = tflow.ttcpflow()
- w.add(f)
- f = tflow.ttcpflow(err=True)
- w.add(f)
- if corrupt:
- tf.write(b"flibble")
- tf.seek(0)
- return tf
-
-
-class mStdin:
- def __init__(self, d):
- self.buffer = d
-
- def isatty(self):
- return False
-
-
-@mock.patch('mitmproxy.master.Master.load_flow')
-def test_read(m, tmpdir):
- rf = readstdin.ReadStdin()
- with taddons.context() as tctx:
- assert not m.called
- rf.running(stdin=mStdin(gen_data()))
- assert m.called
-
- rf.running(stdin=mStdin(None))
- assert tctx.master.logs
- tctx.master.clear()
-
- m.reset_mock()
- assert not m.called
- rf.running(stdin=mStdin(gen_data(corrupt=True)))
- assert m.called
- assert tctx.master.logs
diff --git a/test/mitmproxy/addons/test_stickycookie.py b/test/mitmproxy/addons/test_stickycookie.py
index 9092e09b..f77d019d 100644
--- a/test/mitmproxy/addons/test_stickycookie.py
+++ b/test/mitmproxy/addons/test_stickycookie.py
@@ -110,8 +110,8 @@ class TestStickyCookie:
f.response.headers["Set-Cookie"] = c2
sc.response(f)
googlekey = list(sc.jar.keys())[0]
- assert len(sc.jar[googlekey].keys()) == 1
- assert list(sc.jar[googlekey]["somecookie"].items())[0][1] == "newvalue"
+ assert len(sc.jar[googlekey]) == 1
+ assert sc.jar[googlekey]["somecookie"] == "newvalue"
def test_response_delete(self):
sc = stickycookie.StickyCookie()
diff --git a/test/mitmproxy/data/test_flow_export/locust_task_post.py b/test/mitmproxy/data/test_flow_export/locust_task_post.py
index 989df455..a5f307ee 100644
--- a/test/mitmproxy/data/test_flow_export/locust_task_post.py
+++ b/test/mitmproxy/data/test_flow_export/locust_task_post.py
@@ -2,7 +2,7 @@
def path(self):
url = self.locust.host + '/path'
- data = '''content'''
+ data = '''\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\x0c\r\x0e\x0f\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f !"#$%&\'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~\x7f\x80\x81\x82\x83\x84\x85\x86\x87\x88\x89\x8a\x8b\x8c\x8d\x8e\x8f\x90\x91\x92\x93\x94\x95\x96\x97\x98\x99\x9a\x9b\x9c\x9d\x9e\x9f\xa0\xa1\xa2\xa3\xa4\xa5\xa6\xa7\xa8\xa9\xaa\xab\xac\xad\xae\xaf\xb0\xb1\xb2\xb3\xb4\xb5\xb6\xb7\xb8\xb9\xba\xbb\xbc\xbd\xbe\xbf\xc0\xc1\xc2\xc3\xc4\xc5\xc6\xc7\xc8\xc9\xca\xcb\xcc\xcd\xce\xcf\xd0\xd1\xd2\xd3\xd4\xd5\xd6\xd7\xd8\xd9\xda\xdb\xdc\xdd\xde\xdf\xe0\xe1\xe2\xe3\xe4\xe5\xe6\xe7\xe8\xe9\xea\xeb\xec\xed\xee\xef\xf0\xf1\xf2\xf3\xf4\xf5\xf6\xf7\xf8\xf9\xfa\xfb\xfc\xfd\xfe\xff'''
self.response = self.client.request(
method='POST',
diff --git a/test/mitmproxy/data/test_flow_export/python_post.py b/test/mitmproxy/data/test_flow_export/python_post.py
index 6254adfb..42f1af9a 100644
--- a/test/mitmproxy/data/test_flow_export/python_post.py
+++ b/test/mitmproxy/data/test_flow_export/python_post.py
@@ -2,7 +2,16 @@ import requests
response = requests.post(
'http://address:22/path',
- data=b'content'
+ data=(b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\x0c\r\x0e\x0f\x10\x11\x12\x13'
+ b'\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f !"#$%&\'()*+,-./01234567'
+ b'89:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~\x7f'
+ b'\x80\x81\x82\x83\x84\x85\x86\x87\x88\x89\x8a\x8b\x8c\x8d\x8e\x8f'
+ b'\x90\x91\x92\x93\x94\x95\x96\x97\x98\x99\x9a\x9b\x9c\x9d\x9e\x9f'
+ b'\xa0\xa1\xa2\xa3\xa4\xa5\xa6\xa7\xa8\xa9\xaa\xab\xac\xad\xae\xaf'
+ b'\xb0\xb1\xb2\xb3\xb4\xb5\xb6\xb7\xb8\xb9\xba\xbb\xbc\xbd\xbe\xbf'
+ b'\xc0\xc1\xc2\xc3\xc4\xc5\xc6\xc7\xc8\xc9\xca\xcb\xcc\xcd\xce\xcf'
+ b'\xd0\xd1\xd2\xd3\xd4\xd5\xd6\xd7\xd8\xd9\xda\xdb\xdc\xdd\xde\xdf'
+ b'\xe0\xe1\xe2\xe3\xe4\xe5\xe6\xe7\xe8\xe9\xea\xeb\xec\xed\xee\xef'
+ b'\xf0\xf1\xf2\xf3\xf4\xf5\xf6\xf7\xf8\xf9\xfa\xfb\xfc\xfd\xfe\xff')
)
-
print(response.text)
diff --git a/test/mitmproxy/net/http/test_cookies.py b/test/mitmproxy/net/http/test_cookies.py
index 5c30dbdb..680a5033 100644
--- a/test/mitmproxy/net/http/test_cookies.py
+++ b/test/mitmproxy/net/http/test_cookies.py
@@ -283,6 +283,10 @@ def test_refresh_cookie():
c = "foo/bar=bla"
assert cookies.refresh_set_cookie_header(c, 0)
+ # https://github.com/mitmproxy/mitmproxy/issues/2250
+ c = ""
+ assert cookies.refresh_set_cookie_header(c, 60) == ""
+
@mock.patch('time.time')
def test_get_expiration_ts(*args):
diff --git a/test/mitmproxy/test_addonmanager.py b/test/mitmproxy/test_addonmanager.py
index cba40412..7b461580 100644
--- a/test/mitmproxy/test_addonmanager.py
+++ b/test/mitmproxy/test_addonmanager.py
@@ -76,6 +76,13 @@ def test_defaults():
assert addons.default_addons()
+def test_loader():
+ with taddons.context() as tctx:
+ l = addonmanager.Loader(tctx.master)
+ l.add_option("custom_option", bool, False, "help")
+ l.add_option("custom_option", bool, False, "help")
+
+
def test_simple():
with taddons.context() as tctx:
a = tctx.master.addons
diff --git a/test/mitmproxy/test_export.py b/test/mitmproxy/test_export.py
index 457d8836..b789e6b5 100644
--- a/test/mitmproxy/test_export.py
+++ b/test/mitmproxy/test_export.py
@@ -1,13 +1,15 @@
-from mitmproxy.test import tflow
import re
-from mitmproxy.net.http import Headers
+import pytest
+
from mitmproxy import export # heh
+from mitmproxy.net.http import Headers
+from mitmproxy.test import tflow
from mitmproxy.test import tutils
def clean_blanks(s):
- return re.sub(r"^(\s+)$", "", s, flags=re.MULTILINE)
+ return re.sub(r"^\s+", "", s, flags=re.MULTILINE)
def python_equals(testdata, text):
@@ -19,85 +21,110 @@ def python_equals(testdata, text):
assert clean_blanks(text).rstrip() == clean_blanks(d).rstrip()
-def req_get():
- return tutils.treq(method=b'GET', content=b'', path=b"/path?a=foo&a=bar&b=baz")
+@pytest.fixture
+def get_request():
+ return tflow.tflow(
+ req=tutils.treq(
+ method=b'GET',
+ content=b'',
+ path=b"/path?a=foo&a=bar&b=baz"
+ )
+ )
+
+
+@pytest.fixture
+def post_request():
+ return tflow.tflow(
+ req=tutils.treq(
+ method=b'POST',
+ headers=(),
+ content=bytes(range(256))
+ )
+ )
+
+
+@pytest.fixture
+def patch_request():
+ return tflow.tflow(
+ req=tutils.treq(method=b'PATCH', path=b"/path?query=param")
+ )
-def req_post():
- return tutils.treq(method=b'POST', headers=())
+class TExport:
+ def test_get(self, get_request):
+ raise NotImplementedError()
+ def test_post(self, post_request):
+ raise NotImplementedError()
-def req_patch():
- return tutils.treq(method=b'PATCH', path=b"/path?query=param")
+ def test_patch(self, patch_request):
+ raise NotImplementedError()
-class TestExportCurlCommand:
- def test_get(self):
- flow = tflow.tflow(req=req_get())
+class TestExportCurlCommand(TExport):
+ def test_get(self, get_request):
result = """curl -H 'header:qvalue' -H 'content-length:7' 'http://address:22/path?a=foo&a=bar&b=baz'"""
- assert export.curl_command(flow) == result
+ assert export.curl_command(get_request) == result
- def test_post(self):
- flow = tflow.tflow(req=req_post())
- result = """curl -X POST 'http://address:22/path' --data-binary 'content'"""
- assert export.curl_command(flow) == result
+ def test_post(self, post_request):
+ result = "curl -X POST 'http://address:22/path' --data-binary '{}'".format(
+ str(bytes(range(256)))[2:-1]
+ )
+ assert export.curl_command(post_request) == result
- def test_patch(self):
- flow = tflow.tflow(req=req_patch())
+ def test_patch(self, patch_request):
result = """curl -H 'header:qvalue' -H 'content-length:7' -X PATCH 'http://address:22/path?query=param' --data-binary 'content'"""
- assert export.curl_command(flow) == result
+ assert export.curl_command(patch_request) == result
-class TestExportPythonCode:
- def test_get(self):
- flow = tflow.tflow(req=req_get())
- python_equals("mitmproxy/data/test_flow_export/python_get.py", export.python_code(flow))
+class TestExportPythonCode(TExport):
+ def test_get(self, get_request):
+ python_equals("mitmproxy/data/test_flow_export/python_get.py",
+ export.python_code(get_request))
- def test_post(self):
- flow = tflow.tflow(req=req_post())
- python_equals("mitmproxy/data/test_flow_export/python_post.py", export.python_code(flow))
+ def test_post(self, post_request):
+ python_equals("mitmproxy/data/test_flow_export/python_post.py",
+ export.python_code(post_request))
- def test_post_json(self):
- p = req_post()
- p.content = b'{"name": "example", "email": "example@example.com"}'
- p.headers = Headers(content_type="application/json")
- flow = tflow.tflow(req=p)
- python_equals("mitmproxy/data/test_flow_export/python_post_json.py", export.python_code(flow))
+ def test_post_json(self, post_request):
+ post_request.request.content = b'{"name": "example", "email": "example@example.com"}'
+ post_request.request.headers = Headers(content_type="application/json")
+ python_equals("mitmproxy/data/test_flow_export/python_post_json.py",
+ export.python_code(post_request))
- def test_patch(self):
- flow = tflow.tflow(req=req_patch())
- python_equals("mitmproxy/data/test_flow_export/python_patch.py", export.python_code(flow))
+ def test_patch(self, patch_request):
+ python_equals("mitmproxy/data/test_flow_export/python_patch.py",
+ export.python_code(patch_request))
-class TestExportLocustCode:
- def test_get(self):
- flow = tflow.tflow(req=req_get())
- python_equals("mitmproxy/data/test_flow_export/locust_get.py", export.locust_code(flow))
+class TestExportLocustCode(TExport):
+ def test_get(self, get_request):
+ python_equals("mitmproxy/data/test_flow_export/locust_get.py",
+ export.locust_code(get_request))
- def test_post(self):
- p = req_post()
- p.content = b'content'
- p.headers = ''
- flow = tflow.tflow(req=p)
- python_equals("mitmproxy/data/test_flow_export/locust_post.py", export.locust_code(flow))
+ def test_post(self, post_request):
+ post_request.request.content = b'content'
+ post_request.request.headers.clear()
+ python_equals("mitmproxy/data/test_flow_export/locust_post.py",
+ export.locust_code(post_request))
- def test_patch(self):
- flow = tflow.tflow(req=req_patch())
- python_equals("mitmproxy/data/test_flow_export/locust_patch.py", export.locust_code(flow))
+ def test_patch(self, patch_request):
+ python_equals("mitmproxy/data/test_flow_export/locust_patch.py",
+ export.locust_code(patch_request))
-class TestExportLocustTask:
- def test_get(self):
- flow = tflow.tflow(req=req_get())
- python_equals("mitmproxy/data/test_flow_export/locust_task_get.py", export.locust_task(flow))
+class TestExportLocustTask(TExport):
+ def test_get(self, get_request):
+ python_equals("mitmproxy/data/test_flow_export/locust_task_get.py",
+ export.locust_task(get_request))
- def test_post(self):
- flow = tflow.tflow(req=req_post())
- python_equals("mitmproxy/data/test_flow_export/locust_task_post.py", export.locust_task(flow))
+ def test_post(self, post_request):
+ python_equals("mitmproxy/data/test_flow_export/locust_task_post.py",
+ export.locust_task(post_request))
- def test_patch(self):
- flow = tflow.tflow(req=req_patch())
- python_equals("mitmproxy/data/test_flow_export/locust_task_patch.py", export.locust_task(flow))
+ def test_patch(self, patch_request):
+ python_equals("mitmproxy/data/test_flow_export/locust_task_patch.py",
+ export.locust_task(patch_request))
class TestURL:
diff --git a/test/mitmproxy/test_optmanager.py b/test/mitmproxy/test_optmanager.py
index 31b6e52b..a685570f 100644
--- a/test/mitmproxy/test_optmanager.py
+++ b/test/mitmproxy/test_optmanager.py
@@ -38,12 +38,6 @@ class TM(optmanager.OptManager):
self.add_option("one", typing.Optional[str], None, "help")
-def test_add_option():
- o = TO()
- with pytest.raises(ValueError, match="already exists"):
- o.add_option("one", typing.Optional[int], None, "help")
-
-
def test_defaults():
o = TD2()
defaults = {
diff --git a/test/mitmproxy/utils/test_typecheck.py b/test/mitmproxy/utils/test_typecheck.py
index d99a914f..fd0c6e0c 100644
--- a/test/mitmproxy/utils/test_typecheck.py
+++ b/test/mitmproxy/utils/test_typecheck.py
@@ -79,3 +79,9 @@ def test_check_io():
typecheck.check_type("foo", io.StringIO(), typing.IO[str])
with pytest.raises(TypeError):
typecheck.check_type("foo", "foo", typing.IO[str])
+
+
+def test_check_any():
+ typecheck.check_type("foo", 42, typing.Any)
+ typecheck.check_type("foo", object(), typing.Any)
+ typecheck.check_type("foo", None, typing.Any)
generated by cgit v1.2.3 (git 2.25.1) at 2025-08-12 19:53:56 +0000