diff options
Diffstat (limited to 'test')
38 files changed, 431 insertions, 295 deletions
diff --git a/test/conftest.py b/test/conftest.py index 4b05624b..83823a19 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -2,6 +2,7 @@ import os import pytest import OpenSSL import functools +from contextlib import contextmanager import mitmproxy.net.tcp @@ -29,35 +30,19 @@ skip_appveyor = pytest.mark.skipif( original_pytest_raises = pytest.raises +# TODO: remove this wrapper when pytest 3.1.0 is released +@contextmanager @functools.wraps(original_pytest_raises) def raises(exc, *args, **kwargs): - if isinstance(exc, str): - return RaisesContext(exc) - else: - return original_pytest_raises(exc, *args, **kwargs) + with original_pytest_raises(exc, *args, **kwargs) as exc_info: + yield + if 'match' in kwargs: + assert exc_info.match(kwargs['match']) 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): monkeypatch.setattr(mitmproxy.net.tcp, 'HAS_ALPN', False) @@ -118,15 +103,17 @@ def pytest_runtestloop(session): 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 not any(f.startswith(excluded_f) for excluded_f in excluded_files)] - for name in pytest.config.option.full_cov: + for name in coverage_values.keys(): files = [f for f in measured_files if f.startswith(os.path.normpath(name))] try: with open(os.devnull, 'w') as null: - coverage_values[name] = cov.report(files, ignore_errors=True, file=null) + overall = cov.report(files, ignore_errors=True, file=null) + singles = [(s, cov.report(s, ignore_errors=True, file=null)) for s in files] + coverage_values[name] = (overall, singles) except: pass - if any(v < 100 for v in coverage_values.values()): + if any(v < 100 for v, _ in coverage_values.values()): # make sure we get the EXIT_TESTSFAILED exit code session.testsfailed += 1 else: @@ -147,12 +134,15 @@ def pytest_terminal_summary(terminalreporter, exitstatus): msg = "FAIL: Full test coverage not reached!\n" terminalreporter.write(msg, **markup) - for name, value in coverage_values.items(): - if value < 100: + for name in sorted(coverage_values.keys()): + msg = 'Coverage for {}: {:.2f}%\n'.format(name, coverage_values[name][0]) + if coverage_values[name][0] < 100: markup = {'red': True, 'bold': True} + for s, v in sorted(coverage_values[name][1]): + if v < 100: + msg += ' {}: {:.2f}%\n'.format(s, v) else: markup = {'green': True} - msg = 'Coverage for {}: {:.2f}%\n'.format(name, value) terminalreporter.write(msg, **markup) else: markup = {'green': True} @@ -160,6 +150,7 @@ def pytest_terminal_summary(terminalreporter, exitstatus): msg += '{}\n\n'.format('\n'.join(pytest.config.option.full_cov)) terminalreporter.write(msg, **markup) - msg = 'Excluded files:\n' - msg += '{}\n'.format('\n'.join(pytest.config.option.no_full_cov)) + msg = '\nExcluded files:\n' + for s in sorted(pytest.config.option.no_full_cov): + msg += " {}\n".format(s) terminalreporter.write(msg) diff --git a/test/mitmproxy/addons/test_replace.py b/test/mitmproxy/addons/test_replace.py index 5583b2c0..126c6e3d 100644 --- a/test/mitmproxy/addons/test_replace.py +++ b/test/mitmproxy/addons/test_replace.py @@ -16,16 +16,16 @@ class TestReplace: assert x == ("foo", "bar", "vo/ing/") x = replace.parse_hook("/bar/voing") assert x == (".*", "bar", "voing") - with pytest.raises("invalid replacement"): + with pytest.raises(Exception, match="Invalid replacement"): replace.parse_hook("/") def test_configure(self): r = replace.Replace() with taddons.context() as tctx: tctx.configure(r, replacements=[("one", "two", "three")]) - with pytest.raises("invalid filter pattern"): + with pytest.raises(Exception, match="Invalid filter pattern"): tctx.configure(r, replacements=[("~b", "two", "three")]) - with pytest.raises("invalid regular expression"): + with pytest.raises(Exception, match="Invalid regular expression"): tctx.configure(r, replacements=[("foo", "+", "three")]) tctx.configure(r, replacements=["/a/b/c/"]) diff --git a/test/mitmproxy/addons/test_script.py b/test/mitmproxy/addons/test_script.py index ed7ec5f1..5f196ebf 100644 --- a/test/mitmproxy/addons/test_script.py +++ b/test/mitmproxy/addons/test_script.py @@ -44,14 +44,19 @@ def test_reloadhandler(): rh = script.ReloadHandler(Called()) assert not rh.filter(watchdog.events.DirCreatedEvent("path")) assert not rh.filter(watchdog.events.FileModifiedEvent("/foo/.bar")) - assert rh.filter(watchdog.events.FileModifiedEvent("/foo/bar")) + assert not rh.filter(watchdog.events.FileModifiedEvent("/foo/bar")) + assert rh.filter(watchdog.events.FileModifiedEvent("/foo/bar.py")) assert not rh.callback.called rh.on_modified(watchdog.events.FileModifiedEvent("/foo/bar")) + assert not rh.callback.called + rh.on_modified(watchdog.events.FileModifiedEvent("/foo/bar.py")) assert rh.callback.called rh.callback.called = False rh.on_created(watchdog.events.FileCreatedEvent("foo")) + assert not rh.callback.called + rh.on_created(watchdog.events.FileCreatedEvent("foo.py")) assert rh.callback.called @@ -64,11 +69,11 @@ class TestParseCommand: script.parse_command(" ") def test_no_script_file(self): - with pytest.raises("not found"): + with pytest.raises(Exception, match="not found"): script.parse_command("notfound") with tutils.tmpdir() as dir: - with pytest.raises("not a file"): + with pytest.raises(Exception, match="Not a file"): script.parse_command(dir) def test_parse_args(self): @@ -204,7 +209,7 @@ class TestScriptLoader: f = tflow.tflow(resp=True) with m.handlecontext(): - with pytest.raises("file not found"): + with pytest.raises(Exception, match="file not found"): sl.run_once("nonexistent", [f]) def test_simple(self): diff --git a/test/mitmproxy/addons/test_setheaders.py b/test/mitmproxy/addons/test_setheaders.py index 0091fc96..6355f2be 100644 --- a/test/mitmproxy/addons/test_setheaders.py +++ b/test/mitmproxy/addons/test_setheaders.py @@ -14,13 +14,13 @@ class TestSetHeaders: assert x == ("foo", "bar", "vo/ing/") x = setheaders.parse_setheader("/bar/voing") assert x == (".*", "bar", "voing") - with pytest.raises("invalid replacement"): + with pytest.raises(Exception, match="Invalid replacement"): setheaders.parse_setheader("/") def test_configure(self): sh = setheaders.SetHeaders() with taddons.context() as tctx: - with pytest.raises("invalid setheader filter pattern"): + with pytest.raises(Exception, match="Invalid setheader filter pattern"): tctx.configure(sh, setheaders = [("~b", "one", "two")]) tctx.configure(sh, setheaders = ["/foo/bar/voing"]) diff --git a/test/mitmproxy/addons/test_stickycookie.py b/test/mitmproxy/addons/test_stickycookie.py index 2297fe2c..9092e09b 100644 --- a/test/mitmproxy/addons/test_stickycookie.py +++ b/test/mitmproxy/addons/test_stickycookie.py @@ -16,7 +16,7 @@ class TestStickyCookie: def test_config(self): sc = stickycookie.StickyCookie() with taddons.context() as tctx: - with pytest.raises("invalid filter"): + with pytest.raises(Exception, match="invalid filter"): tctx.configure(sc, stickycookie="~b") tctx.configure(sc, stickycookie="foo") @@ -39,7 +39,6 @@ class TestStickyCookie: assert "cookie" not in f.request.headers f = f.copy() - f.reply.acked = False sc.request(f) assert f.request.headers["cookie"] == "foo=bar" diff --git a/test/mitmproxy/addons/test_streamfile.py b/test/mitmproxy/addons/test_streamfile.py index 79e66c1e..4922fc0b 100644 --- a/test/mitmproxy/addons/test_streamfile.py +++ b/test/mitmproxy/addons/test_streamfile.py @@ -18,7 +18,7 @@ def test_configure(): p = os.path.join(tdir, "foo") with pytest.raises(exceptions.OptionsError): tctx.configure(sa, streamfile=tdir) - with pytest.raises("invalid filter"): + with pytest.raises(Exception, match="Invalid filter"): tctx.configure(sa, streamfile=p, filtstr="~~") tctx.configure(sa, filtstr="foo") assert sa.filt diff --git a/test/mitmproxy/addons/test_view.py b/test/mitmproxy/addons/test_view.py index 7145bfc8..a063416f 100644 --- a/test/mitmproxy/addons/test_view.py +++ b/test/mitmproxy/addons/test_view.py @@ -396,11 +396,11 @@ def test_configure(): v = view.View() with taddons.context(options=Options()) as tctx: tctx.configure(v, filter="~q") - with pytest.raises("invalid interception filter"): + with pytest.raises(Exception, match="Invalid interception filter"): tctx.configure(v, filter="~~") tctx.configure(v, console_order="method") - with pytest.raises("unknown flow order"): + with pytest.raises(Exception, match="Unknown flow order"): tctx.configure(v, console_order="no") tctx.configure(v, console_order_reversed=True) diff --git a/test/mitmproxy/console/test_master.py b/test/mitmproxy/console/test_master.py index 8195b1d1..1c89467c 100644 --- a/test/mitmproxy/console/test_master.py +++ b/test/mitmproxy/console/test_master.py @@ -4,7 +4,7 @@ from mitmproxy.tools import console from mitmproxy import proxy from mitmproxy import options from mitmproxy.tools.console import common -from .. import mastertest +from .. import tservers def test_format_keyvals(): @@ -23,7 +23,7 @@ def test_options(): assert options.Options(replay_kill_extra=True) -class TestMaster(mastertest.MasterTest): +class TestMaster(tservers.MasterTest): def mkmaster(self, **opts): if "verbosity" not in opts: opts["verbosity"] = 1 diff --git a/test/mitmproxy/data/addonscripts/recorder.py b/test/mitmproxy/data/addonscripts/recorder.py index 5be88e5c..6b9b6ea8 100644 --- a/test/mitmproxy/data/addonscripts/recorder.py +++ b/test/mitmproxy/data/addonscripts/recorder.py @@ -1,5 +1,5 @@ from mitmproxy import controller -from mitmproxy import events +from mitmproxy import eventsequence from mitmproxy import ctx import sys @@ -11,7 +11,7 @@ class CallLogger: self.name = name def __getattr__(self, attr): - if attr in events.Events: + if attr in eventsequence.Events: def prox(*args, **kwargs): lg = (self.name, attr, args, kwargs) if attr != "log": diff --git a/test/mitmproxy/mastertest.py b/test/mitmproxy/mastertest.py deleted file mode 100644 index e3bb3069..00000000 --- a/test/mitmproxy/mastertest.py +++ /dev/null @@ -1,56 +0,0 @@ -import contextlib - -from mitmproxy.test import tflow - -import mitmproxy.test.tutils - -from mitmproxy import master -from mitmproxy import io -from mitmproxy import proxy -from mitmproxy import http -from mitmproxy import options - - -class MasterTest: - - def cycle(self, master, content): - f = tflow.tflow(req=mitmproxy.test.tutils.treq(content=content)) - master.clientconnect(f.client_conn) - master.serverconnect(f.server_conn) - master.request(f) - if not f.error: - f.response = http.HTTPResponse.wrap( - mitmproxy.test.tutils.tresp(content=content) - ) - master.response(f) - master.clientdisconnect(f) - return f - - def dummy_cycle(self, master, n, content): - for i in range(n): - self.cycle(master, content) - master.shutdown() - - def flowfile(self, path): - f = open(path, "wb") - fw = io.FlowWriter(f) - t = tflow.tflow(resp=True) - fw.add(t) - f.close() - - -class RecordingMaster(master.Master): - def __init__(self, *args, **kwargs): - master.Master.__init__(self, *args, **kwargs) - self.event_log = [] - - def add_log(self, e, level): - self.event_log.append((level, e)) - - -@contextlib.contextmanager -def mockctx(): - o = options.Options(refresh_server_playback = True, keepserving=False) - m = RecordingMaster(o, proxy.DummyServer(o)) - with m.handlecontext(): - yield diff --git a/test/mitmproxy/net/http/http1/test_read.py b/test/mitmproxy/net/http/http1/test_read.py index 01d03e7c..642b91c0 100644 --- a/test/mitmproxy/net/http/http1/test_read.py +++ b/test/mitmproxy/net/http/http1/test_read.py @@ -356,11 +356,11 @@ def test_read_chunked(): assert b"".join(_read_chunked(BytesIO(data))) == b"ab" data = b"\r\n" - with pytest.raises("closed prematurely"): + with pytest.raises(Exception, match="closed prematurely"): b"".join(_read_chunked(BytesIO(data))) data = b"1\r\nfoo" - with pytest.raises("malformed chunked body"): + with pytest.raises(Exception, match="Malformed chunked body"): b"".join(_read_chunked(BytesIO(data))) data = b"foo\r\nfoo" @@ -368,5 +368,5 @@ def test_read_chunked(): b"".join(_read_chunked(BytesIO(data))) data = b"5\r\naaaaa\r\n0\r\n\r\n" - with pytest.raises("too large"): + with pytest.raises(Exception, match="too large"): b"".join(_read_chunked(BytesIO(data), limit=2)) diff --git a/test/mitmproxy/net/http/test_request.py b/test/mitmproxy/net/http/test_request.py index 5c588c47..6fe57010 100644 --- a/test/mitmproxy/net/http/test_request.py +++ b/test/mitmproxy/net/http/test_request.py @@ -1,5 +1,4 @@ -# -*- coding: utf-8 -*- - +from unittest import mock import pytest from mitmproxy.net.http import Headers @@ -9,8 +8,19 @@ from .test_message import _test_decoded_attr, _test_passthrough_attr class TestRequestData: def test_init(self): + with pytest.raises(UnicodeEncodeError): + treq(method="fööbär") + with pytest.raises(UnicodeEncodeError): + treq(scheme="fööbär") + assert treq(host="fööbär").host == "fööbär" + with pytest.raises(UnicodeEncodeError): + treq(path="/fööbär") + with pytest.raises(UnicodeEncodeError): + treq(http_version="föö/bä.r") with pytest.raises(ValueError): treq(headers="foobar") + with pytest.raises(ValueError): + treq(content="foobar") assert isinstance(treq(headers=()).headers, Headers) @@ -25,15 +35,19 @@ class TestRequestCore: request.host = None assert repr(request) == "Request(GET /path)" - def replace(self): + def test_replace(self): r = treq() r.path = b"foobarfoo" r.replace(b"foo", "bar") - assert r.path == b"barbarbar" + assert r.path == "barbarbar" r.path = b"foobarfoo" r.replace(b"foo", "bar", count=1) - assert r.path == b"barbarfoo" + assert r.path == "barbarfoo" + + r.path = "foobarfoo" + r.replace("foo", "bar", count=1) + assert r.path == "barbarfoo" def test_first_line_format(self): _test_passthrough_attr(treq(), "first_line_format") @@ -43,6 +57,7 @@ class TestRequestCore: def test_scheme(self): _test_decoded_attr(treq(), "scheme") + assert treq(scheme=None).scheme is None def test_port(self): _test_passthrough_attr(treq(), "port") @@ -172,6 +187,9 @@ class TestRequestUtils: request.query["foo"] = "bar" assert request.query["foo"] == "bar" assert request.path == "/path?foo=bar" + request.query = [('foo', 'bar')] + assert request.query["foo"] == "bar" + assert request.path == "/path?foo=bar" def test_get_cookies_none(self): request = treq() @@ -206,6 +224,9 @@ class TestRequestUtils: result = request.cookies result["cookiename"] = "foo" assert request.cookies["cookiename"] == "foo" + request.cookies = [["one", "uno"], ["two", "due"]] + assert request.cookies["one"] == "uno" + assert request.cookies["two"] == "due" def test_get_path_components(self): request = treq(path=b"/foo/bar") @@ -258,6 +279,8 @@ class TestRequestUtils: request.headers["Content-Type"] = "application/x-www-form-urlencoded" assert list(request.urlencoded_form.items()) == [("foobar", "baz")] + request.raw_content = b"\xFF" + assert len(request.urlencoded_form) == 0 def test_set_urlencoded_form(self): request = treq() @@ -271,3 +294,12 @@ class TestRequestUtils: request.headers["Content-Type"] = "multipart/form-data" assert list(request.multipart_form.items()) == [] + + with mock.patch('mitmproxy.net.http.multipart.decode') as m: + m.side_effect = ValueError + assert list(request.multipart_form.items()) == [] + + def test_set_multipart_form(self): + request = treq(content=b"foobar") + with pytest.raises(NotImplementedError): + request.multipart_form = "foobar" diff --git a/test/mitmproxy/net/http/test_response.py b/test/mitmproxy/net/http/test_response.py index 9c528fd0..fa1770fe 100644 --- a/test/mitmproxy/net/http/test_response.py +++ b/test/mitmproxy/net/http/test_response.py @@ -1,6 +1,7 @@ import email import time import pytest +from unittest import mock from mitmproxy.net.http import Headers from mitmproxy.net.http import Response @@ -13,6 +14,12 @@ class TestResponseData: def test_init(self): with pytest.raises(ValueError): tresp(headers="foobar") + with pytest.raises(UnicodeEncodeError): + tresp(http_version="föö/bä.r") + with pytest.raises(UnicodeEncodeError): + tresp(reason="fööbär") + with pytest.raises(ValueError): + tresp(content="foobar") assert isinstance(tresp(headers=()).headers, Headers) @@ -133,9 +140,10 @@ class TestResponseUtils: def test_set_cookies(self): resp = tresp() resp.cookies["foo"] = ("bar", {}) - assert len(resp.cookies) == 1 assert resp.cookies["foo"] == ("bar", CookieAttrs()) + resp.cookies = [["one", ("uno", CookieAttrs())], ["two", ("due", CookieAttrs())]] + assert list(resp.cookies.keys()) == ["one", "two"] def test_refresh(self): r = tresp() @@ -156,3 +164,7 @@ class TestResponseUtils: r.refresh() # Cookie refreshing is tested in test_cookies, we just make sure that it's triggered here. assert cookie != r.headers["set-cookie"] + + with mock.patch('mitmproxy.net.http.cookies.refresh_set_cookie_header') as m: + m.side_effect = ValueError + r.refresh(n) diff --git a/test/mitmproxy/net/test_socks.py b/test/mitmproxy/net/test_socks.py index c1bd0603..e00dd410 100644 --- a/test/mitmproxy/net/test_socks.py +++ b/test/mitmproxy/net/test_socks.py @@ -181,9 +181,8 @@ def test_message_ipv6(): def test_message_invalid_host(): raw = tutils.treader(b"\xEE\x01\x00\x03\x0bexample@com\xDE\xAD\xBE\xEF") - with pytest.raises(socks.SocksError) as exc_info: + with pytest.raises(socks.SocksError, match="Invalid hostname: b'example@com'"): socks.Message.from_file(raw) - assert exc_info.match("Invalid hostname: b'example@com'") def test_message_invalid_rsv(): diff --git a/test/mitmproxy/net/test_tcp.py b/test/mitmproxy/net/test_tcp.py index eb09f328..ff6362c8 100644 --- a/test/mitmproxy/net/test_tcp.py +++ b/test/mitmproxy/net/test_tcp.py @@ -430,7 +430,7 @@ class TestServerCipherListError(tservers.ServerTestBase): def test_echo(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): - with pytest.raises("handshake error"): + with pytest.raises(Exception, match="handshake error"): c.convert_to_ssl(sni="foo.com") @@ -443,7 +443,7 @@ class TestClientCipherListError(tservers.ServerTestBase): def test_echo(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): - with pytest.raises("cipher specification"): + with pytest.raises(Exception, match="cipher specification"): c.convert_to_ssl(sni="foo.com", cipher_list="bogus") diff --git a/test/mitmproxy/net/websockets/test_frame.py b/test/mitmproxy/net/websockets/test_frame.py index 183c7caa..2a5bd556 100644 --- a/test/mitmproxy/net/websockets/test_frame.py +++ b/test/mitmproxy/net/websockets/test_frame.py @@ -108,9 +108,9 @@ class TestFrameHeader: assert not f2.mask def test_violations(self): - with pytest.raises("opcode"): + with pytest.raises(Exception, match="opcode"): websockets.FrameHeader(opcode=17) - with pytest.raises("masking key"): + with pytest.raises(Exception, match="Masking key"): websockets.FrameHeader(masking_key=b"x") def test_automask(self): diff --git a/test/mitmproxy/protocol/test_http1.py b/test/mitmproxy/protocol/test_http1.py index cd937ada..44a9effa 100644 --- a/test/mitmproxy/protocol/test_http1.py +++ b/test/mitmproxy/protocol/test_http1.py @@ -30,6 +30,16 @@ class TestInvalidRequests(tservers.HTTPProxyTest): assert b"Invalid HTTP request form" in r.content +class TestProxyMisconfiguration(tservers.TransparentProxyTest): + + def test_absolute_request(self): + p = self.pathoc() + with p.connect(): + r = p.request("get:'http://localhost:%d/p/200'" % self.server.port) + assert r.status_code == 400 + assert b"misconfiguration" in r.content + + class TestExpectHeader(tservers.HTTPProxyTest): def test_simple(self): diff --git a/test/mitmproxy/protocol/test_websocket.py b/test/mitmproxy/protocol/test_websocket.py index e42250e0..73ee8b35 100644 --- a/test/mitmproxy/protocol/test_websocket.py +++ b/test/mitmproxy/protocol/test_websocket.py @@ -179,16 +179,15 @@ class TestSimple(_WebSocketTest): assert isinstance(self.master.state.flows[1], WebSocketFlow) assert len(self.master.state.flows[1].messages) == 5 assert self.master.state.flows[1].messages[0].content == b'server-foobar' - assert self.master.state.flows[1].messages[0].type == 'text' + assert self.master.state.flows[1].messages[0].type == websockets.OPCODE.TEXT assert self.master.state.flows[1].messages[1].content == b'client-foobar' - assert self.master.state.flows[1].messages[1].type == 'text' + assert self.master.state.flows[1].messages[1].type == websockets.OPCODE.TEXT assert self.master.state.flows[1].messages[2].content == b'client-foobar' - assert self.master.state.flows[1].messages[2].type == 'text' + assert self.master.state.flows[1].messages[2].type == websockets.OPCODE.TEXT assert self.master.state.flows[1].messages[3].content == b'\xde\xad\xbe\xef' - assert self.master.state.flows[1].messages[3].type == 'binary' + assert self.master.state.flows[1].messages[3].type == websockets.OPCODE.BINARY assert self.master.state.flows[1].messages[4].content == b'\xde\xad\xbe\xef' - assert self.master.state.flows[1].messages[4].type == 'binary' - assert [m.info for m in self.master.state.flows[1].messages] + assert self.master.state.flows[1].messages[4].type == websockets.OPCODE.BINARY class TestSimpleTLS(_WebSocketTest): diff --git a/test/mitmproxy/script/test_concurrent.py b/test/mitmproxy/script/test_concurrent.py index 90cdc3d8..fb932d9a 100644 --- a/test/mitmproxy/script/test_concurrent.py +++ b/test/mitmproxy/script/test_concurrent.py @@ -7,7 +7,7 @@ from mitmproxy.addons import script import time -from test.mitmproxy import mastertest +from .. import tservers class Thing: @@ -16,7 +16,7 @@ class Thing: self.live = True -class TestConcurrent(mastertest.MasterTest): +class TestConcurrent(tservers.MasterTest): def test_concurrent(self): with taddons.context() as tctx: sc = script.Script( diff --git a/test/mitmproxy/test_eventsequence.py b/test/mitmproxy/test_eventsequence.py index 262df4b0..fe0f92b3 100644 --- a/test/mitmproxy/test_eventsequence.py +++ b/test/mitmproxy/test_eventsequence.py @@ -1,81 +1,57 @@ -from mitmproxy import events -import contextlib -from . import tservers - - -class Eventer: - def __init__(self, **handlers): - self.failure = None - self.called = [] - self.handlers = handlers - for i in events.Events - {"tick"}: - def mkprox(): - evt = i - - def prox(*args, **kwargs): - self.called.append(evt) - if evt in self.handlers: - try: - handlers[evt](*args, **kwargs) - except AssertionError as e: - self.failure = e - return prox - setattr(self, i, mkprox()) - - def fail(self): - pass - - -class SequenceTester: - @contextlib.contextmanager - def addon(self, addon): - self.master.addons.add(addon) - yield - self.master.addons.remove(addon) - if addon.failure: - raise addon.failure - - -class TestBasic(tservers.HTTPProxyTest, SequenceTester): - ssl = True - - def test_requestheaders(self): - - def hdrs(f): - assert f.request.headers - assert not f.request.content - - def req(f): - assert f.request.headers - assert f.request.content - - with self.addon(Eventer(requestheaders=hdrs, request=req)): - p = self.pathoc() - with p.connect(): - assert p.request("get:'/p/200':b@10").status_code == 200 - - def test_100_continue_fail(self): - e = Eventer() - with self.addon(e): - p = self.pathoc() - with p.connect(): - p.request( - """ - get:'/p/200' - h'expect'='100-continue' - h'content-length'='1000' - da - """ - ) - assert "requestheaders" in e.called - assert "responseheaders" not in e.called - - def test_connect(self): - e = Eventer() - with self.addon(e): - p = self.pathoc() - with p.connect(): - p.request("get:'/p/200:b@1'") - assert "http_connect" in e.called - assert e.called.count("requestheaders") == 1 - assert e.called.count("request") == 1 +import pytest + +from mitmproxy import eventsequence +from mitmproxy.test import tflow + + +@pytest.mark.parametrize("resp, err", [ + (False, False), + (True, False), + (False, True), + (True, True), +]) +def test_http_flow(resp, err): + f = tflow.tflow(resp=resp, err=err) + i = eventsequence.iterate(f) + assert next(i) == ("requestheaders", f) + assert next(i) == ("request", f) + if resp: + assert next(i) == ("responseheaders", f) + assert next(i) == ("response", f) + if err: + assert next(i) == ("error", f) + + +@pytest.mark.parametrize("err", [False, True]) +def test_websocket_flow(err): + f = tflow.twebsocketflow(err=err) + i = eventsequence.iterate(f) + assert next(i) == ("websocket_start", f) + assert len(f.messages) == 0 + assert next(i) == ("websocket_message", f) + assert len(f.messages) == 1 + assert next(i) == ("websocket_message", f) + assert len(f.messages) == 2 + if err: + assert next(i) == ("websocket_error", f) + assert next(i) == ("websocket_end", f) + + +@pytest.mark.parametrize("err", [False, True]) +def test_tcp_flow(err): + f = tflow.ttcpflow(err=err) + i = eventsequence.iterate(f) + assert next(i) == ("tcp_start", f) + assert len(f.messages) == 0 + assert next(i) == ("tcp_message", f) + assert len(f.messages) == 1 + assert next(i) == ("tcp_message", f) + assert len(f.messages) == 2 + if err: + assert next(i) == ("tcp_error", f) + assert next(i) == ("tcp_end", f) + + +def test_invalid(): + with pytest.raises(TypeError): + next(eventsequence.iterate(42)) diff --git a/test/mitmproxy/test_examples.py b/test/mitmproxy/test_examples.py index dae52e23..f3603fca 100644 --- a/test/mitmproxy/test_examples.py +++ b/test/mitmproxy/test_examples.py @@ -14,7 +14,7 @@ from mitmproxy.test import tutils from mitmproxy.net.http import Headers from mitmproxy.net.http import cookies -from . import mastertest +from . import tservers example_dir = tutils.test_data.push("../examples") @@ -38,7 +38,7 @@ def tscript(cmd, args=""): return m, sc -class TestScripts(mastertest.MasterTest): +class TestScripts(tservers.MasterTest): def test_add_header(self): m, _ = tscript("simple/add_header.py") f = tflow.tflow(resp=tutils.tresp()) diff --git a/test/mitmproxy/test_flow.py b/test/mitmproxy/test_flow.py index f546d61b..65e6845f 100644 --- a/test/mitmproxy/test_flow.py +++ b/test/mitmproxy/test_flow.py @@ -10,6 +10,7 @@ from mitmproxy.exceptions import FlowReadException, Kill from mitmproxy import flow from mitmproxy import http from mitmproxy import connections +from mitmproxy import tcp from mitmproxy.proxy import ProxyConfig from mitmproxy.proxy.server import DummyServer from mitmproxy import master @@ -156,8 +157,99 @@ class TestHTTPFlow: assert f.response.raw_content == b"abarb" +class TestWebSocketFlow: + + def test_copy(self): + f = tflow.twebsocketflow() + f.get_state() + f2 = f.copy() + a = f.get_state() + b = f2.get_state() + del a["id"] + del b["id"] + del a["handshake_flow"]["id"] + del b["handshake_flow"]["id"] + assert a == b + assert not f == f2 + assert f is not f2 + + assert f.client_key == f2.client_key + assert f.client_protocol == f2.client_protocol + assert f.client_extensions == f2.client_extensions + assert f.server_accept == f2.server_accept + assert f.server_protocol == f2.server_protocol + assert f.server_extensions == f2.server_extensions + assert f.messages is not f2.messages + assert f.handshake_flow is not f2.handshake_flow + + for m in f.messages: + m2 = m.copy() + m2.set_state(m2.get_state()) + assert m is not m2 + assert m.get_state() == m2.get_state() + + f = tflow.twebsocketflow(err=True) + f2 = f.copy() + assert f is not f2 + assert f.handshake_flow is not f2.handshake_flow + assert f.error.get_state() == f2.error.get_state() + assert f.error is not f2.error + + def test_match(self): + f = tflow.twebsocketflow() + assert not flowfilter.match("~b nonexistent", f) + assert flowfilter.match(None, f) + assert not flowfilter.match("~b nonexistent", f) + + f = tflow.twebsocketflow(err=True) + assert flowfilter.match("~e", f) + + with pytest.raises(ValueError): + flowfilter.match("~", f) + + def test_repr(self): + f = tflow.twebsocketflow() + assert 'WebSocketFlow' in repr(f) + assert 'binary message: ' in repr(f.messages[0]) + assert 'text message: ' in repr(f.messages[1]) + + class TestTCPFlow: + def test_copy(self): + f = tflow.ttcpflow() + f.get_state() + f2 = f.copy() + a = f.get_state() + b = f2.get_state() + del a["id"] + del b["id"] + assert a == b + assert not f == f2 + assert f is not f2 + + assert f.messages is not f2.messages + + for m in f.messages: + assert m.get_state() + m2 = m.copy() + assert not m == m2 + assert m is not m2 + + a = m.get_state() + b = m2.get_state() + assert a == b + + m = tcp.TCPMessage(False, 'foo') + m.set_state(f.messages[0].get_state()) + assert m.timestamp == f.messages[0].timestamp + + f = tflow.ttcpflow(err=True) + f2 = f.copy() + assert f is not f2 + assert f.error.get_state() == f2.error.get_state() + assert f.error is not f2.error + def test_match(self): f = tflow.ttcpflow() assert not flowfilter.match("~b nonexistent", f) @@ -170,6 +262,11 @@ class TestTCPFlow: with pytest.raises(ValueError): flowfilter.match("~", f) + def test_repr(self): + f = tflow.ttcpflow() + assert 'TCPFlow' in repr(f) + assert '-> ' in repr(f.messages[0]) + class TestSerialize: @@ -251,9 +348,8 @@ class TestSerialize: sio.write(b"bogus") sio.seek(0) r = mitmproxy.io.FlowReader(sio) - with pytest.raises(FlowReadException) as exc_info: + with pytest.raises(FlowReadException, match='Invalid data format'): list(r.stream()) - assert exc_info.match('Invalid data format') sio = io.BytesIO() f = tflow.tdummyflow() @@ -261,9 +357,8 @@ class TestSerialize: w.add(f) sio.seek(0) r = mitmproxy.io.FlowReader(sio) - with pytest.raises(FlowReadException) as exc_info: + with pytest.raises(FlowReadException, match='Unknown flow type'): list(r.stream()) - assert exc_info.match('Unknown flow type') f = FlowReadException("foo") assert str(f) == "foo" @@ -277,7 +372,7 @@ class TestSerialize: sio.seek(0) r = mitmproxy.io.FlowReader(sio) - with pytest.raises("version"): + with pytest.raises(Exception, match="version"): list(r.stream()) @@ -287,15 +382,15 @@ class TestFlowMaster: fm = master.Master(None, DummyServer()) f = tflow.tflow(resp=True) f.request.content = None - with pytest.raises("missing"): + with pytest.raises(Exception, match="missing"): fm.replay_request(f) f.intercepted = True - with pytest.raises("intercepted"): + with pytest.raises(Exception, match="intercepted"): fm.replay_request(f) f.live = True - with pytest.raises("live"): + with pytest.raises(Exception, match="live"): fm.replay_request(f) def test_create_flow(self): @@ -310,11 +405,11 @@ class TestFlowMaster: fm.clientconnect(f.client_conn) f.request = http.HTTPRequest.wrap(mitmproxy.test.tutils.treq()) fm.request(f) - assert s.flow_count() == 1 + assert len(s.flows) == 1 f.response = http.HTTPResponse.wrap(mitmproxy.test.tutils.tresp()) fm.response(f) - assert s.flow_count() == 1 + assert len(s.flows) == 1 fm.clientdisconnect(f.client_conn) diff --git a/test/mitmproxy/test_optmanager.py b/test/mitmproxy/test_optmanager.py index f177df7b..65691fdf 100644 --- a/test/mitmproxy/test_optmanager.py +++ b/test/mitmproxy/test_optmanager.py @@ -71,9 +71,9 @@ def test_options(): with pytest.raises(TypeError): TO(nonexistent = "value") - with pytest.raises("no such option"): + with pytest.raises(Exception, match="No such option"): o.nonexistent = "value" - with pytest.raises("no such option"): + with pytest.raises(Exception, match="No such option"): o.update(nonexistent = "value") rec = [] @@ -97,7 +97,7 @@ def test_setter(): f = o.setter("two") f("xxx") assert o.two == "xxx" - with pytest.raises("no such option"): + with pytest.raises(Exception, match="No such option"): o.setter("nonexistent") @@ -108,7 +108,7 @@ def test_toggler(): assert o.two is False f() assert o.two is True - with pytest.raises("no such option"): + with pytest.raises(Exception, match="No such option"): o.toggler("nonexistent") @@ -193,11 +193,11 @@ def test_serialize(): assert o2 == o t = "invalid: foo\ninvalid" - with pytest.raises("config error"): + with pytest.raises(Exception, match="Config error"): o2.load(t) t = "invalid" - with pytest.raises("config error"): + with pytest.raises(Exception, match="Config error"): o2.load(t) t = "" diff --git a/test/mitmproxy/test_platform_pf.py b/test/mitmproxy/test_platform_pf.py index ebb011fe..f644bcc5 100644 --- a/test/mitmproxy/test_platform_pf.py +++ b/test/mitmproxy/test_platform_pf.py @@ -14,7 +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) - with pytest.raises("Could not resolve original destination"): + with pytest.raises(Exception, match="Could not resolve original destination"): pf.lookup("192.168.1.112", 40000, d) - with pytest.raises("Could not resolve original destination"): + with pytest.raises(Exception, match="Could not resolve original destination"): pf.lookup("192.168.1.111", 40001, d) diff --git a/test/mitmproxy/test_proxy.py b/test/mitmproxy/test_proxy.py index 9d793572..a14c851e 100644 --- a/test/mitmproxy/test_proxy.py +++ b/test/mitmproxy/test_proxy.py @@ -96,27 +96,27 @@ class TestProcessProxyOptions: @mock.patch("mitmproxy.platform.original_addr", None) def test_no_transparent(self): - with pytest.raises("transparent mode not supported"): + with pytest.raises(Exception, match="Transparent mode not supported"): self.p("-T") @mock.patch("mitmproxy.platform.original_addr") def test_modes(self, _): self.assert_noerr("-R", "http://localhost") - with pytest.raises("expected one argument"): + with pytest.raises(Exception, match="expected one argument"): self.p("-R") - with pytest.raises("Invalid server specification"): + with pytest.raises(Exception, match="Invalid server specification"): self.p("-R", "reverse") self.assert_noerr("-T") self.assert_noerr("-U", "http://localhost") - with pytest.raises("Invalid server specification"): + with pytest.raises(Exception, match="Invalid server specification"): self.p("-U", "upstream") self.assert_noerr("--upstream-auth", "test:test") - with pytest.raises("expected one argument"): + with pytest.raises(Exception, match="expected one argument"): self.p("--upstream-auth") - with pytest.raises("mutually exclusive"): + with pytest.raises(Exception, match="mutually exclusive"): self.p("-R", "http://localhost", "-T") def test_client_certs(self): @@ -125,14 +125,14 @@ class TestProcessProxyOptions: self.assert_noerr( "--client-certs", os.path.join(tutils.test_data.path("mitmproxy/data/clientcert"), "client.pem")) - with pytest.raises("path does not exist"): + with pytest.raises(Exception, match="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")) - with pytest.raises("does not exist"): + with pytest.raises(Exception, match="does not exist"): self.p("--cert", "nonexistent") def test_insecure(self): @@ -156,12 +156,12 @@ class TestProxyServer: def test_err(self): # 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"): + with pytest.raises(Exception, match="Error starting proxy server"): ProxyServer(conf) def test_err_2(self): conf = ProxyConfig(options.Options(listen_host="invalidhost")) - with pytest.raises("error starting proxy server"): + with pytest.raises(Exception, match="Error starting proxy server"): ProxyServer(conf) diff --git a/test/mitmproxy/test_proxy_config.py b/test/mitmproxy/test_proxy_config.py index 27563e3a..4272d952 100644 --- a/test/mitmproxy/test_proxy_config.py +++ b/test/mitmproxy/test_proxy_config.py @@ -3,7 +3,7 @@ from mitmproxy.proxy import config def test_parse_server_spec(): - with pytest.raises("Invalid server specification"): + with pytest.raises(Exception, match="Invalid server specification"): config.parse_server_spec("") assert config.parse_server_spec("http://foo.com:88") == ( "http", ("foo.com", 88) @@ -14,7 +14,7 @@ def test_parse_server_spec(): assert config.parse_server_spec("https://foo.com") == ( "https", ("foo.com", 443) ) - with pytest.raises("Invalid server specification"): + with pytest.raises(Exception, match="Invalid server specification"): config.parse_server_spec("foo.com") - with pytest.raises("Invalid server specification"): + with pytest.raises(Exception, match="Invalid server specification"): config.parse_server_spec("http://") diff --git a/test/mitmproxy/test_server.py b/test/mitmproxy/test_server.py index 6e7ca275..9cd47cea 100644 --- a/test/mitmproxy/test_server.py +++ b/test/mitmproxy/test_server.py @@ -984,16 +984,39 @@ class TestUpstreamProxySSL( assert req.status_code == 418 # CONNECT from pathoc to chain[0], - assert self.proxy.tmaster.state.flow_count() == 1 + assert len(self.proxy.tmaster.state.flows) == 1 assert self.proxy.tmaster.state.flows[0].server_conn.via # request from pathoc to chain[0] # CONNECT from proxy to chain[1], - assert self.chain[0].tmaster.state.flow_count() == 1 + assert len(self.chain[0].tmaster.state.flows) == 1 assert self.chain[0].tmaster.state.flows[0].server_conn.via # request from proxy to chain[1] # request from chain[0] (regular proxy doesn't store CONNECTs) assert not self.chain[1].tmaster.state.flows[0].server_conn.via - assert self.chain[1].tmaster.state.flow_count() == 1 + assert len(self.chain[1].tmaster.state.flows) == 1 + + def test_change_upstream_proxy_connect(self): + # skip chain[0]. + self.proxy.tmaster.addons.add( + UpstreamProxyChanger( + ("127.0.0.1", self.chain[1].port) + ) + ) + p = self.pathoc() + with p.connect(): + req = p.request("get:'/p/418'") + + assert req.status_code == 418 + assert len(self.chain[0].tmaster.state.flows) == 0 + assert len(self.chain[1].tmaster.state.flows) == 1 + + +class UpstreamProxyChanger: + def __init__(self, addr): + self.address = addr + + def request(self, f): + f.live.change_upstream_proxy_server(self.address) class RequestKiller: @@ -1027,17 +1050,17 @@ class TestProxyChainingSSLReconnect(tservers.HTTPUpstreamProxyTest): assert req.status_code == 418 # First request goes through all three proxies exactly once - assert self.proxy.tmaster.state.flow_count() == 1 - assert self.chain[0].tmaster.state.flow_count() == 1 - assert self.chain[1].tmaster.state.flow_count() == 1 + assert len(self.proxy.tmaster.state.flows) == 1 + assert len(self.chain[0].tmaster.state.flows) == 1 + assert len(self.chain[1].tmaster.state.flows) == 1 req = p.request("get:'/p/418:b\"content2\"'") assert req.status_code == 502 - assert self.proxy.tmaster.state.flow_count() == 2 - assert self.chain[0].tmaster.state.flow_count() == 2 + assert len(self.proxy.tmaster.state.flows) == 2 + assert len(self.chain[0].tmaster.state.flows) == 2 # Upstream sees two requests due to reconnection attempt - assert self.chain[1].tmaster.state.flow_count() == 3 + assert len(self.chain[1].tmaster.state.flows) == 3 assert not self.chain[1].tmaster.state.flows[-1].response assert not self.chain[1].tmaster.state.flows[-2].response diff --git a/test/mitmproxy/test_stateobject.py b/test/mitmproxy/test_stateobject.py index b9ffe7ae..7b8e30d0 100644 --- a/test/mitmproxy/test_stateobject.py +++ b/test/mitmproxy/test_stateobject.py @@ -1,4 +1,5 @@ from typing import List +import pytest from mitmproxy.stateobject import StateObject @@ -17,6 +18,9 @@ class Child(StateObject): obj.set_state(state) return obj + def __eq__(self, other): + return isinstance(other, Child) and self.x == other.x + class Container(StateObject): def __init__(self): @@ -60,4 +64,18 @@ def test_container_list(): "child": None, "children": [{"x": 42}, {"x": 44}] } - assert len(a.copy().children) == 2 + copy = a.copy() + assert len(copy.children) == 2 + assert copy.children is not a.children + assert copy.children[0] is not a.children[0] + + +def test_too_much_state(): + a = Container() + a.child = Child(42) + s = a.get_state() + s['foo'] = 'bar' + b = Container() + + with pytest.raises(RuntimeWarning): + b.set_state(s) diff --git a/test/mitmproxy/test_tools_dump.py b/test/mitmproxy/test_tools_dump.py index a471354c..f8a88871 100644 --- a/test/mitmproxy/test_tools_dump.py +++ b/test/mitmproxy/test_tools_dump.py @@ -8,10 +8,10 @@ from mitmproxy import controller from mitmproxy.tools import dump from mitmproxy.test import tutils -from . import mastertest +from . import tservers -class TestDumpMaster(mastertest.MasterTest): +class TestDumpMaster(tservers.MasterTest): def mkmaster(self, flt, **options): o = dump.Options(filtstr=flt, verbosity=-1, flow_detail=0, **options) m = dump.DumpMaster(o, proxy.DummyServer(), with_termlog=False, with_dumper=False) diff --git a/test/mitmproxy/test_version.py b/test/mitmproxy/test_version.py new file mode 100644 index 00000000..f87b0851 --- /dev/null +++ b/test/mitmproxy/test_version.py @@ -0,0 +1,10 @@ +import runpy + +from mitmproxy import version + + +def test_version(capsys): + runpy.run_module('mitmproxy.version', run_name='__main__') + stdout, stderr = capsys.readouterr() + assert len(stdout) > 0 + assert stdout.strip() == version.VERSION diff --git a/test/mitmproxy/test_web_master.py b/test/mitmproxy/test_web_master.py index 3591284d..d4190ffb 100644 --- a/test/mitmproxy/test_web_master.py +++ b/test/mitmproxy/test_web_master.py @@ -3,10 +3,10 @@ from mitmproxy import proxy from mitmproxy import options from mitmproxy.proxy.config import ProxyConfig -from . import mastertest +from . import tservers -class TestWebMaster(mastertest.MasterTest): +class TestWebMaster(tservers.MasterTest): def mkmaster(self, **opts): o = options.Options(**opts) c = ProxyConfig(o) diff --git a/test/mitmproxy/tservers.py b/test/mitmproxy/tservers.py index 170a4917..298fddcb 100644 --- a/test/mitmproxy/tservers.py +++ b/test/mitmproxy/tservers.py @@ -7,11 +7,43 @@ import mitmproxy.platform from mitmproxy.proxy.config import ProxyConfig from mitmproxy.proxy.server import ProxyServer from mitmproxy import master -import pathod.test -import pathod.pathoc from mitmproxy import controller from mitmproxy import options from mitmproxy import exceptions +from mitmproxy import io +from mitmproxy import http +import pathod.test +import pathod.pathoc + +from mitmproxy.test import tflow +from mitmproxy.test import tutils + + +class MasterTest: + + def cycle(self, master, content): + f = tflow.tflow(req=tutils.treq(content=content)) + master.clientconnect(f.client_conn) + master.serverconnect(f.server_conn) + master.request(f) + if not f.error: + f.response = http.HTTPResponse.wrap( + tutils.tresp(content=content) + ) + master.response(f) + master.clientdisconnect(f) + return f + + def dummy_cycle(self, master, n, content): + for i in range(n): + self.cycle(master, content) + master.shutdown() + + def flowfile(self, path): + with open(path, "wb") as f: + fw = io.FlowWriter(f) + t = tflow.tflow(resp=True) + fw.add(t) class TestState: @@ -35,10 +67,6 @@ class TestState: # if f not in self.flows: # self.flows.append(f) - # FIXME: compat with old state - remove in favor of len(state.flows) - def flow_count(self): - return len(self.flows) - class TestMaster(master.Master): diff --git a/test/pathod/test_language_base.py b/test/pathod/test_language_base.py index 190c39b3..85e9e53b 100644 --- a/test/pathod/test_language_base.py +++ b/test/pathod/test_language_base.py @@ -149,11 +149,11 @@ class TestTokValueFile: v = base.TokValue.parseString("<path2")[0] with pytest.raises(exceptions.FileAccessDenied): v.get_generator(language.Settings(staticdir=t)) - with pytest.raises("access disabled"): + with pytest.raises(Exception, match="access disabled"): v.get_generator(language.Settings()) v = base.TokValue.parseString("</outside")[0] - with pytest.raises("outside"): + with pytest.raises(Exception, match="outside"): v.get_generator(language.Settings(staticdir=t)) def test_spec(self): @@ -194,32 +194,27 @@ class TestMisc: v3 = v2.freeze({}) assert v2.value.val == v3.value.val - def test_fixedlengthvalue(self): + def test_fixedlengthvalue(self, tmpdir): class TT(base.FixedLengthValue): preamble = "m" length = 4 e = TT.expr() assert e.parseString("m@4") - with pytest.raises("invalid value length"): + with pytest.raises(Exception, match="Invalid value length"): e.parseString("m@100") - with pytest.raises("invalid value length"): + with pytest.raises(Exception, match="Invalid value length"): e.parseString("m@1") - with tutils.tmpdir() as t: - p = os.path.join(t, "path") - s = base.Settings(staticdir=t) - with open(p, "wb") as f: - f.write(b"a" * 20) - v = e.parseString("m<path")[0] - with pytest.raises("invalid value length"): - v.values(s) + s = base.Settings(staticdir=str(tmpdir)) + tmpdir.join("path").write_binary(b"a" * 20, ensure=True) + v = e.parseString("m<path")[0] + with pytest.raises(Exception, match="Invalid value length"): + v.values(s) - p = os.path.join(t, "path") - with open(p, "wb") as f: - f.write(b"a" * 4) - v = e.parseString("m<path")[0] - assert v.values(s) + tmpdir.join("path2").write_binary(b"a" * 4, ensure=True) + v = e.parseString("m<path2")[0] + assert v.values(s) class TKeyValue(base.KeyValue): @@ -282,7 +277,7 @@ def test_intfield(): assert v.value == 4 assert v.spec() == "t4" - with pytest.raises("can't exceed"): + with pytest.raises(Exception, match="can't exceed"): e.parseString("t5") @@ -324,9 +319,9 @@ def test_integer(): class BInt(base.Integer): bounds = (1, 5) - with pytest.raises("must be between"): + with pytest.raises(Exception, match="must be between"): BInt(0) - with pytest.raises("must be between"): + with pytest.raises(Exception, match="must be between"): BInt(6) assert BInt(5) assert BInt(1) diff --git a/test/pathod/test_language_http.py b/test/pathod/test_language_http.py index 199fdf64..6ab43fe0 100644 --- a/test/pathod/test_language_http.py +++ b/test/pathod/test_language_http.py @@ -20,7 +20,7 @@ def test_make_error_response(): class TestRequest: def test_nonascii(self): - with pytest.raises("ascii"): + with pytest.raises(Exception, match="ASCII"): parse_request("get:\xf0") def test_err(self): @@ -226,7 +226,7 @@ class TestResponse: assert str(v) def test_nonascii(self): - with pytest.raises("ascii"): + with pytest.raises(Exception, match="ASCII"): language.parse_pathod("foo:b\xf0") def test_parse_header(self): @@ -263,7 +263,7 @@ class TestResponse: def test_websockets(self): r = next(language.parse_pathod("ws")) - with pytest.raises("no websocket key"): + with pytest.raises(Exception, match="No websocket key"): r.resolve(language.Settings()) res = r.resolve(language.Settings(websocket_key=b"foo")) assert res.status_code.string() == b"101" @@ -351,5 +351,5 @@ def test_nested_response_freeze(): def test_unique_components(): - with pytest.raises("multiple body clauses"): + with pytest.raises(Exception, match="multiple body clauses"): language.parse_pathod("400:b@1:b@1") diff --git a/test/pathod/test_language_http2.py b/test/pathod/test_language_http2.py index fdb65a63..4f89adb8 100644 --- a/test/pathod/test_language_http2.py +++ b/test/pathod/test_language_http2.py @@ -39,7 +39,7 @@ class TestRequest: assert req.values(default_settings()) == req.values(default_settings()) def test_nonascii(self): - with pytest.raises("ascii"): + with pytest.raises(Exception, match="ASCII"): parse_request("get:\xf0") def test_err(self): @@ -168,7 +168,7 @@ class TestResponse: assert res.values(default_settings()) == res.values(default_settings()) def test_nonascii(self): - with pytest.raises("ascii"): + with pytest.raises(Exception, match="ASCII"): parse_response("200:\xf0") def test_err(self): diff --git a/test/pathod/test_language_websocket.py b/test/pathod/test_language_websocket.py index 20f6a3a6..e5046591 100644 --- a/test/pathod/test_language_websocket.py +++ b/test/pathod/test_language_websocket.py @@ -130,7 +130,7 @@ class TestWebsocketFrame: assert frm.payload == b"abc" def test_knone(self): - with pytest.raises("expected 4 bytes"): + with pytest.raises(Exception, match="Expected 4 bytes"): self.fr("wf:b'foo':mask:knone") def test_length(self): @@ -138,5 +138,5 @@ class TestWebsocketFrame: frm = self.fr("wf:l2:b'foo'") assert frm.header.payload_length == 2 assert frm.payload == b"fo" - with pytest.raises("expected 1024 bytes"): + with pytest.raises(Exception, match="Expected 1024 bytes"): self.fr("wf:l1024:b'foo'") diff --git a/test/pathod/test_pathoc.py b/test/pathod/test_pathoc.py index a8f79e67..2dd29e20 100644 --- a/test/pathod/test_pathoc.py +++ b/test/pathod/test_pathoc.py @@ -173,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 pytest.raises("connect failed"): + with pytest.raises(Exception, match="CONNECT failed"): c.http_connect(to) c.rfile = io.BytesIO( b"HTTP/1.1 500 OK\r\n" ) - with pytest.raises("connect failed"): + with pytest.raises(Exception, match="CONNECT failed"): c.http_connect(to) c.rfile = io.BytesIO( b"HTTP/1.1 200 OK\r\n" @@ -195,14 +195,14 @@ class TestDaemon(PathocTestDaemon): c.rfile = tutils.treader( b"\x05\xEE" ) - with pytest.raises("SOCKS without authentication"): + with pytest.raises(Exception, match="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" ) - with pytest.raises("SOCKS server error"): + with pytest.raises(Exception, match="SOCKS server error"): c.socks_connect(("example.com", 0xDEAD)) c.rfile = tutils.treader( diff --git a/test/pathod/test_pathod.py b/test/pathod/test_pathod.py index 60ac8072..88480a59 100644 --- a/test/pathod/test_pathod.py +++ b/test/pathod/test_pathod.py @@ -134,7 +134,7 @@ class CommonTests(tservers.DaemonTests): assert len(self.d.log()) == 0 def test_disconnect(self): - with pytest.raises("unexpected eof"): + with pytest.raises(Exception, match="Unexpected EOF"): self.get("202:b@100k:d200") def test_parserr(self): |