aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-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_export.py145
-rw-r--r--test/mitmproxy/utils/test_typecheck.py6
9 files changed, 202 insertions, 168 deletions
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_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/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-04-09 17:23:02 +0000