kylefang/mitmproxy
1
0
Fork 0
mirror of https://github.com/zhigang1992/mitmproxy.git synced 2026-04-23 03:49:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1980 from Kriechi/improve-tests

Browse Source 
improve tests
This commit is contained in:
Thomas Kriechbaumer
2017-02-02 17:23:11 +01:00
committed by GitHub
parent c1bc1ea584 4f0b2bc4de
commit 3f4d472c80
59 changed files with 533 additions and 620 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
mitmproxy/test/tflow.py
View File

@@ -115,6 +115,26 @@ def tflow(client_conn=True, server_conn=True, req=True, resp=None, err=None):
return f
def tdummyflow(client_conn=True, server_conn=True, err=None):
class DummyFlow(flow.Flow):
"""A flow that is neither HTTP nor TCP."""
def __init__(self, client_conn, server_conn, live=None):
super().__init__("dummy", client_conn, server_conn, live)
if client_conn is True:
client_conn = tclient_conn()
if server_conn is True:
server_conn = tserver_conn()
if err is True:
err = terr()
f = DummyFlow(client_conn, server_conn)
f.error = err
f.reply = controller.DummyReply()
return f
def tclient_conn():
"""
@return: mitmproxy.proxy.connection.ClientConnection

69
mitmproxy/test/tutils.py
View File

@@ -4,19 +4,13 @@ import os
import time
import shutil
from contextlib import contextmanager
import sys
from mitmproxy.utils import data
from mitmproxy.net import tcp
from mitmproxy.net import http
def treader(bytes):
"""
Construct a tcp.Read object from bytes.
"""
fp = BytesIO(bytes)
return tcp.Reader(fp)
test_data = data.Data(__name__).push("../../test/")
@contextmanager
@@ -31,65 +25,12 @@ def tmpdir(*args, **kwargs):
shutil.rmtree(temp_workdir)
def _check_exception(expected, actual, exc_tb):
if isinstance(expected, str):
if expected.lower() not in str(actual).lower():
raise AssertionError(
"Expected %s, but caught %s" % (
repr(expected), repr(actual)
)
)
else:
if not isinstance(actual, expected):
raise AssertionError(
"Expected %s, but caught %s %s" % (
expected.__name__, actual.__class__.__name__, repr(actual)
)
)
def raises(expected_exception, obj=None, *args, **kwargs):
def treader(bytes):
"""
Assert that a callable raises a specified exception.
:exc An exception class or a string. If a class, assert that an
exception of this type is raised. If a string, assert that the string
occurs in the string representation of the exception, based on a
case-insenstivie match.
:obj A callable object.
:args Arguments to be passsed to the callable.
:kwargs Arguments to be passed to the callable.
Construct a tcp.Read object from bytes.
"""
if obj is None:
return RaisesContext(expected_exception)
else:
try:
ret = obj(*args, **kwargs)
except Exception as actual:
_check_exception(expected_exception, actual, sys.exc_info()[2])
else:
raise AssertionError("No exception raised. Return value: {}".format(ret))
class RaisesContext:
def __init__(self, expected_exception):
self.expected_exception = expected_exception
def __enter__(self):
return
def __exit__(self, exc_type, exc_val, exc_tb):
if not exc_type:
raise AssertionError("No exception raised.")
else:
_check_exception(self.expected_exception, exc_val, exc_tb)
return True
test_data = data.Data(__name__).push("../../test/")
fp = BytesIO(bytes)
return tcp.Reader(fp)
def treq(**kwargs):

50
test/conftest.py
View File

@@ -1,6 +1,7 @@
import os
import pytest
import OpenSSL
import functools
import mitmproxy.net.tcp
@@ -9,6 +10,53 @@ requires_alpn = pytest.mark.skipif(
not mitmproxy.net.tcp.HAS_ALPN,
reason='requires OpenSSL with ALPN support')
skip_windows = pytest.mark.skipif(
os.name == "nt",
reason='Skipping due to Windows'
)
skip_not_windows = pytest.mark.skipif(
os.name != "nt",
reason='Skipping due to not Windows'
)
skip_appveyor = pytest.mark.skipif(
"APPVEYOR" in os.environ,
reason='Skipping due to Appveyor'
)
original_pytest_raises = pytest.raises
def raises(exc, *args, **kwargs):
functools.wraps(original_pytest_raises)
if isinstance(exc, str):
return RaisesContext(exc)
else:
return original_pytest_raises(exc, *args, **kwargs)
pytest.raises = raises
class RaisesContext:
def __init__(self, expected_exception):
self.expected_exception = expected_exception
def __enter__(self):
return
def __exit__(self, exc_type, exc_val, exc_tb):
if not exc_type:
raise AssertionError("No exception raised.")
else:
if self.expected_exception.lower() not in str(exc_val).lower():
raise AssertionError(
"Expected %s, but caught %s" % (repr(self.expected_exception), repr(exc_val))
)
return True
@pytest.fixture()
def disable_alpn(monkeypatch):
@@ -68,7 +116,7 @@ def pytest_runtestloop(session):
prefix = os.getcwd()
excluded_files = [os.path.normpath(f) for f in pytest.config.option.no_full_cov]
measured_files = [os.path.normpath(os.path.relpath(f, prefix)) for f in cov.get_data().measured_files()]
measured_files = [f for f in measured_files if f not in excluded_files]
measured_files = [f for f in measured_files if not any(f.startswith(excluded_f) for excluded_f in excluded_files)]
for name in pytest.config.option.full_cov:
files = [f for f in measured_files if f.startswith(os.path.normpath(name))]

9
test/mitmproxy/addons/test_clientplayback.py
View File

@@ -1,4 +1,5 @@
import os
import pytest
from unittest import mock
from mitmproxy.test import tflow
@@ -57,9 +58,5 @@ class TestClientPlayback:
tctx.configure(cp, client_replay=[path])
tctx.configure(cp, client_replay=[])
tctx.configure(cp)
tutils.raises(
exceptions.OptionsError,
tctx.configure,
cp,
client_replay=["nonexistent"]
)
with pytest.raises(exceptions.OptionsError):
tctx.configure(cp, client_replay=["nonexistent"])

4
test/mitmproxy/addons/test_dumper.py
View File

@@ -1,5 +1,6 @@
import io
import shutil
import pytest
from unittest import mock
from mitmproxy.test import tflow
@@ -25,7 +26,8 @@ def test_configure():
ctx.configure(d, filtstr=None)
assert not d.filter
tutils.raises(exceptions.OptionsError, ctx.configure, d, filtstr="~~")
with pytest.raises(exceptions.OptionsError):
ctx.configure(d, filtstr="~~")
assert not d.filter

11
test/mitmproxy/addons/test_intercept.py
View File

@@ -1,8 +1,9 @@
import pytest
from mitmproxy.addons import intercept
from mitmproxy import options
from mitmproxy import exceptions
from mitmproxy.test import taddons
from mitmproxy.test import tutils
from mitmproxy.test import tflow
@@ -18,12 +19,8 @@ def test_simple():
assert not r.filt
tctx.configure(r, intercept="~q")
assert r.filt
tutils.raises(
exceptions.OptionsError,
tctx.configure,
r,
intercept="~~"
)
with pytest.raises(exceptions.OptionsError):
tctx.configure(r, intercept="~~")
tctx.configure(r, intercept=None)
assert not r.filt

43
test/mitmproxy/addons/test_proxyauth.py
View File

@@ -1,4 +1,5 @@
import binascii
import pytest
from mitmproxy import exceptions
from mitmproxy.test import taddons
@@ -20,10 +21,8 @@ def test_parse_http_basic_auth():
def test_configure():
up = proxyauth.ProxyAuth()
with taddons.context() as ctx:
tutils.raises(
exceptions.OptionsError,
ctx.configure, up, auth_singleuser="foo"
)
with pytest.raises(exceptions.OptionsError):
ctx.configure(up, auth_singleuser="foo")
ctx.configure(up, auth_singleuser="foo:bar")
assert up.singleuser == ["foo", "bar"]
@@ -36,20 +35,10 @@ def test_configure():
ctx.configure(up, auth_nonanonymous=False)
assert not up.nonanonymous
tutils.raises(
exceptions.OptionsError,
ctx.configure,
up,
auth_htpasswd = tutils.test_data.path(
"mitmproxy/net/data/server.crt"
)
)
tutils.raises(
exceptions.OptionsError,
ctx.configure,
up,
auth_htpasswd = "nonexistent"
)
with pytest.raises(exceptions.OptionsError):
ctx.configure(up, auth_htpasswd=tutils.test_data.path("mitmproxy/net/data/server.crt"))
with pytest.raises(exceptions.OptionsError):
ctx.configure(up, auth_htpasswd="nonexistent")
ctx.configure(
up,
@@ -63,20 +52,10 @@ def test_configure():
ctx.configure(up, auth_htpasswd = None)
assert not up.htpasswd
tutils.raises(
exceptions.OptionsError,
ctx.configure,
up,
auth_nonanonymous = True,
mode = "transparent"
)
tutils.raises(
exceptions.OptionsError,
ctx.configure,
up,
auth_nonanonymous = True,
mode = "socks5"
)
with pytest.raises(exceptions.OptionsError):
ctx.configure(up, auth_nonanonymous=True, mode="transparent")
with pytest.raises(exceptions.OptionsError):
ctx.configure(up, auth_nonanonymous=True, mode="socks5")
def test_check():

20
test/mitmproxy/addons/test_replace.py
View File

@@ -1,4 +1,5 @@
import os.path
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import tutils
@@ -15,24 +16,17 @@ class TestReplace:
assert x == ("foo", "bar", "vo/ing/")
x = replace.parse_hook("/bar/voing")
assert x == (".*", "bar", "voing")
tutils.raises("invalid replacement", replace.parse_hook, "/")
with pytest.raises("invalid replacement"):
replace.parse_hook("/")
def test_configure(self):
r = replace.Replace()
with taddons.context() as tctx:
tctx.configure(r, replacements=[("one", "two", "three")])
tutils.raises(
"invalid filter pattern",
tctx.configure,
r,
replacements=[("~b", "two", "three")]
)
tutils.raises(
"invalid regular expression",
tctx.configure,
r,
replacements=[("foo", "+", "three")]
)
with pytest.raises("invalid filter pattern"):
tctx.configure(r, replacements=[("~b", "two", "three")])
with pytest.raises("invalid regular expression"):
tctx.configure(r, replacements=[("foo", "+", "three")])
tctx.configure(r, replacements=["/a/b/c/"])
def test_simple(self):

29
test/mitmproxy/addons/test_script.py
View File

@@ -2,6 +2,8 @@ import traceback
import sys
import time
import re
import watchdog.events
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import tutils
@@ -11,12 +13,9 @@ from mitmproxy import options
from mitmproxy import proxy
from mitmproxy import master
from mitmproxy import utils
from mitmproxy.addons import script
import watchdog.events
from .. import tutils as ttutils
from ...conftest import skip_not_windows
def test_scriptenv():
@@ -58,18 +57,18 @@ def test_reloadhandler():
class TestParseCommand:
def test_empty_command(self):
with tutils.raises(ValueError):
with pytest.raises(ValueError):
script.parse_command("")
with tutils.raises(ValueError):
with pytest.raises(ValueError):
script.parse_command(" ")
def test_no_script_file(self):
with tutils.raises("not found"):
with pytest.raises("not found"):
script.parse_command("notfound")
with tutils.tmpdir() as dir:
with tutils.raises("not a file"):
with pytest.raises("not a file"):
script.parse_command(dir)
def test_parse_args(self):
@@ -84,7 +83,7 @@ class TestParseCommand:
"mitmproxy/data/addonscripts/recorder.py 'foo bar'"
) == ("mitmproxy/data/addonscripts/recorder.py", ["foo bar"])
@ttutils.skip_not_windows
@skip_not_windows
def test_parse_windows(self):
with utils.chdir(tutils.test_data.dirname):
assert script.parse_command(
@@ -205,12 +204,8 @@ class TestScriptLoader:
f = tflow.tflow(resp=True)
with m.handlecontext():
tutils.raises(
"file not found",
sl.run_once,
"nonexistent",
[f]
)
with pytest.raises("file not found"):
sl.run_once("nonexistent", [f])
def test_simple(self):
o = options.Options(scripts=[])
@@ -231,14 +226,14 @@ class TestScriptLoader:
o = options.Options(scripts=["one", "one"])
m = master.Master(o, proxy.DummyServer())
sc = script.ScriptLoader()
with tutils.raises(exceptions.OptionsError):
with pytest.raises(exceptions.OptionsError):
m.addons.add(o, sc)
def test_nonexistent(self):
o = options.Options(scripts=["nonexistent"])
m = master.Master(o, proxy.DummyServer())
sc = script.ScriptLoader()
with tutils.raises(exceptions.OptionsError):
with pytest.raises(exceptions.OptionsError):
m.addons.add(o, sc)
def test_order(self):

9
test/mitmproxy/addons/test_serverplayback.py
View File

@@ -1,5 +1,6 @@
import os
import urllib
import pytest
from mitmproxy.test import tutils
from mitmproxy.test import tflow
@@ -25,12 +26,8 @@ def test_config():
fpath = os.path.join(p, "flows")
tdump(fpath, [tflow.tflow(resp=True)])
tctx.configure(s, server_replay=[fpath])
tutils.raises(
exceptions.OptionsError,
tctx.configure,
s,
server_replay=[p]
)
with pytest.raises(exceptions.OptionsError):
tctx.configure(s, server_replay=[p])
def test_tick():

14
test/mitmproxy/addons/test_setheaders.py
View File

@@ -1,5 +1,6 @@
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.addons import setheaders
@@ -13,17 +14,14 @@ class TestSetHeaders:
assert x == ("foo", "bar", "vo/ing/")
x = setheaders.parse_setheader("/bar/voing")
assert x == (".*", "bar", "voing")
tutils.raises("invalid replacement", setheaders.parse_setheader, "/")
with pytest.raises("invalid replacement"):
setheaders.parse_setheader("/")
def test_configure(self):
sh = setheaders.SetHeaders()
with taddons.context() as tctx:
tutils.raises(
"invalid setheader filter pattern",
tctx.configure,
sh,
setheaders = [("~b", "one", "two")]
)
with pytest.raises("invalid setheader filter pattern"):
tctx.configure(sh, setheaders = [("~b", "one", "two")])
tctx.configure(sh, setheaders = ["/foo/bar/voing"])
def test_setheaders(self):

11
test/mitmproxy/addons/test_stickyauth.py
View File

@@ -1,6 +1,7 @@
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import taddons
from mitmproxy.test import tutils
from mitmproxy.addons import stickyauth
from mitmproxy import exceptions
@@ -10,12 +11,8 @@ def test_configure():
r = stickyauth.StickyAuth()
with taddons.context() as tctx:
tctx.configure(r, stickyauth="~s")
tutils.raises(
exceptions.OptionsError,
tctx.configure,
r,
stickyauth="~~"
)
with pytest.raises(exceptions.OptionsError):
tctx.configure(r, stickyauth="~~")
tctx.configure(r, stickyauth=None)
assert not r.flt

8
test/mitmproxy/addons/test_stickycookie.py
View File

@@ -1,5 +1,6 @@
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.addons import stickycookie
@@ -15,9 +16,8 @@ class TestStickyCookie:
def test_config(self):
sc = stickycookie.StickyCookie()
with taddons.context() as tctx:
tutils.raises(
"invalid filter", tctx.configure, sc, stickycookie="~b"
)
with pytest.raises("invalid filter"):
tctx.configure(sc, stickycookie="~b")
tctx.configure(sc, stickycookie="foo")
assert sc.flt

16
test/mitmproxy/addons/test_streamfile.py
View File

@@ -1,8 +1,10 @@
import os.path
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
import os.path
from mitmproxy import io
from mitmproxy import exceptions
from mitmproxy.tools import dump
@@ -14,14 +16,10 @@ def test_configure():
with taddons.context(options=dump.Options()) as tctx:
with tutils.tmpdir() as tdir:
p = os.path.join(tdir, "foo")
tutils.raises(
exceptions.OptionsError,
tctx.configure, sa, streamfile=tdir
)
tutils.raises(
"invalid filter",
tctx.configure, sa, streamfile=p, filtstr="~~"
)
with pytest.raises(exceptions.OptionsError):
tctx.configure(sa, streamfile=tdir)
with pytest.raises("invalid filter"):
tctx.configure(sa, streamfile=p, filtstr="~~")
tctx.configure(sa, filtstr="foo")
assert sa.filt
tctx.configure(sa, filtstr=None)

26
test/mitmproxy/addons/test_upstream_auth.py
View File

@@ -1,9 +1,9 @@
import base64
import pytest
from mitmproxy import exceptions
from mitmproxy.test import taddons
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.addons import upstream_auth
@@ -19,24 +19,12 @@ def test_configure():
tctx.configure(up, upstream_auth=None)
assert not up.auth
tutils.raises(
exceptions.OptionsError,
tctx.configure,
up,
upstream_auth=""
)
tutils.raises(
exceptions.OptionsError,
tctx.configure,
up,
upstream_auth=":"
)
tutils.raises(
exceptions.OptionsError,
tctx.configure,
up,
upstream_auth=":test"
)
with pytest.raises(exceptions.OptionsError):
tctx.configure(up, upstream_auth="")
with pytest.raises(exceptions.OptionsError):
tctx.configure(up, upstream_auth=":")
with pytest.raises(exceptions.OptionsError):
tctx.configure(up, upstream_auth=":test")
def test_simple():

27
test/mitmproxy/addons/test_view.py
View File

@@ -1,5 +1,6 @@
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.addons import view
from mitmproxy import flowfilter
@@ -170,8 +171,10 @@ def test_reversed():
assert v[0].request.timestamp_start == 3
assert v[-1].request.timestamp_start == 1
assert v[2].request.timestamp_start == 1
tutils.raises(IndexError, v.__getitem__, 5)
tutils.raises(IndexError, v.__getitem__, -5)
with pytest.raises(IndexError):
v.__getitem__(5)
with pytest.raises(IndexError):
v.__getitem__(-5)
assert v._bisect(v[0]) == 1
assert v._bisect(v[2]) == 3
@@ -314,8 +317,10 @@ def test_focus():
assert f.flow is v[0]
# Try to set to something not in view
tutils.raises(ValueError, f.__setattr__, "flow", tft())
tutils.raises(ValueError, f.__setattr__, "index", 99)
with pytest.raises(ValueError):
f.__setattr__("flow", tft())
with pytest.raises(ValueError):
f.__setattr__("index", 99)
v.add(tft(start=0))
assert f.index == 1
@@ -362,13 +367,15 @@ def test_settings():
v = view.View()
f = tft()
tutils.raises(KeyError, v.settings.__getitem__, f)
with pytest.raises(KeyError):
v.settings.__getitem__(f)
v.add(f)
v.settings[f]["foo"] = "bar"
assert v.settings[f]["foo"] == "bar"
assert len(list(v.settings)) == 1
v.remove(f)
tutils.raises(KeyError, v.settings.__getitem__, f)
with pytest.raises(KeyError):
v.settings.__getitem__(f)
assert not v.settings.keys()
v.add(f)
@@ -382,10 +389,12 @@ def test_configure():
v = view.View()
with taddons.context(options=Options()) as tctx:
tctx.configure(v, filter="~q")
tutils.raises("invalid interception filter", tctx.configure, v, filter="~~")
with pytest.raises("invalid interception filter"):
tctx.configure(v, filter="~~")
tctx.configure(v, console_order="method")
tutils.raises("unknown flow order", tctx.configure, v, console_order="no")
with pytest.raises("unknown flow order"):
tctx.configure(v, console_order="no")
tctx.configure(v, console_order_reversed=True)

5
test/mitmproxy/console/test_common.py
View File

@@ -1,9 +1,10 @@
from mitmproxy.test import tflow
from mitmproxy.tools.console import common
from .. import tutils
from ...conftest import skip_appveyor
@tutils.skip_appveyor
@skip_appveyor
def test_format_flow():
f = tflow.tflow(resp=True)
assert common.format_flow(f, True)

5
test/mitmproxy/console/test_help.py
View File

@@ -1,8 +1,9 @@
import mitmproxy.tools.console.help as help
from .. import tutils
from ...conftest import skip_appveyor
@tutils.skip_appveyor
@skip_appveyor
class TestHelp:
def test_helptext(self):

5
test/mitmproxy/console/test_palettes.py
View File

@@ -1,8 +1,9 @@
import mitmproxy.tools.console.palettes as palettes
from .. import tutils
from ...conftest import skip_appveyor
@tutils.skip_appveyor
@skip_appveyor
class TestPalette:
def test_helptext(self):

5
test/mitmproxy/contentviews/test_api.py
View File

@@ -1,4 +1,5 @@
from unittest import mock
import pytest
from mitmproxy import contentviews
from mitmproxy.exceptions import ContentViewException
@@ -17,11 +18,11 @@ def test_add_remove():
contentviews.add(tcv)
# repeated addition causes exception
with tutils.raises(ContentViewException):
with pytest.raises(ContentViewException):
contentviews.add(tcv)
# Same shortcut doesn't work either.
with tutils.raises(ContentViewException):
with pytest.raises(ContentViewException):
contentviews.add(TestContentView())
contentviews.remove(tcv)

10
test/mitmproxy/net/http/http1/test_assemble.py
View File

@@ -1,3 +1,5 @@
import pytest
from mitmproxy import exceptions
from mitmproxy.net.http import Headers
from mitmproxy.net.http.http1.assemble import (
@@ -5,7 +7,7 @@ from mitmproxy.net.http.http1.assemble import (
assemble_response_head, _assemble_request_line, _assemble_request_headers,
_assemble_response_headers,
assemble_body)
from mitmproxy.test.tutils import treq, raises, tresp
from mitmproxy.test.tutils import treq, tresp
def test_assemble_request():
@@ -18,7 +20,7 @@ def test_assemble_request():
b"content"
)
with raises(exceptions.HttpException):
with pytest.raises(exceptions.HttpException):
assemble_request(treq(content=None))
@@ -39,7 +41,7 @@ def test_assemble_response():
b"message"
)
with raises(exceptions.HttpException):
with pytest.raises(exceptions.HttpException):
assemble_response(tresp(content=None))
@@ -70,7 +72,7 @@ def test_assemble_request_line():
absolute_request = treq(first_line_format="absolute").data
assert _assemble_request_line(absolute_request) == b"GET http://address:22/path HTTP/1.1"
with raises(RuntimeError):
with pytest.raises(RuntimeError):
_assemble_request_line(treq(first_line_format="invalid_form").data)

61
test/mitmproxy/net/http/http1/test_read.py
View File

@@ -10,7 +10,7 @@ from mitmproxy.net.http.http1.read import (
_read_request_line, _parse_authority_form, _read_response_line, _check_http_version,
_read_headers, _read_chunked, get_header_tokens
)
from mitmproxy.test.tutils import treq, tresp, raises
from mitmproxy.test.tutils import treq, tresp
def test_get_header_tokens():
@@ -45,7 +45,8 @@ def test_read_request(input):
])
def test_read_request_error(input):
rfile = BytesIO(input)
raises(exceptions.HttpException, read_request, rfile)
with pytest.raises(exceptions.HttpException):
read_request(rfile)
def test_read_request_head():
@@ -116,12 +117,12 @@ class TestReadBody:
def test_known_size_limit(self):
rfile = BytesIO(b"foobar")
with raises(exceptions.HttpException):
with pytest.raises(exceptions.HttpException):
b"".join(read_body(rfile, 3, 2))
def test_known_size_too_short(self):
rfile = BytesIO(b"foo")
with raises(exceptions.HttpException):
with pytest.raises(exceptions.HttpException):
b"".join(read_body(rfile, 6))
def test_unknown_size(self):
@@ -131,7 +132,7 @@ class TestReadBody:
def test_unknown_size_limit(self):
rfile = BytesIO(b"foobar")
with raises(exceptions.HttpException):
with pytest.raises(exceptions.HttpException):
b"".join(read_body(rfile, -1, 3))
def test_max_chunk_size(self):
@@ -185,7 +186,7 @@ def test_expected_http_body_size():
# explicit length
for val in (b"foo", b"-7"):
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
expected_http_body_size(
treq(headers=Headers(content_length=val))
)
@@ -209,11 +210,11 @@ def test_get_first_line():
rfile = BytesIO(b"\r\nfoo\r\nbar")
assert _get_first_line(rfile) == b"foo"
with raises(exceptions.HttpReadDisconnect):
with pytest.raises(exceptions.HttpReadDisconnect):
rfile = BytesIO(b"")
_get_first_line(rfile)
with raises(exceptions.HttpReadDisconnect):
with pytest.raises(exceptions.HttpReadDisconnect):
rfile = Mock()
rfile.readline.side_effect = exceptions.TcpDisconnect
_get_first_line(rfile)
@@ -232,23 +233,23 @@ def test_read_request_line():
assert (t(b"GET http://foo:42/bar HTTP/1.1") ==
("absolute", b"GET", b"http", b"foo", 42, b"/bar", b"HTTP/1.1"))
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
t(b"GET / WTF/1.1")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
t(b"this is not http")
with raises(exceptions.HttpReadDisconnect):
with pytest.raises(exceptions.HttpReadDisconnect):
t(b"")
def test_parse_authority_form():
assert _parse_authority_form(b"foo:42") == (b"foo", 42)
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
_parse_authority_form(b"foo")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
_parse_authority_form(b"foo:bar")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
_parse_authority_form(b"foo:99999999")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
_parse_authority_form(b"f\x00oo:80")
@@ -262,14 +263,14 @@ def test_read_response_line():
# https://github.com/mitmproxy/mitmproxy/issues/784
assert t(b"HTTP/1.1 200 Non-Autoris\xc3\xa9") == (b"HTTP/1.1", 200, b"Non-Autoris\xc3\xa9")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
assert t(b"HTTP/1.1")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
t(b"HTTP/1.1 OK OK")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
t(b"WTF/1.1 200 OK")
with raises(exceptions.HttpReadDisconnect):
with pytest.raises(exceptions.HttpReadDisconnect):
t(b"")
@@ -278,11 +279,11 @@ def test_check_http_version():
_check_http_version(b"HTTP/1.0")
_check_http_version(b"HTTP/1.1")
_check_http_version(b"HTTP/2.0")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
_check_http_version(b"WTF/1.0")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
_check_http_version(b"HTTP/1.10")
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
_check_http_version(b"HTTP/1.b")
@@ -321,17 +322,17 @@ class TestReadHeaders:
def test_read_continued_err(self):
data = b"\tfoo: bar\r\n"
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
self._read(data)
def test_read_err(self):
data = b"foo"
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
self._read(data)
def test_read_empty_name(self):
data = b":foo"
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
self._read(data)
def test_read_empty_value(self):
@@ -345,7 +346,7 @@ def test_read_chunked():
req.headers["Transfer-Encoding"] = "chunked"
data = b"1\r\na\r\n0\r\n"
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
b"".join(_read_chunked(BytesIO(data)))
data = b"1\r\na\r\n0\r\n\r\n"
@@ -355,17 +356,17 @@ def test_read_chunked():
assert b"".join(_read_chunked(BytesIO(data))) == b"ab"
data = b"\r\n"
with raises("closed prematurely"):
with pytest.raises("closed prematurely"):
b"".join(_read_chunked(BytesIO(data)))
data = b"1\r\nfoo"
with raises("malformed chunked body"):
with pytest.raises("malformed chunked body"):
b"".join(_read_chunked(BytesIO(data)))
data = b"foo\r\nfoo"
with raises(exceptions.HttpSyntaxException):
with pytest.raises(exceptions.HttpSyntaxException):
b"".join(_read_chunked(BytesIO(data)))
data = b"5\r\naaaaa\r\n0\r\n\r\n"
with raises("too large"):
with pytest.raises("too large"):
b"".join(_read_chunked(BytesIO(data), limit=2))

4
test/mitmproxy/net/http/test_cookies.py
View File

@@ -1,8 +1,8 @@
import time
import pytest
from unittest import mock
from mitmproxy.net.http import cookies
from mitmproxy.test.tutils import raises
cookie_pairs = [
@@ -270,7 +270,7 @@ def test_refresh_cookie():
assert "00:21:38" in cookies.refresh_set_cookie_header(c, 60)
c = "foo,bar"
with raises(ValueError):
with pytest.raises(ValueError):
cookies.refresh_set_cookie_header(c, 60)
# https://github.com/mitmproxy/mitmproxy/issues/773

5
test/mitmproxy/net/http/test_encoding.py
View File

@@ -2,7 +2,6 @@ from unittest import mock
import pytest
from mitmproxy.net.http import encoding
from mitmproxy.test import tutils
@pytest.mark.parametrize("encoder", [
@@ -12,7 +11,7 @@ from mitmproxy.test import tutils
def test_identity(encoder):
assert b"string" == encoding.decode(b"string", encoder)
assert b"string" == encoding.encode(b"string", encoder)
with tutils.raises(ValueError):
with pytest.raises(ValueError):
encoding.encode(b"string", "nonexistent encoding")
@@ -40,7 +39,7 @@ def test_encoders(encoder):
encoder
)
with tutils.raises(ValueError):
with pytest.raises(ValueError):
encoding.decode(b"foobar", encoder)

6
test/mitmproxy/net/http/test_headers.py
View File

@@ -1,7 +1,7 @@
import collections
import pytest
from mitmproxy.net.http.headers import Headers, parse_content_type, assemble_content_type
from mitmproxy.test.tutils import raises
class TestHeaders:
@@ -40,7 +40,7 @@ class TestHeaders:
assert headers["Host"] == "example.com"
assert headers["Accept"] == "text/plain"
with raises(TypeError):
with pytest.raises(TypeError):
Headers([[b"Host", u"not-bytes"]])
def test_set(self):
@@ -48,7 +48,7 @@ class TestHeaders:
headers[u"foo"] = u"1"
headers[b"bar"] = b"2"
headers["baz"] = b"3"
with raises(TypeError):
with pytest.raises(TypeError):
headers["foobar"] = 42
assert len(headers) == 3

18
test/mitmproxy/net/http/test_message.py
View File

@@ -1,5 +1,7 @@
# -*- coding: utf-8 -*-
import pytest
from mitmproxy.test import tutils
from mitmproxy.net import http
@@ -129,14 +131,14 @@ class TestMessageContentEncoding:
r.decode()
assert r.raw_content == b"foo"
with tutils.raises(TypeError):
with pytest.raises(TypeError):
r.content = u"foo"
def test_unknown_ce(self):
r = tutils.tresp()
r.headers["content-encoding"] = "zopfli"
r.raw_content = b"foo"
with tutils.raises(ValueError):
with pytest.raises(ValueError):
assert r.content
assert r.headers["content-encoding"]
assert r.get_content(strict=False) == b"foo"
@@ -145,7 +147,7 @@ class TestMessageContentEncoding:
r = tutils.tresp()
r.headers["content-encoding"] = "utf8"
r.raw_content = b"foo"
with tutils.raises(ValueError):
with pytest.raises(ValueError):
assert r.content
assert r.headers["content-encoding"]
assert r.get_content(strict=False) == b"foo"
@@ -154,12 +156,12 @@ class TestMessageContentEncoding:
r = tutils.tresp()
r.encode("gzip")
r.raw_content = b"foo"
with tutils.raises(ValueError):
with pytest.raises(ValueError):
assert r.content
assert r.headers["content-encoding"]
assert r.get_content(strict=False) == b"foo"
with tutils.raises(ValueError):
with pytest.raises(ValueError):
r.decode()
assert r.raw_content == b"foo"
assert "content-encoding" in r.headers
@@ -188,7 +190,7 @@ class TestMessageContentEncoding:
assert "content-encoding" not in r.headers
assert r.raw_content == b"foo"
with tutils.raises(ValueError):
with pytest.raises(ValueError):
r.encode("zopfli")
assert r.raw_content == b"foo"
assert "content-encoding" not in r.headers
@@ -240,7 +242,7 @@ class TestMessageText:
r = tutils.tresp()
r.headers["content-type"] = "text/html; charset=wtf"
r.raw_content = b"foo"
with tutils.raises(ValueError):
with pytest.raises(ValueError):
assert r.text == u"foo"
assert r.get_text(strict=False) == u"foo"
@@ -248,7 +250,7 @@ class TestMessageText:
r = tutils.tresp()
r.headers["content-type"] = "text/html; charset=utf8"
r.raw_content = b"\xFF"
with tutils.raises(ValueError):
with pytest.raises(ValueError):
assert r.text
assert r.get_text(strict=False) == '\udcff'

8
test/mitmproxy/net/http/test_request.py
View File

@@ -1,13 +1,15 @@
# -*- coding: utf-8 -*-
import pytest
from mitmproxy.net.http import Headers
from mitmproxy.test.tutils import treq, raises
from mitmproxy.test.tutils import treq
from .test_message import _test_decoded_attr, _test_passthrough_attr
class TestRequestData:
def test_init(self):
with raises(ValueError):
with pytest.raises(ValueError):
treq(headers="foobar")
assert isinstance(treq(headers=()).headers, Headers)
@@ -105,7 +107,7 @@ class TestRequestUtils:
assert request.port == 42
assert request.path == "/foo"
with raises(ValueError):
with pytest.raises(ValueError):
request.url = "not-a-url"
def test_url_options(self):

10
test/mitmproxy/net/http/test_response.py
View File

@@ -1,17 +1,17 @@
import email
import time
import pytest
from mitmproxy.net.http import Headers
from mitmproxy.net.http import Response
from mitmproxy.net.http.cookies import CookieAttrs
from mitmproxy.test.tutils import raises, tresp
from mitmproxy.test.tutils import tresp
from .test_message import _test_passthrough_attr
class TestResponseData:
def test_init(self):
with raises(ValueError):
with pytest.raises(ValueError):
tresp(headers="foobar")
assert isinstance(tresp(headers=()).headers, Headers)
@@ -39,7 +39,7 @@ class TestResponseCore:
Response.make(content=b"foo")
Response.make(content="foo")
with raises(TypeError):
with pytest.raises(TypeError):
Response.make(content=42)
r = Response.make(headers=[(b"foo", b"bar")])
@@ -48,7 +48,7 @@ class TestResponseCore:
r = Response.make(headers=({"foo": "baz"}))
assert r.headers["foo"] == "baz"
with raises(TypeError):
with pytest.raises(TypeError):
Response.make(headers=42)
def test_status_code(self):

15
test/mitmproxy/net/http/test_url.py
View File

@@ -1,12 +1,11 @@
import pytest
import sys
from mitmproxy.test import tutils
from mitmproxy.net.http import url
def test_parse():
with tutils.raises(ValueError):
with pytest.raises(ValueError):
url.parse("")
s, h, po, pa = url.parse(b"http://foo.com:8888/test")
@@ -33,27 +32,27 @@ def test_parse():
s, h, po, pa = url.parse(b"https://foo")
assert po == 443
with tutils.raises(ValueError):
with pytest.raises(ValueError):
url.parse(b"https://foo:bar")
# Invalid IDNA
with tutils.raises(ValueError):
with pytest.raises(ValueError):
url.parse("http://\xfafoo")
# Invalid PATH
with tutils.raises(ValueError):
with pytest.raises(ValueError):
url.parse("http:/\xc6/localhost:56121")
# Null byte in host
with tutils.raises(ValueError):
with pytest.raises(ValueError):
url.parse("http://foo\0")
# Invalid IPv6 URL - see http://www.ietf.org/rfc/rfc2732.txt
with tutils.raises(ValueError):
with pytest.raises(ValueError):
url.parse('http://lo[calhost')
@pytest.mark.skipif(sys.version_info < (3, 6), reason='requires Python 3.6 or higher')
def test_parse_port_range():
# Port out of range
with tutils.raises(ValueError):
with pytest.raises(ValueError):
url.parse("http://foo:999999")

30
test/mitmproxy/net/test_socks.py
View File

@@ -1,5 +1,7 @@
import ipaddress
from io import BytesIO
import pytest
from mitmproxy.net import socks
from mitmproxy.net import tcp
from mitmproxy.test import tutils
@@ -22,7 +24,8 @@ def test_client_greeting():
def test_client_greeting_assert_socks5():
raw = tutils.treader(b"\x00\x00")
msg = socks.ClientGreeting.from_file(raw)
tutils.raises(socks.SocksError, msg.assert_socks5)
with pytest.raises(socks.SocksError):
msg.assert_socks5()
raw = tutils.treader(b"HTTP/1.1 200 OK" + b" " * 100)
msg = socks.ClientGreeting.from_file(raw)
@@ -45,11 +48,8 @@ def test_client_greeting_assert_socks5():
assert False
raw = tutils.treader(b"XX")
tutils.raises(
socks.SocksError,
socks.ClientGreeting.from_file,
raw,
fail_early=True)
with pytest.raises(socks.SocksError):
socks.ClientGreeting.from_file(raw, fail_early=True)
def test_server_greeting():
@@ -103,7 +103,8 @@ def test_username_password_auth():
def test_username_password_auth_assert_ver1():
raw = tutils.treader(b"\x02\x03usr\x03psd\xBE\xEF")
auth = socks.UsernamePasswordAuth.from_file(raw)
tutils.raises(socks.SocksError, auth.assert_authver1)
with pytest.raises(socks.SocksError):
auth.assert_authver1()
def test_username_password_auth_response():
@@ -122,7 +123,8 @@ def test_username_password_auth_response():
def test_username_password_auth_response_auth_assert_ver1():
raw = tutils.treader(b"\x02\x00\xBE\xEF")
auth = socks.UsernamePasswordAuthResponse.from_file(raw)
tutils.raises(socks.SocksError, auth.assert_authver1)
with pytest.raises(socks.SocksError):
auth.assert_authver1()
def test_message():
@@ -143,7 +145,8 @@ def test_message():
def test_message_assert_socks5():
raw = tutils.treader(b"\xEE\x01\x00\x03\x0bexample.com\xDE\xAD\xBE\xEF")
msg = socks.Message.from_file(raw)
tutils.raises(socks.SocksError, msg.assert_socks5)
with pytest.raises(socks.SocksError):
msg.assert_socks5()
def test_message_ipv4():
@@ -178,12 +181,15 @@ def test_message_ipv6():
def test_message_invalid_rsv():
raw = tutils.treader(b"\x05\x01\xFF\x01\x7f\x00\x00\x01\xDE\xAD\xBE\xEF")
tutils.raises(socks.SocksError, socks.Message.from_file, raw)
with pytest.raises(socks.SocksError):
socks.Message.from_file(raw)
def test_message_unknown_atyp():
raw = tutils.treader(b"\x05\x02\x00\x02\x7f\x00\x00\x01\xDE\xAD\xBE\xEF")
tutils.raises(socks.SocksError, socks.Message.from_file, raw)
with pytest.raises(socks.SocksError):
socks.Message.from_file(raw)
m = socks.Message(5, 1, 0x02, tcp.Address(("example.com", 5050)))
tutils.raises(socks.SocksError, m.to_file, BytesIO())
with pytest.raises(socks.SocksError):
m.to_file(BytesIO())

77
test/mitmproxy/net/test_tcp.py
View File

@@ -197,14 +197,15 @@ class TestSSLv3Only(tservers.ServerTestBase):
def test_failure(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
tutils.raises(exceptions.TlsException, c.convert_to_ssl, sni="foo.com")
with pytest.raises(exceptions.TlsException):
c.convert_to_ssl(sni="foo.com")
class TestInvalidTrustFile(tservers.ServerTestBase):
def test_invalid_trust_file_should_fail(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
with tutils.raises(exceptions.TlsException):
with pytest.raises(exceptions.TlsException):
c.convert_to_ssl(
sni="example.mitmproxy.org",
verify_options=SSL.VERIFY_PEER,
@@ -250,7 +251,7 @@ class TestSSLUpstreamCertVerificationWBadServerCert(tservers.ServerTestBase):
def test_mode_strict_should_fail(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
with tutils.raises(exceptions.InvalidCertificateException):
with pytest.raises(exceptions.InvalidCertificateException):
c.convert_to_ssl(
sni="example.mitmproxy.org",
verify_options=SSL.VERIFY_PEER,
@@ -275,7 +276,7 @@ class TestSSLUpstreamCertVerificationWBadHostname(tservers.ServerTestBase):
def test_should_fail_without_sni(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
with tutils.raises(exceptions.TlsException):
with pytest.raises(exceptions.TlsException):
c.convert_to_ssl(
verify_options=SSL.VERIFY_PEER,
ca_pemfile=tutils.test_data.path("mitmproxy/net/data/verificationcerts/trusted-root.crt")
@@ -284,7 +285,7 @@ class TestSSLUpstreamCertVerificationWBadHostname(tservers.ServerTestBase):
def test_should_fail(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
with tutils.raises(exceptions.InvalidCertificateException):
with pytest.raises(exceptions.InvalidCertificateException):
c.convert_to_ssl(
sni="mitmproxy.org",
verify_options=SSL.VERIFY_PEER,
@@ -361,11 +362,8 @@ class TestSSLClientCert(tservers.ServerTestBase):
def test_clientcert_err(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
tutils.raises(
exceptions.TlsException,
c.convert_to_ssl,
cert=tutils.test_data.path("mitmproxy/net/data/clientcert/make")
)
with pytest.raises(exceptions.TlsException):
c.convert_to_ssl(cert=tutils.test_data.path("mitmproxy/net/data/clientcert/make"))
class TestSNI(tservers.ServerTestBase):
@@ -432,7 +430,8 @@ class TestServerCipherListError(tservers.ServerTestBase):
def test_echo(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
tutils.raises("handshake error", c.convert_to_ssl, sni="foo.com")
with pytest.raises("handshake error"):
c.convert_to_ssl(sni="foo.com")
class TestClientCipherListError(tservers.ServerTestBase):
@@ -444,12 +443,8 @@ class TestClientCipherListError(tservers.ServerTestBase):
def test_echo(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
tutils.raises(
"cipher specification",
c.convert_to_ssl,
sni="foo.com",
cipher_list="bogus"
)
with pytest.raises("cipher specification"):
c.convert_to_ssl(sni="foo.com", cipher_list="bogus")
class TestSSLDisconnect(tservers.ServerTestBase):
@@ -468,8 +463,10 @@ class TestSSLDisconnect(tservers.ServerTestBase):
# Excercise SSL.ZeroReturnError
c.rfile.read(10)
c.close()
tutils.raises(exceptions.TcpDisconnect, c.wfile.write, b"foo")
tutils.raises(queue.Empty, self.q.get_nowait)
with pytest.raises(exceptions.TcpDisconnect):
c.wfile.write(b"foo")
with pytest.raises(queue.Empty):
self.q.get_nowait()
class TestSSLHardDisconnect(tservers.ServerTestBase):
@@ -483,7 +480,8 @@ class TestSSLHardDisconnect(tservers.ServerTestBase):
# Exercise SSL.SysCallError
c.rfile.read(10)
c.close()
tutils.raises(exceptions.TcpDisconnect, c.wfile.write, b"foo")
with pytest.raises(exceptions.TcpDisconnect):
c.wfile.write(b"foo")
class TestDisconnect(tservers.ServerTestBase):
@@ -524,7 +522,8 @@ class TestTimeOut(tservers.ServerTestBase):
with c.connect():
c.settimeout(0.1)
assert c.gettimeout() == 0.1
tutils.raises(exceptions.TcpTimeout, c.rfile.read, 10)
with pytest.raises(exceptions.TcpTimeout):
c.rfile.read(10)
class TestCryptographyALPN:
@@ -583,7 +582,8 @@ class TestSSLTimeOut(tservers.ServerTestBase):
with c.connect():
c.convert_to_ssl()
c.settimeout(0.1)
tutils.raises(exceptions.TcpTimeout, c.rfile.read, 10)
with pytest.raises(exceptions.TcpTimeout):
c.rfile.read(10)
class TestDHParams(tservers.ServerTestBase):
@@ -613,7 +613,8 @@ class TestTCPClient:
def test_conerr(self):
c = tcp.TCPClient(("127.0.0.1", 0))
tutils.raises(exceptions.TcpException, c.connect)
with pytest.raises(exceptions.TcpException):
c.connect()
class TestFileLike:
@@ -664,7 +665,8 @@ class TestFileLike:
s.read(1)
assert s.get_log() == b"o"
s.stop_log()
tutils.raises(ValueError, s.get_log)
with pytest.raises(ValueError):
s.get_log()
def test_writelog(self):
s = BytesIO()
@@ -682,7 +684,8 @@ class TestFileLike:
o = mock.MagicMock()
o.flush = mock.MagicMock(side_effect=socket.error)
s.o = o
tutils.raises(exceptions.TcpDisconnect, s.flush)
with pytest.raises(exceptions.TcpDisconnect):
s.flush()
def test_reader_read_error(self):
s = BytesIO(b"foobar\nfoobar")
@@ -690,7 +693,8 @@ class TestFileLike:
o = mock.MagicMock()
o.read = mock.MagicMock(side_effect=socket.error)
s.o = o
tutils.raises(exceptions.TcpDisconnect, s.read, 10)
with pytest.raises(exceptions.TcpDisconnect):
s.read(10)
def test_reset_timestamps(self):
s = BytesIO(b"foobar\nfoobar")
@@ -721,24 +725,28 @@ class TestFileLike:
s = mock.MagicMock()
s.read = mock.MagicMock(side_effect=SSL.Error())
s = tcp.Reader(s)
tutils.raises(exceptions.TlsException, s.read, 1)
with pytest.raises(exceptions.TlsException):
s.read(1)
def test_read_syscall_ssl_error(self):
s = mock.MagicMock()
s.read = mock.MagicMock(side_effect=SSL.SysCallError())
s = tcp.Reader(s)
tutils.raises(exceptions.TlsException, s.read, 1)
with pytest.raises(exceptions.TlsException):
s.read(1)
def test_reader_readline_disconnect(self):
o = mock.MagicMock()
o.read = mock.MagicMock(side_effect=socket.error)
s = tcp.Reader(o)
tutils.raises(exceptions.TcpDisconnect, s.readline, 10)
with pytest.raises(exceptions.TcpDisconnect):
s.readline(10)
def test_reader_incomplete_error(self):
s = BytesIO(b"foobar")
s = tcp.Reader(s)
tutils.raises(exceptions.TcpReadIncomplete, s.safe_read, 10)
with pytest.raises(exceptions.TcpReadIncomplete):
s.safe_read(10)
class TestPeek(tservers.ServerTestBase):
@@ -759,7 +767,7 @@ class TestPeek(tservers.ServerTestBase):
assert c.rfile.readline() == testval
c.close()
with tutils.raises(exceptions.NetlibException):
with pytest.raises(exceptions.NetlibException):
if c.rfile.peek(1) == b"":
# Workaround for Python 2 on Unix:
# Peeking a closed connection does not raise an exception here.
@@ -830,8 +838,5 @@ class TestSSLInvalidMethod(tservers.ServerTestBase):
fake_ssl_method = 100500
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
tutils.raises(
exceptions.TlsException,
c.convert_to_ssl,
method=fake_ssl_method
)
with pytest.raises(exceptions.TlsException):
c.convert_to_ssl(method=fake_ssl_method)

6
test/mitmproxy/net/websockets/test_frame.py
View File

@@ -108,8 +108,10 @@ class TestFrameHeader:
assert not f2.mask
def test_violations(self):
tutils.raises("opcode", websockets.FrameHeader, opcode=17)
tutils.raises("masking key", websockets.FrameHeader, masking_key=b"x")
with pytest.raises("opcode"):
websockets.FrameHeader(opcode=17)
with pytest.raises("masking key"):
websockets.FrameHeader(masking_key=b"x")
def test_automask(self):
f = websockets.FrameHeader(mask=True)

20
test/mitmproxy/test_controller.py
View File

@@ -1,14 +1,12 @@
from threading import Thread, Event
from unittest.mock import Mock
from mitmproxy import controller
import queue
import pytest
from mitmproxy.exceptions import Kill, ControlException
from mitmproxy import proxy
from mitmproxy import controller
from mitmproxy import master
from mitmproxy.test import tutils
from mitmproxy import proxy
class TMsg:
@@ -80,7 +78,7 @@ class TestChannel:
done = Event()
done.set()
channel = controller.Channel(q, done)
with tutils.raises(Kill):
with pytest.raises(Kill):
channel.ask("test", Mock(name="test_ask_shutdown"))
@@ -98,7 +96,7 @@ class TestReply:
reply.take()
assert reply.state == "taken"
with tutils.raises(queue.Empty):
with pytest.raises(queue.Empty):
reply.q.get_nowait()
reply.commit()
assert reply.state == "committed"
@@ -132,7 +130,7 @@ class TestReply:
reply = controller.Reply(46)
reply.handle()
reply.take()
with tutils.raises(ControlException):
with pytest.raises(ControlException):
reply.commit()
reply.ack()
reply.commit()
@@ -141,7 +139,7 @@ class TestReply:
reply = controller.Reply(47)
reply.handle()
reply.send(1)
with tutils.raises(ControlException):
with pytest.raises(ControlException):
reply.send(2)
reply.take()
reply.commit()
@@ -163,13 +161,13 @@ class TestReply:
if state in ok:
getattr(r, fn)()
else:
with tutils.raises(ControlException):
with pytest.raises(ControlException):
getattr(r, fn)()
r._state = "committed" # hide warnings on deletion
def test_del(self):
reply = controller.Reply(47)
with tutils.raises(ControlException):
with pytest.raises(ControlException):
reply.__del__()
reply.handle()
reply.ack()

10
test/mitmproxy/test_examples.py
View File

@@ -1,15 +1,15 @@
import json
from mitmproxy.test import tflow
import os
import shlex
import pytest
from mitmproxy import options
from mitmproxy import contentviews
from mitmproxy import proxy
from mitmproxy.addons import script
from mitmproxy import master
from mitmproxy.addons import script
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.net.http import Headers
from mitmproxy.net.http import cookies
@@ -52,7 +52,7 @@ class TestScripts(mastertest.MasterTest):
assert any(b'tEST!' in val[0][1] for val in fmt)
def test_iframe_injector(self):
with tutils.raises(ScriptError):
with pytest.raises(ScriptError):
tscript("simple/modify_body_inject_iframe.py")
m, sc = tscript("simple/modify_body_inject_iframe.py", "http://example.org/evil_iframe")
@@ -141,7 +141,7 @@ class TestHARDump:
)
def test_no_file_arg(self):
with tutils.raises(ScriptError):
with pytest.raises(ScriptError):
tscript("complex/har_dump.py")
def test_simple(self):

28
test/mitmproxy/test_flow.py
View File

@@ -1,7 +1,7 @@
from mitmproxy.test import tflow
import io
import pytest
from mitmproxy.test import tutils
from mitmproxy.test import tflow
from mitmproxy.net.http import Headers
import mitmproxy.io
from mitmproxy import flowfilter, options
@@ -54,7 +54,8 @@ class TestHTTPFlow:
f = tflow.tflow(err=True)
assert flowfilter.match("~e", f)
tutils.raises(ValueError, flowfilter.match, "~", f)
with pytest.raises(ValueError):
flowfilter.match("~", f)
def test_backup(self):
f = tflow.tflow()
@@ -166,7 +167,8 @@ class TestTCPFlow:
f = tflow.ttcpflow(err=True)
assert flowfilter.match("~e", f)
tutils.raises(ValueError, flowfilter.match, "~", f)
with pytest.raises(ValueError):
flowfilter.match("~", f)
class TestSerialize:
@@ -249,7 +251,8 @@ class TestSerialize:
sio.write(b"bogus")
sio.seek(0)
r = mitmproxy.io.FlowReader(sio)
tutils.raises(FlowReadException, list, r.stream())
with pytest.raises(FlowReadException):
list(r.stream())
f = FlowReadException("foo")
assert str(f) == "foo"
@@ -263,7 +266,8 @@ class TestSerialize:
sio.seek(0)
r = mitmproxy.io.FlowReader(sio)
tutils.raises("version", list, r.stream())
with pytest.raises("version"):
list(r.stream())
class TestFlowMaster:
@@ -272,13 +276,16 @@ class TestFlowMaster:
fm = master.Master(None, DummyServer())
f = tflow.tflow(resp=True)
f.request.content = None
tutils.raises("missing", fm.replay_request, f)
with pytest.raises("missing"):
fm.replay_request(f)
f.intercepted = True
tutils.raises("intercepted", fm.replay_request, f)
with pytest.raises("intercepted"):
fm.replay_request(f)
f.live = True
tutils.raises("live", fm.replay_request, f)
with pytest.raises("live"):
fm.replay_request(f)
def test_create_flow(self):
fm = master.Master(None, DummyServer())
@@ -313,7 +320,8 @@ class TestRequest:
r = f.request
u = r.url
r.url = u
tutils.raises(ValueError, setattr, r, "url", "")
with pytest.raises(ValueError):
setattr(r, "url", "")
assert r.url == u
r2 = r.copy()
assert r.get_state() == r2.get_state()

4
test/mitmproxy/test_flow_format_compat.py
View File

@@ -1,3 +1,5 @@
import pytest
from mitmproxy import io
from mitmproxy import exceptions
from mitmproxy.test import tutils
@@ -22,5 +24,5 @@ def test_load_018():
def test_cannot_convert():
with open(tutils.test_data.path("mitmproxy/data/dumpfile-010"), "rb") as f:
flow_reader = io.FlowReader(f)
with tutils.raises(exceptions.FlowReadException):
with pytest.raises(exceptions.FlowReadException):
list(flow_reader.stream())

5
test/mitmproxy/test_flowfilter.py
View File

@@ -4,7 +4,6 @@ from unittest.mock import patch
from mitmproxy.test import tflow
from mitmproxy import flowfilter
from . import tutils as ttutils
class TestParsing:
@@ -382,10 +381,10 @@ class TestMatchingTCPFlow:
class TestMatchingDummyFlow:
def flow(self):
return ttutils.tdummyflow()
return tflow.tdummyflow()
def err(self):
return ttutils.tdummyflow(err=True)
return tflow.tdummyflow(err=True)
def q(self, q, o):
return flowfilter.parse(q)(o)

17
test/mitmproxy/test_optmanager.py
View File

@@ -1,5 +1,6 @@
import copy
import os
import pytest
from mitmproxy import options
from mitmproxy import optmanager
@@ -68,11 +69,11 @@ def test_options():
o.one = "one"
assert o.one == "one"
with tutils.raises(TypeError):
with pytest.raises(TypeError):
TO(nonexistent = "value")
with tutils.raises("no such option"):
with pytest.raises("no such option"):
o.nonexistent = "value"
with tutils.raises("no such option"):
with pytest.raises("no such option"):
o.update(nonexistent = "value")
rec = []
@@ -96,7 +97,7 @@ def test_setter():
f = o.setter("two")
f("xxx")
assert o.two == "xxx"
with tutils.raises("no such option"):
with pytest.raises("no such option"):
o.setter("nonexistent")
@@ -107,7 +108,7 @@ def test_toggler():
assert o.two is False
f()
assert o.two is True
with tutils.raises("no such option"):
with pytest.raises("no such option"):
o.toggler("nonexistent")
@@ -192,10 +193,12 @@ def test_serialize():
assert o2 == o
t = "invalid: foo\ninvalid"
tutils.raises("config error", o2.load, t)
with pytest.raises("config error"):
o2.load(t)
t = "invalid"
tutils.raises("config error", o2.load, t)
with pytest.raises("config error"):
o2.load(t)
t = ""
o2.load(t)

17
test/mitmproxy/test_platform_pf.py
View File

@@ -1,4 +1,5 @@
import sys
import pytest
from mitmproxy.platform import pf
from mitmproxy.test import tutils
@@ -13,15 +14,7 @@ class TestLookup:
p = tutils.test_data.path("mitmproxy/data/pf01")
d = open(p, "rb").read()
assert pf.lookup("192.168.1.111", 40000, d) == ("5.5.5.5", 80)
tutils.raises(
"Could not resolve original destination",
pf.lookup,
"192.168.1.112",
40000,
d)
tutils.raises(
"Could not resolve original destination",
pf.lookup,
"192.168.1.111",
40001,
d)
with pytest.raises("Could not resolve original destination"):
pf.lookup("192.168.1.112", 40000, d)
with pytest.raises("Could not resolve original destination"):
pf.lookup("192.168.1.111", 40001, d)

64
test/mitmproxy/test_proxy.py
View File

@@ -1,8 +1,10 @@
from mitmproxy.test import tflow
import os
from unittest import mock
import argparse
from unittest import mock
from OpenSSL import SSL
import pytest
from mitmproxy.test import tflow
from mitmproxy.tools import cmdline
from mitmproxy import options
@@ -15,7 +17,7 @@ from pathod import test
from mitmproxy.net.http import http1
from mitmproxy.test import tutils
from . import tutils as ttutils
from ..conftest import skip_windows
class TestServerConnection:
@@ -80,9 +82,6 @@ class TestProcessProxyOptions:
pconf = config.ProxyConfig(opts)
return parser, pconf
def assert_err(self, err, *args):
tutils.raises(err, self.p, *args)
def assert_noerr(self, *args):
m, p = self.p(*args)
assert p
@@ -97,22 +96,28 @@ class TestProcessProxyOptions:
@mock.patch("mitmproxy.platform.original_addr", None)
def test_no_transparent(self):
self.assert_err("transparent mode not supported", "-T")
with pytest.raises("transparent mode not supported"):
self.p("-T")
@mock.patch("mitmproxy.platform.original_addr")
def test_modes(self, _):
self.assert_noerr("-R", "http://localhost")
self.assert_err("expected one argument", "-R")
self.assert_err("Invalid server specification", "-R", "reverse")
with pytest.raises("expected one argument"):
self.p("-R")
with pytest.raises("Invalid server specification"):
self.p("-R", "reverse")
self.assert_noerr("-T")
self.assert_noerr("-U", "http://localhost")
self.assert_err("Invalid server specification", "-U", "upstream")
with pytest.raises("Invalid server specification"):
self.p("-U", "upstream")
self.assert_noerr("--upstream-auth", "test:test")
self.assert_err("expected one argument", "--upstream-auth")
self.assert_err("mutually exclusive", "-R", "http://localhost", "-T")
with pytest.raises("expected one argument"):
self.p("--upstream-auth")
with pytest.raises("mutually exclusive"):
self.p("-R", "http://localhost", "-T")
def test_client_certs(self):
with tutils.tmpdir() as cadir:
@@ -120,16 +125,15 @@ class TestProcessProxyOptions:
self.assert_noerr(
"--client-certs",
os.path.join(tutils.test_data.path("mitmproxy/data/clientcert"), "client.pem"))
self.assert_err(
"path does not exist",
"--client-certs",
"nonexistent")
with pytest.raises("path does not exist"):
self.p("--client-certs", "nonexistent")
def test_certs(self):
self.assert_noerr(
"--cert",
tutils.test_data.path("mitmproxy/data/testkey.pem"))
self.assert_err("does not exist", "--cert", "nonexistent")
with pytest.raises("does not exist"):
self.p("--cert", "nonexistent")
def test_insecure(self):
p = self.assert_noerr("--insecure")
@@ -147,20 +151,18 @@ class TestProcessProxyOptions:
class TestProxyServer:
# binding to 0.0.0.0:1 works without special permissions on Windows
@ttutils.skip_windows
@skip_windows
def test_err(self):
conf = ProxyConfig(
options.Options(listen_port=1),
)
tutils.raises("error starting proxy server", ProxyServer, conf)
# binding to 0.0.0.0:1 works without special permissions on Windows
conf = ProxyConfig(options.Options(listen_port=1))
with pytest.raises("error starting proxy server"):
ProxyServer(conf)
def test_err_2(self):
conf = ProxyConfig(
options.Options(listen_host="invalidhost"),
)
tutils.raises("error starting proxy server", ProxyServer, conf)
conf = ProxyConfig(options.Options(listen_host="invalidhost"))
with pytest.raises("error starting proxy server"):
ProxyServer(conf)
class TestDummyServer:
@@ -173,7 +175,7 @@ class TestDummyServer:
class TestConnectionHandler:
def test_fatal_error(self):
def test_fatal_error(self, capsys):
config = mock.Mock()
root_layer = mock.Mock()
root_layer.side_effect = RuntimeError
@@ -189,5 +191,7 @@ class TestConnectionHandler:
config,
channel
)
with ttutils.capture_stderr(c.handle) as output:
assert "mitmproxy has crashed" in output
c.handle()
_, err = capsys.readouterr()
assert "mitmproxy has crashed" in err

21
test/mitmproxy/test_proxy_config.py
View File

@@ -1,11 +1,10 @@
from mitmproxy.test import tutils
import pytest
from mitmproxy.proxy import config
def test_parse_server_spec():
tutils.raises(
"Invalid server specification", config.parse_server_spec, ""
)
with pytest.raises("Invalid server specification"):
config.parse_server_spec("")
assert config.parse_server_spec("http://foo.com:88") == (
"http", ("foo.com", 88)
)
@@ -15,13 +14,7 @@ def test_parse_server_spec():
assert config.parse_server_spec("https://foo.com") == (
"https", ("foo.com", 443)
)
tutils.raises(
"Invalid server specification",
config.parse_server_spec,
"foo.com"
)
tutils.raises(
"Invalid server specification",
config.parse_server_spec,
"http://"
)
with pytest.raises("Invalid server specification"):
config.parse_server_spec("foo.com")
with pytest.raises("Invalid server specification"):
config.parse_server_spec("http://")

25
test/mitmproxy/test_server.py
View File

@@ -1,7 +1,7 @@
import os
import socket
import time
import pytest
from unittest import mock
from mitmproxy.test import tutils
@@ -21,9 +21,9 @@ from mitmproxy.net.tcp import Address
from pathod import pathoc
from pathod import pathod
from . import tutils as ttutils
from . import tservers
from ..conftest import skip_appveyor
"""
Note that the choice of response code in these tests matters more than you
@@ -159,7 +159,7 @@ class TcpMixin:
# mitmproxy responds with bad gateway
assert self.pathod(spec).status_code == 502
self._ignore_on()
with tutils.raises(exceptions.HttpException):
with pytest.raises(exceptions.HttpException):
self.pathod(spec) # pathoc tries to parse answer as HTTP
self._ignore_off()
@@ -238,7 +238,7 @@ class TestHTTP(tservers.HTTPProxyTest, CommonMixin):
# There's a race here, which means we can get any of a number of errors.
# Rather than introduce yet another sleep into the test suite, we just
# relax the Exception specification.
with tutils.raises(Exception):
with pytest.raises(Exception):
p.request("get:'%s'" % response)
def test_reconnect(self):
@@ -611,7 +611,7 @@ class TestProxy(tservers.HTTPProxyTest):
assert "host" in f.request.headers
assert f.response.status_code == 304
@ttutils.skip_appveyor
@skip_appveyor
def test_response_timestamps(self):
# test that we notice at least 1 sec delay between timestamps
# in response object
@@ -622,7 +622,7 @@ class TestProxy(tservers.HTTPProxyTest):
# timestamp_start might fire a bit late, so we play safe and only require 300ms.
assert 0.3 <= response.timestamp_end - response.timestamp_start
@ttutils.skip_appveyor
@skip_appveyor
def test_request_timestamps(self):
# test that we notice a delay between timestamps in request object
connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -857,7 +857,7 @@ class TestKillRequest(tservers.HTTPProxyTest):
masterclass = MasterKillRequest
def test_kill(self):
with tutils.raises(exceptions.HttpReadDisconnect):
with pytest.raises(exceptions.HttpReadDisconnect):
self.pathod("200")
# Nothing should have hit the server
assert not self.server.last_log()
@@ -874,7 +874,7 @@ class TestKillResponse(tservers.HTTPProxyTest):
masterclass = MasterKillResponse
def test_kill(self):
with tutils.raises(exceptions.HttpReadDisconnect):
with pytest.raises(exceptions.HttpReadDisconnect):
self.pathod("200")
# The server should have seen a request
assert self.server.last_log()
@@ -1027,11 +1027,8 @@ class TestProxyChainingSSLReconnect(tservers.HTTPUpstreamProxyTest):
assert not self.chain[1].tmaster.state.flows[-2].response
# Reconnection failed, so we're now disconnected
tutils.raises(
exceptions.HttpException,
p.request,
"get:'/p/418:b\"content3\"'"
)
with pytest.raises(exceptions.HttpException):
p.request("get:'/p/418:b\"content3\"'")
class AddUpstreamCertsToClientChainMixin:

