diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/mitmproxy/addons/test_readfile.py | 142 | ||||
| -rw-r--r-- | test/mitmproxy/addons/test_readstdin.py | 53 | ||||
| -rw-r--r-- | test/mitmproxy/utils/test_typecheck.py | 6 |
3 files changed, 98 insertions, 103 deletions
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/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) |