12
test/mitmproxy/test_tools_dump.py
View File

@@ -25,14 +25,10 @@ class TestDumpMaster(mastertest.MasterTest):
self.mkmaster(None, rfile=p),
1, b"",
)
tutils.raises(
dump.DumpError,
self.mkmaster, None, rfile="/nonexistent"
)
tutils.raises(
dump.DumpError,
self.mkmaster, None, rfile="test_dump.py"
)
with pytest.raises(dump.DumpError):
self.mkmaster(None, rfile="/nonexistent")
with pytest.raises(dump.DumpError):
self.mkmaster(None, rfile="test_dump.py")
def test_has_error(self):
m = self.mkmaster(None)

8
test/mitmproxy/test_types_bidi.py
View File

@@ -1,5 +1,5 @@
import pytest
from mitmproxy.types import bidi
from mitmproxy.test import tutils
def test_bidi():
@@ -7,5 +7,7 @@ def test_bidi():
assert b.a == 1
assert b.get_name(1) == "a"
assert b.get_name(5) is None
tutils.raises(AttributeError, getattr, b, "c")
tutils.raises(ValueError, bidi.BiDi, one=1, two=1)
with pytest.raises(AttributeError):
getattr(b, "c")
with pytest.raises(ValueError):
bidi.BiDi(one=1, two=1)

9
test/mitmproxy/test_types_multidict.py
View File

@@ -1,4 +1,5 @@
from mitmproxy.test import tutils
import pytest
from mitmproxy.types import multidict
@@ -40,7 +41,7 @@ class TestMultiDict:
assert "Foo" in md
assert md["foo"] == "bar"
with tutils.raises(KeyError):
with pytest.raises(KeyError):
assert md["bar"]
md_multi = TMultiDict(
@@ -65,7 +66,7 @@ class TestMultiDict:
assert "foo" not in md
assert "bar" in md
with tutils.raises(KeyError):
with pytest.raises(KeyError):
del md["foo"]
del md["bar"]
@@ -103,7 +104,7 @@ class TestMultiDict:
it should not implement __hash__(), since the implementation of hashable
collections requires that a key's hash value is immutable.
"""
with tutils.raises(TypeError):
with pytest.raises(TypeError):
assert hash(TMultiDict())
def test_get_all(self):

68
test/mitmproxy/tutils.py
View File

@@ -1,68 +0,0 @@
import sys
from contextlib import contextmanager
from unittest.case import SkipTest
import io
import mitmproxy.test.tutils
import os
from mitmproxy import controller
from mitmproxy import flow
import mitmproxy.test.tflow
def _skip_windows(*args):
raise SkipTest("Skipped on Windows.")
def skip_windows(fn):
if os.name == "nt":
return _skip_windows
else:
return fn
def skip_not_windows(fn):
if os.name == "nt":
return fn
else:
return _skip_windows
def _skip_appveyor(*args):
raise SkipTest("Skipped on AppVeyor.")
def skip_appveyor(fn):
if "APPVEYOR" in os.environ:
return _skip_appveyor
else:
return fn
class DummyFlow(flow.Flow):
"""A flow that is neither HTTP nor TCP."""
def __init__(self, client_conn, server_conn, live=None):
super().__init__("dummy", client_conn, server_conn, live)
def tdummyflow(client_conn=True, server_conn=True, err=None):
if client_conn is True:
client_conn = mitmproxy.test.tflow.tclient_conn()
if server_conn is True:
server_conn = mitmproxy.test.tflow.tserver_conn()
if err is True:
err = mitmproxy.test.tflow.terr()
f = DummyFlow(client_conn, server_conn)
f.error = err
f.reply = controller.DummyReply()
return f
@contextmanager
def capture_stderr(command, *args, **kwargs):
out, sys.stderr = sys.stderr, io.StringIO()
command(*args, **kwargs)
yield sys.stderr.getvalue()
sys.stderr = out

8
test/mitmproxy/utils/test_human.py
View File

@@ -1,6 +1,6 @@
import time
import pytest
from mitmproxy.utils import human
from mitmproxy.test import tutils
def test_format_timestamp():
@@ -18,8 +18,10 @@ def test_parse_size():
assert human.parse_size("1k") == 1024
assert human.parse_size("1m") == 1024**2
assert human.parse_size("1g") == 1024**3
tutils.raises(ValueError, human.parse_size, "1f")
tutils.raises(ValueError, human.parse_size, "ak")
with pytest.raises(ValueError):
human.parse_size("1f")
with pytest.raises(ValueError):
human.parse_size("ak")
def test_pretty_size():

15
test/mitmproxy/utils/test_strutils.py
View File

@@ -1,18 +1,19 @@
import pytest
from mitmproxy.utils import strutils
from mitmproxy.test import tutils
def test_always_bytes():
assert strutils.always_bytes(bytes(range(256))) == bytes(range(256))
assert strutils.always_bytes("foo") == b"foo"
with tutils.raises(ValueError):
with pytest.raises(ValueError):
strutils.always_bytes(u"\u2605", "ascii")
with tutils.raises(TypeError):
with pytest.raises(TypeError):
strutils.always_bytes(42, "ascii")
def test_always_str():
with tutils.raises(TypeError):
with pytest.raises(TypeError):
strutils.always_str(42)
assert strutils.always_str("foo") == "foo"
assert strutils.always_str(b"foo") == "foo"
@@ -36,7 +37,7 @@ def test_escape_control_characters():
u'=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~.'
)
with tutils.raises(ValueError):
with pytest.raises(ValueError):
strutils.escape_control_characters(b"foo")
@@ -59,7 +60,7 @@ def test_bytes_to_escaped_str():
assert strutils.bytes_to_escaped_str(b"\\\n", True) == "\\ \\ \n".replace(" ", "")
assert strutils.bytes_to_escaped_str(b"\\\\n", True) == "\\ \\ \\ \\ n".replace(" ", "")
with tutils.raises(ValueError):
with pytest.raises(ValueError):
strutils.bytes_to_escaped_str(u"such unicode")
@@ -71,7 +72,7 @@ def test_escaped_str_to_bytes():
assert strutils.escaped_str_to_bytes(u"&!?=\\\\)") == br"&!?=\)"
assert strutils.escaped_str_to_bytes(u"\u00fc") == b'\xc3\xbc'
with tutils.raises(ValueError):
with pytest.raises(ValueError):
strutils.escaped_str_to_bytes(b"very byte")

41
test/pathod/test_language_base.py
View File

@@ -1,4 +1,6 @@
import os
import pytest
from pathod import language
from pathod.language import base, exceptions
@@ -145,23 +147,14 @@ class TestTokValueFile:
assert v.get_generator(language.Settings(staticdir=t))
v = base.TokValue.parseString("<path2")[0]
tutils.raises(
exceptions.FileAccessDenied,
v.get_generator,
language.Settings(staticdir=t)
)
tutils.raises(
"access disabled",
v.get_generator,
language.Settings()
)
with pytest.raises(exceptions.FileAccessDenied):
v.get_generator(language.Settings(staticdir=t))
with pytest.raises("access disabled"):
v.get_generator(language.Settings())
v = base.TokValue.parseString("</outside")[0]
tutils.raises(
"outside",
v.get_generator,
language.Settings(staticdir=t)
)
with pytest.raises("outside"):
v.get_generator(language.Settings(staticdir=t))
def test_spec(self):
v = base.TokValue.parseString("<'one two'")[0]
@@ -208,8 +201,10 @@ class TestMisc:
e = TT.expr()
assert e.parseString("m@4")
tutils.raises("invalid value length", e.parseString, "m@100")
tutils.raises("invalid value length", e.parseString, "m@1")
with pytest.raises("invalid value length"):
e.parseString("m@100")
with pytest.raises("invalid value length"):
e.parseString("m@1")
with tutils.tmpdir() as t:
p = os.path.join(t, "path")
@@ -217,7 +212,8 @@ class TestMisc:
with open(p, "wb") as f:
f.write(b"a" * 20)
v = e.parseString("m<path")[0]
tutils.raises("invalid value length", v.values, s)
with pytest.raises("invalid value length"):
v.values(s)
p = os.path.join(t, "path")
with open(p, "wb") as f:
@@ -286,7 +282,8 @@ def test_intfield():
assert v.value == 4
assert v.spec() == "t4"
tutils.raises("can't exceed", e.parseString, "t5")
with pytest.raises("can't exceed"):
e.parseString("t5")
def test_options_or_value():
@@ -327,8 +324,10 @@ def test_integer():
class BInt(base.Integer):
bounds = (1, 5)
tutils.raises("must be between", BInt, 0)
tutils.raises("must be between", BInt, 6)
with pytest.raises("must be between"):
BInt(0)
with pytest.raises("must be between"):
BInt(6)
assert BInt(5)
assert BInt(1)
assert BInt(3)

34
test/pathod/test_language_http.py
View File

@@ -1,8 +1,9 @@
import io
import pytest
from pathod import language
from pathod.language import http, base
from mitmproxy.test import tutils
from . import tservers
@@ -19,10 +20,12 @@ def test_make_error_response():
class TestRequest:
def test_nonascii(self):
tutils.raises("ascii", parse_request, "get:\xf0")
with pytest.raises("ascii"):
parse_request("get:\xf0")
def test_err(self):
tutils.raises(language.ParseException, parse_request, 'GET')
with pytest.raises(language.ParseException):
parse_request('GET')
def test_simple(self):
r = parse_request('GET:"/foo"')
@@ -214,9 +217,8 @@ class TestResponse:
testlen(r)
def test_parse_err(self):
tutils.raises(
language.ParseException, language.parse_pathod, "400:msg,b:"
)
with pytest.raises(language.ParseException):
language.parse_pathod("400:msg,b:")
try:
language.parse_pathod("400'msg':b:")
except language.ParseException as v:
@@ -224,7 +226,8 @@ class TestResponse:
assert str(v)
def test_nonascii(self):
tutils.raises("ascii", language.parse_pathod, "foo:b\xf0")
with pytest.raises("ascii"):
language.parse_pathod("foo:b\xf0")
def test_parse_header(self):
r = next(language.parse_pathod('400:h"foo"="bar"'))
@@ -260,7 +263,8 @@ class TestResponse:
def test_websockets(self):
r = next(language.parse_pathod("ws"))
tutils.raises("no websocket key", r.resolve, language.Settings())
with pytest.raises("no websocket key"):
r.resolve(language.Settings())
res = r.resolve(language.Settings(websocket_key=b"foo"))
assert res.status_code.string() == b"101"
@@ -327,11 +331,8 @@ def test_nested_response():
e = http.NestedResponse.expr()
v = e.parseString("s'200'")[0]
assert v.value.val == b"200"
tutils.raises(
language.ParseException,
e.parseString,
"s'foo'"
)
with pytest.raises(language.ParseException):
e.parseString("s'foo'")
v = e.parseString('s"200:b@1"')[0]
assert "@1" in v.spec()
@@ -350,8 +351,5 @@ def test_nested_response_freeze():
def test_unique_components():
tutils.raises(
"multiple body clauses",
language.parse_pathod,
"400:b@1:b@1"
)
with pytest.raises("multiple body clauses"):
language.parse_pathod("400:b@1:b@1")

15
test/pathod/test_language_http2.py
View File

@@ -1,4 +1,5 @@
import io
import pytest
from mitmproxy.net import tcp
from mitmproxy.net.http import user_agents
@@ -7,8 +8,6 @@ from pathod import language
from pathod.language import http2
from pathod.protocols.http2 import HTTP2StateProtocol
from mitmproxy.test import tutils
def parse_request(s):
return next(language.parse_pathoc(s, True))
@@ -40,10 +39,12 @@ class TestRequest:
assert req.values(default_settings()) == req.values(default_settings())
def test_nonascii(self):
tutils.raises("ascii", parse_request, "get:\xf0")
with pytest.raises("ascii"):
parse_request("get:\xf0")
def test_err(self):
tutils.raises(language.ParseException, parse_request, 'GET')
with pytest.raises(language.ParseException):
parse_request('GET')
def test_simple(self):
r = parse_request('GET:"/foo"')
@@ -167,10 +168,12 @@ class TestResponse:
assert res.values(default_settings()) == res.values(default_settings())
def test_nonascii(self):
tutils.raises("ascii", parse_response, "200:\xf0")
with pytest.raises("ascii"):
parse_response("200:\xf0")
def test_err(self):
tutils.raises(language.ParseException, parse_response, 'GET:/')
with pytest.raises(language.ParseException):
parse_response('GET:/')
def test_raw_content_length(self):
r = parse_response('200:r')

15
test/pathod/test_language_websocket.py
View File

@@ -1,8 +1,9 @@
import pytest
from pathod import language
from pathod.language import websockets
import mitmproxy.net.websockets
from mitmproxy.test import tutils
from . import tservers
@@ -45,11 +46,8 @@ class TestWebsocketFrame:
def test_parse_websocket_frames(self):
wf = language.parse_websocket_frame("wf:x10")
assert len(list(wf)) == 10
tutils.raises(
language.ParseException,
language.parse_websocket_frame,
"wf:x"
)
with pytest.raises(language.ParseException):
language.parse_websocket_frame("wf:x")
def test_client_values(self):
specs = [
@@ -132,7 +130,7 @@ class TestWebsocketFrame:
assert frm.payload == b"abc"
def test_knone(self):
with tutils.raises("expected 4 bytes"):
with pytest.raises("expected 4 bytes"):
self.fr("wf:b'foo':mask:knone")
def test_length(self):
@@ -140,4 +138,5 @@ class TestWebsocketFrame:
frm = self.fr("wf:l2:b'foo'")
assert frm.header.payload_length == 2
assert frm.payload == b"fo"
tutils.raises("expected 1024 bytes", self.fr, "wf:l1024:b'foo'")
with pytest.raises("expected 1024 bytes"):
self.fr("wf:l1024:b'foo'")

16
test/pathod/test_pathoc.py
View File

@@ -78,7 +78,8 @@ class TestDaemonSSL(PathocTestDaemon):
ssl=False,
fp=fp
)
tutils.raises(NotImplementedError, c.connect)
with pytest.raises(NotImplementedError):
c.connect()
class TestDaemon(PathocTestDaemon):
@@ -172,12 +173,12 @@ class TestDaemon(PathocTestDaemon):
to = ("foobar", 80)
c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None)
c.rfile, c.wfile = io.BytesIO(), io.BytesIO()
with tutils.raises("connect failed"):
with pytest.raises("connect failed"):
c.http_connect(to)
c.rfile = io.BytesIO(
b"HTTP/1.1 500 OK\r\n"
)
with tutils.raises("connect failed"):
with pytest.raises("connect failed"):
c.http_connect(to)
c.rfile = io.BytesIO(
b"HTTP/1.1 200 OK\r\n"
@@ -188,18 +189,21 @@ class TestDaemon(PathocTestDaemon):
to = ("foobar", 80)
c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None)
c.rfile, c.wfile = tutils.treader(b""), io.BytesIO()
tutils.raises(pathoc.PathocError, c.socks_connect, to)
with pytest.raises(pathoc.PathocError):
c.socks_connect(to)
c.rfile = tutils.treader(
b"\x05\xEE"
)
tutils.raises("SOCKS without authentication", c.socks_connect, ("example.com", 0xDEAD))
with pytest.raises("SOCKS without authentication"):
c.socks_connect(("example.com", 0xDEAD))
c.rfile = tutils.treader(
b"\x05\x00" +
b"\x05\xEE\x00\x03\x0bexample.com\xDE\xAD"
)
tutils.raises("SOCKS server error", c.socks_connect, ("example.com", 0xDEAD))
with pytest.raises("SOCKS server error"):
c.socks_connect(("example.com", 0xDEAD))
c.rfile = tutils.treader(
b"\x05\x00" +

5
test/pathod/test_pathoc_cmdline.py
View File

@@ -1,4 +1,5 @@
import io
import pytest
from unittest import mock
from pathod import pathoc_cmdline as cmdline
@@ -10,7 +11,7 @@ from mitmproxy.test import tutils
def test_pathoc(perror):
assert cmdline.args_pathoc(["pathoc", "foo.com", "get:/"])
s = io.StringIO()
with tutils.raises(SystemExit):
with pytest.raises(SystemExit):
cmdline.args_pathoc(["pathoc", "--show-uas"], s, s)
a = cmdline.args_pathoc(["pathoc", "foo.com:8888", "get:/"])
@@ -57,5 +58,5 @@ def test_pathoc(perror):
)
assert len(list(a.requests)) == 1
with tutils.raises(SystemExit):
with pytest.raises(SystemExit):
cmdline.args_pathoc(["pathoc", "foo.com", "invalid"], s, s)

27
test/pathod/test_pathod.py
View File

@@ -36,7 +36,8 @@ class TestTimeout(tservers.DaemonTests):
# increase test performance
# This is a bodge - we have some platform difference that causes
# different exceptions to be raised here.
tutils.raises(Exception, self.pathoc, ["get:/:p1,1"])
with pytest.raises(Exception):
self.pathoc(["get:/:p1,1"])
assert self.d.last_log()["type"] == "timeout"
@@ -133,7 +134,8 @@ class CommonTests(tservers.DaemonTests):
assert len(self.d.log()) == 0
def test_disconnect(self):
tutils.raises("unexpected eof", self.get, "202:b@100k:d200")
with pytest.raises("unexpected eof"):
self.get("202:b@100k:d200")
def test_parserr(self):
rsp = self.get("400:msg,b:")
@@ -160,17 +162,15 @@ class CommonTests(tservers.DaemonTests):
assert "foo" in l["msg"]
def test_invalid_content_length(self):
tutils.raises(
exceptions.HttpException,
self.pathoc,
["get:/:h'content-length'='foo'"]
)
with pytest.raises(exceptions.HttpException):
self.pathoc(["get:/:h'content-length'='foo'"])
l = self.d.last_log()
assert l["type"] == "error"
assert "Unparseable Content Length" in l["msg"]
def test_invalid_headers(self):
tutils.raises(exceptions.HttpException, self.pathoc, ["get:/:h'\t'='foo'"])
with pytest.raises(exceptions.HttpException):
self.pathoc(["get:/:h'\t'='foo'"])
l = self.d.last_log()
assert l["type"] == "error"
assert "Invalid headers" in l["msg"]
@@ -228,12 +228,8 @@ class TestDaemon(CommonTests):
assert r[0].status_code == 202
def test_connect_err(self):
tutils.raises(
exceptions.HttpException,
self.pathoc,
[r"get:'http://foo.com/p/202':da"],
connect_to=("localhost", self.d.port)
)
with pytest.raises(exceptions.HttpException):
self.pathoc([r"get:'http://foo.com/p/202':da"], connect_to=("localhost", self.d.port))
class TestDaemonSSL(CommonTests):
@@ -245,7 +241,8 @@ class TestDaemonSSL(CommonTests):
c.wbufsize = 0
with c.connect():
c.wfile.write(b"\0\0\0\0")
tutils.raises(exceptions.TlsException, c.convert_to_ssl)
with pytest.raises(exceptions.TlsException):
c.convert_to_ssl()
l = self.d.last_log()
assert l["type"] == "error"
assert "SSL" in l["msg"]

8
test/pathod/test_protocols_http2.py
View File

@@ -1,9 +1,9 @@
from unittest import mock
import codecs
import pytest
import hyperframe
from mitmproxy.net import tcp, http
from mitmproxy.test.tutils import raises
from mitmproxy.net.http import http2
from mitmproxy import exceptions
@@ -95,7 +95,7 @@ class TestCheckALPNMismatch(net_tservers.ServerTestBase):
with c.connect():
c.convert_to_ssl(alpn_protos=[b'h2'])
protocol = HTTP2StateProtocol(c)
with raises(NotImplementedError):
with pytest.raises(NotImplementedError):
protocol.check_alpn()
@@ -132,7 +132,7 @@ class TestPerformServerConnectionPreface(net_tservers.ServerTestBase):
protocol.perform_server_connection_preface()
assert protocol.connection_preface_performed
with raises(exceptions.TcpDisconnect):
with pytest.raises(exceptions.TcpDisconnect):
protocol.perform_server_connection_preface(force=True)

8
test/pathod/test_test.py
View File

@@ -1,5 +1,7 @@
import logging
import requests
import pytest
from pathod import test
from mitmproxy.test import tutils
@@ -17,7 +19,7 @@ class TestDaemonManual:
rsp = requests.get("http://localhost:%s/p/202:da" % d.port)
assert rsp.ok
assert rsp.status_code == 202
with tutils.raises(requests.ConnectionError):
with pytest.raises(requests.ConnectionError):
requests.get("http://localhost:%s/p/202:da" % d.port)
def test_startstop_ssl(self):
@@ -29,7 +31,7 @@ class TestDaemonManual:
assert rsp.ok
assert rsp.status_code == 202
d.shutdown()
with tutils.raises(requests.ConnectionError):
with pytest.raises(requests.ConnectionError):
requests.get("http://localhost:%s/p/202:da" % d.port)
def test_startstop_ssl_explicit(self):
@@ -46,5 +48,5 @@ class TestDaemonManual:
assert rsp.ok
assert rsp.status_code == 202
d.shutdown()
with tutils.raises(requests.ConnectionError):
with pytest.raises(requests.ConnectionError):
requests.get("http://localhost:%s/p/202:da" % d.port)

7
test/pathod/test_utils.py
View File

@@ -1,6 +1,6 @@
from pathod import utils
import pytest
from mitmproxy.test import tutils
from pathod import utils
def test_membool():
@@ -13,4 +13,5 @@ def test_membool():
def test_data_path():
tutils.raises(ValueError, utils.data.path, "nonexistent")
with pytest.raises(ValueError):
utils.data.path("nonexistent")

10
tox.ini
View File

@@ -14,14 +14,10 @@ commands =
pytest --timeout 60 --cov-report='' --cov=mitmproxy --cov=pathod \
--full-cov=mitmproxy/addons/ \
--full-cov=mitmproxy/contentviews/ --no-full-cov=mitmproxy/contentviews/__init__.py --no-full-cov=mitmproxy/contentviews/protobuf.py --no-full-cov=mitmproxy/contentviews/wbxml.py --no-full-cov=mitmproxy/contentviews/xml_html.py \
--full-cov=mitmproxy/net/http/ --no-full-cov=mitmproxy/net/http/cookies.py --no-full-cov=mitmproxy/net/http/encoding.py --no-full-cov=mitmproxy/net/http/message.py --no-full-cov=mitmproxy/net/http/request.py --no-full-cov=mitmproxy/net/http/response.py --no-full-cov=mitmproxy/net/http/url.py \
--full-cov=mitmproxy/net/websockets/ \
--full-cov=mitmproxy/net/wsgi.py \
--full-cov=mitmproxy/proxy/__init__.py \
--full-cov=mitmproxy/proxy/modes/ --no-full-cov=mitmproxy/proxy/modes/socks_proxy.py \
--full-cov=mitmproxy/proxy/protocol/__init__.py \
--full-cov=mitmproxy/net/ --no-full-cov=mitmproxy/net/check.py --no-full-cov=mitmproxy/net/socks.py --no-full-cov=mitmproxy/net/tcp.py --no-full-cov=mitmproxy/net/http/cookies.py --no-full-cov=mitmproxy/net/http/encoding.py --no-full-cov=mitmproxy/net/http/message.py --no-full-cov=mitmproxy/net/http/request.py --no-full-cov=mitmproxy/net/http/response.py --no-full-cov=mitmproxy/net/http/url.py \
--full-cov=mitmproxy/proxy/ --no-full-cov=mitmproxy/proxy/protocol/ --no-full-cov=mitmproxy/proxy/modes/socks_proxy.py --no-full-cov=mitmproxy/proxy/config.py --no-full-cov=mitmproxy/proxy/root_context.py --no-full-cov=mitmproxy/proxy/server.py \
--full-cov=mitmproxy/script/ \
--full-cov=mitmproxy/test/ --no-full-cov=mitmproxy/test/tutils.py \
--full-cov=mitmproxy/test/ \
--full-cov=mitmproxy/types/ --no-full-cov=mitmproxy/types/basethread.py \
--full-cov=mitmproxy/utils/ \
--full-cov=mitmproxy/__init__.py \