aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/mitmproxy/builtins/test_dumper.py86
-rw-r--r--test/mitmproxy/builtins/test_filestreamer.py (renamed from test/mitmproxy/builtins/test_stream.py)4
-rw-r--r--test/mitmproxy/builtins/test_setheaders.py64
-rw-r--r--test/mitmproxy/mastertest.py7
-rw-r--r--test/mitmproxy/test_dump.py83
-rw-r--r--test/mitmproxy/test_flow.py60
-rw-r--r--test/mitmproxy/test_options.py11
-rw-r--r--test/mitmproxy/test_protocol_http2.py177
-rw-r--r--test/netlib/http/http1/test_read.py26
9 files changed, 363 insertions, 155 deletions
diff --git a/test/mitmproxy/builtins/test_dumper.py b/test/mitmproxy/builtins/test_dumper.py
new file mode 100644
index 00000000..57e3d036
--- /dev/null
+++ b/test/mitmproxy/builtins/test_dumper.py
@@ -0,0 +1,86 @@
+from .. import tutils, mastertest
+from six.moves import cStringIO as StringIO
+
+from mitmproxy.builtins import dumper
+from mitmproxy.flow import state
+from mitmproxy import exceptions
+from mitmproxy import dump
+from mitmproxy import models
+import netlib.tutils
+import mock
+
+
+class TestDumper(mastertest.MasterTest):
+ def test_simple(self):
+ d = dumper.Dumper()
+ sio = StringIO()
+
+ d.configure(dump.Options(tfile = sio, flow_detail = 0))
+ d.response(tutils.tflow())
+ assert not sio.getvalue()
+
+ d.configure(dump.Options(tfile = sio, flow_detail = 4))
+ d.response(tutils.tflow())
+ assert sio.getvalue()
+
+ sio = StringIO()
+ d.configure(dump.Options(tfile = sio, flow_detail = 4))
+ d.response(tutils.tflow(resp=True))
+ assert "<<" in sio.getvalue()
+
+ sio = StringIO()
+ d.configure(dump.Options(tfile = sio, flow_detail = 4))
+ d.response(tutils.tflow(err=True))
+ assert "<<" in sio.getvalue()
+
+ sio = StringIO()
+ d.configure(dump.Options(tfile = sio, flow_detail = 4))
+ flow = tutils.tflow()
+ flow.request = netlib.tutils.treq()
+ flow.request.stickycookie = True
+ flow.client_conn = mock.MagicMock()
+ flow.client_conn.address.host = "foo"
+ flow.response = netlib.tutils.tresp(content=None)
+ flow.response.is_replay = True
+ flow.response.status_code = 300
+ d.response(flow)
+ assert sio.getvalue()
+
+ sio = StringIO()
+ d.configure(dump.Options(tfile = sio, flow_detail = 4))
+ flow = tutils.tflow(resp=netlib.tutils.tresp(content=b"{"))
+ flow.response.headers["content-type"] = "application/json"
+ flow.response.status_code = 400
+ d.response(flow)
+ assert sio.getvalue()
+
+ sio = StringIO()
+ d.configure(dump.Options(tfile = sio))
+ flow = tutils.tflow()
+ flow.request.content = None
+ flow.response = models.HTTPResponse.wrap(netlib.tutils.tresp())
+ flow.response.content = None
+ d.response(flow)
+ assert "content missing" in sio.getvalue()
+
+
+class TestContentView(mastertest.MasterTest):
+ @mock.patch("mitmproxy.contentviews.get_content_view")
+ def test_contentview(self, get_content_view):
+ se = exceptions.ContentViewException(""), ("x", iter([]))
+ get_content_view.side_effect = se
+
+ s = state.State()
+ sio = StringIO()
+ m = mastertest.RecordingMaster(
+ dump.Options(
+ flow_detail=4,
+ verbosity=3,
+ tfile=sio,
+ ),
+ None, s
+ )
+ d = dumper.Dumper()
+ m.addons.add(d)
+ self.invoke(m, "response", tutils.tflow())
+ assert "Content viewer failed" in m.event_log[0][1]
diff --git a/test/mitmproxy/builtins/test_stream.py b/test/mitmproxy/builtins/test_filestreamer.py
index edaa41d2..002006b7 100644
--- a/test/mitmproxy/builtins/test_stream.py
+++ b/test/mitmproxy/builtins/test_filestreamer.py
@@ -4,7 +4,7 @@ from .. import tutils, mastertest
import os.path
-from mitmproxy.builtins import stream
+from mitmproxy.builtins import filestreamer
from mitmproxy.flow import master, FlowReader
from mitmproxy.flow import state
from mitmproxy.flow import options
@@ -27,7 +27,7 @@ class TestStream(mastertest.MasterTest):
None,
s
)
- sa = stream.Stream()
+ sa = filestreamer.FileStreamer()
m.addons.add(sa)
f = tutils.tflow(resp=True)
diff --git a/test/mitmproxy/builtins/test_setheaders.py b/test/mitmproxy/builtins/test_setheaders.py
new file mode 100644
index 00000000..1a8d048c
--- /dev/null
+++ b/test/mitmproxy/builtins/test_setheaders.py
@@ -0,0 +1,64 @@
+from .. import tutils, mastertest
+
+from mitmproxy.builtins import setheaders
+from mitmproxy.flow import state
+from mitmproxy.flow import options
+
+
+class TestSetHeaders(mastertest.MasterTest):
+ def mkmaster(self, **opts):
+ s = state.State()
+ m = mastertest.RecordingMaster(options.Options(**opts), None, s)
+ sh = setheaders.SetHeaders()
+ m.addons.add(sh)
+ return m, sh
+
+ def test_configure(self):
+ sh = setheaders.SetHeaders()
+ tutils.raises(
+ "invalid setheader filter pattern",
+ sh.configure,
+ options.Options(
+ setheaders = [("~b", "one", "two")]
+ )
+ )
+
+ def test_setheaders(self):
+ m, sh = self.mkmaster(
+ setheaders = [
+ ("~q", "one", "two"),
+ ("~s", "one", "three")
+ ]
+ )
+ f = tutils.tflow()
+ f.request.headers["one"] = "xxx"
+ self.invoke(m, "request", f)
+ assert f.request.headers["one"] == "two"
+
+ f = tutils.tflow(resp=True)
+ f.response.headers["one"] = "xxx"
+ self.invoke(m, "response", f)
+ assert f.response.headers["one"] == "three"
+
+ m, sh = self.mkmaster(
+ setheaders = [
+ ("~s", "one", "two"),
+ ("~s", "one", "three")
+ ]
+ )
+ f = tutils.tflow(resp=True)
+ f.request.headers["one"] = "xxx"
+ f.response.headers["one"] = "xxx"
+ self.invoke(m, "response", f)
+ assert f.response.headers.get_all("one") == ["two", "three"]
+
+ m, sh = self.mkmaster(
+ setheaders = [
+ ("~q", "one", "two"),
+ ("~q", "one", "three")
+ ]
+ )
+ f = tutils.tflow()
+ f.request.headers["one"] = "xxx"
+ self.invoke(m, "request", f)
+ assert f.request.headers.get_all("one") == ["two", "three"]
diff --git a/test/mitmproxy/mastertest.py b/test/mitmproxy/mastertest.py
index d1fe8cb4..dcc0dc48 100644
--- a/test/mitmproxy/mastertest.py
+++ b/test/mitmproxy/mastertest.py
@@ -8,11 +8,12 @@ from mitmproxy import flow, proxy, models, controller
class MasterTest:
- def invoke(self, master, handler, message):
+ def invoke(self, master, handler, *message):
with master.handlecontext():
func = getattr(master, handler)
- func(message)
- message.reply = controller.DummyReply()
+ func(*message)
+ if message:
+ message[0].reply = controller.DummyReply()
def cycle(self, master, content):
f = tutils.tflow(req=netlib.tutils.treq(content=content))
diff --git a/test/mitmproxy/test_dump.py b/test/mitmproxy/test_dump.py
index c94630a9..90f33264 100644
--- a/test/mitmproxy/test_dump.py
+++ b/test/mitmproxy/test_dump.py
@@ -1,67 +1,11 @@
import os
from six.moves import cStringIO as StringIO
-from mitmproxy.exceptions import ContentViewException
-import netlib.tutils
-
-from mitmproxy import dump, flow, models, exceptions
+from mitmproxy import dump, flow, exceptions
from . import tutils, mastertest
import mock
-def test_strfuncs():
- o = dump.Options(
- tfile = StringIO(),
- flow_detail = 0,
- )
- m = dump.DumpMaster(None, o)
-
- m.o.flow_detail = 0
- m.echo_flow(tutils.tflow())
- assert not o.tfile.getvalue()
-
- m.o.flow_detail = 4
- m.echo_flow(tutils.tflow())
- assert o.tfile.getvalue()
-
- o.tfile = StringIO()
- m.echo_flow(tutils.tflow(resp=True))
- assert "<<" in o.tfile.getvalue()
-
- o.tfile = StringIO()
- m.echo_flow(tutils.tflow(err=True))
- assert "<<" in o.tfile.getvalue()
-
- flow = tutils.tflow()
- flow.request = netlib.tutils.treq()
- flow.request.stickycookie = True
- flow.client_conn = mock.MagicMock()
- flow.client_conn.address.host = "foo"
- flow.response = netlib.tutils.tresp(content=None)
- flow.response.is_replay = True
- flow.response.status_code = 300
- m.echo_flow(flow)
-
- flow = tutils.tflow(resp=netlib.tutils.tresp(content=b"{"))
- flow.response.headers["content-type"] = "application/json"
- flow.response.status_code = 400
- m.echo_flow(flow)
-
-
-@mock.patch("mitmproxy.contentviews.get_content_view")
-def test_contentview(get_content_view):
- get_content_view.side_effect = ContentViewException(""), ("x", iter([]))
-
- o = dump.Options(
- flow_detail=4,
- verbosity=3,
- tfile=StringIO(),
- )
- m = dump.DumpMaster(None, o)
- m.echo_flow(tutils.tflow())
- assert "Content viewer failed" in m.options.tfile.getvalue()
-
-
class TestDumpMaster(mastertest.MasterTest):
def dummy_cycle(self, master, n, content):
mastertest.MasterTest.dummy_cycle(self, master, n, content)
@@ -72,11 +16,7 @@ class TestDumpMaster(mastertest.MasterTest):
options["verbosity"] = 0
if "flow_detail" not in options:
options["flow_detail"] = 0
- o = dump.Options(
- filtstr=filt,
- tfile=StringIO(),
- **options
- )
+ o = dump.Options(filtstr=filt, tfile=StringIO(), **options)
return dump.DumpMaster(None, o)
def test_basic(self):
@@ -104,24 +44,10 @@ class TestDumpMaster(mastertest.MasterTest):
)
m = dump.DumpMaster(None, o)
f = tutils.tflow(err=True)
- m.request(f)
+ m.error(f)
assert m.error(f)
assert "error" in o.tfile.getvalue()
- def test_missing_content(self):
- o = dump.Options(
- flow_detail=3,
- tfile=StringIO(),
- )
- m = dump.DumpMaster(None, o)
- f = tutils.tflow()
- f.request.content = None
- m.request(f)
- f.response = models.HTTPResponse.wrap(netlib.tutils.tresp())
- f.response.content = None
- m.response(f)
- assert "content missing" in o.tfile.getvalue()
-
def test_replay(self):
o = dump.Options(server_replay=["nonexistent"], kill=True)
tutils.raises(dump.DumpError, dump.DumpMaster, None, o)
@@ -155,9 +81,8 @@ class TestDumpMaster(mastertest.MasterTest):
self.flowfile(p)
assert "GET" in self.dummy_cycle(
self.mkmaster(None, flow_detail=1, rfile=p),
- 0, b"",
+ 1, b"",
)
-
tutils.raises(
dump.DumpError,
self.mkmaster, None, verbosity=1, rfile="/nonexistent"
diff --git a/test/mitmproxy/test_flow.py b/test/mitmproxy/test_flow.py
index 10163401..90f7f915 100644
--- a/test/mitmproxy/test_flow.py
+++ b/test/mitmproxy/test_flow.py
@@ -751,7 +751,11 @@ class TestFlowMaster:
f = tutils.tflow(resp=True)
pb = [tutils.tflow(resp=True), f]
- fm = flow.FlowMaster(None, DummyServer(ProxyConfig()), s)
+ fm = flow.FlowMaster(
+ flow.options.Options(),
+ DummyServer(ProxyConfig()),
+ s
+ )
assert not fm.start_server_playback(
pb,
False,
@@ -779,7 +783,7 @@ class TestFlowMaster:
f.response = HTTPResponse.wrap(netlib.tutils.tresp(content=f.request))
pb = [f]
- fm = flow.FlowMaster(None, None, s)
+ fm = flow.FlowMaster(flow.options.Options(), None, s)
fm.refresh_server_playback = True
assert not fm.do_server_playback(tutils.tflow())
@@ -961,55 +965,3 @@ class TestClientConnection:
assert c3.get_state() == c.get_state()
assert str(c)
-
-
-def test_setheaders():
- h = flow.SetHeaders()
- h.add("~q", "foo", "bar")
- assert h.lst
-
- h.set(
- [
- (".*", "one", "two"),
- (".*", "three", "four"),
- ]
- )
- assert h.count() == 2
-
- h.clear()
- assert not h.lst
-
- h.add("~q", "foo", "bar")
- h.add("~s", "foo", "bar")
-
- v = h.get_specs()
- assert v == [('~q', 'foo', 'bar'), ('~s', 'foo', 'bar')]
- assert h.count() == 2
- h.clear()
- assert h.count() == 0
-
- f = tutils.tflow()
- f.request.content = b"foo"
- h.add("~s", "foo", "bar")
- h.run(f)
- assert f.request.content == b"foo"
-
- h.clear()
- h.add("~s", "one", "two")
- h.add("~s", "one", "three")
- f = tutils.tflow(resp=True)
- f.request.headers["one"] = "xxx"
- f.response.headers["one"] = "xxx"
- h.run(f)
- assert f.request.headers["one"] == "xxx"
- assert f.response.headers.get_all("one") == ["two", "three"]
-
- h.clear()
- h.add("~q", "one", "two")
- h.add("~q", "one", "three")
- f = tutils.tflow()
- f.request.headers["one"] = "xxx"
- h.run(f)
- assert f.request.headers.get_all("one") == ["two", "three"]
-
- assert not h.add("~", "foo", "bar")
diff --git a/test/mitmproxy/test_options.py b/test/mitmproxy/test_options.py
index cdb0d765..af619b27 100644
--- a/test/mitmproxy/test_options.py
+++ b/test/mitmproxy/test_options.py
@@ -52,6 +52,17 @@ def test_setter():
o.setter("nonexistent")
+def test_toggler():
+ o = TO(two=True)
+ f = o.toggler("two")
+ f()
+ assert o.two is False
+ f()
+ assert o.two is True
+ with tutils.raises("no such option"):
+ o.toggler("nonexistent")
+
+
def test_rollback():
o = TO(one="two")
diff --git a/test/mitmproxy/test_protocol_http2.py b/test/mitmproxy/test_protocol_http2.py
index a100ac2d..b8f724bd 100644
--- a/test/mitmproxy/test_protocol_http2.py
+++ b/test/mitmproxy/test_protocol_http2.py
@@ -87,31 +87,31 @@ class _Http2ServerBase(netlib_tservers.ServerTestBase):
class _Http2TestBase(object):
@classmethod
- def setup_class(self):
- self.config = ProxyConfig(**self.get_proxy_config())
+ def setup_class(cls):
+ cls.config = ProxyConfig(**cls.get_proxy_config())
- tmaster = tservers.TestMaster(self.config)
+ tmaster = tservers.TestMaster(cls.config)
tmaster.start_app(APP_HOST, APP_PORT)
- self.proxy = tservers.ProxyThread(tmaster)
- self.proxy.start()
+ cls.proxy = tservers.ProxyThread(tmaster)
+ cls.proxy.start()
@classmethod
def teardown_class(cls):
cls.proxy.shutdown()
- @property
- def master(self):
- return self.proxy.tmaster
-
@classmethod
def get_proxy_config(cls):
cls.cadir = os.path.join(tempfile.gettempdir(), "mitmproxy")
return dict(
- no_upstream_cert = False,
- cadir = cls.cadir,
- authenticator = None,
+ no_upstream_cert=False,
+ cadir=cls.cadir,
+ authenticator=None,
)
+ @property
+ def master(self):
+ return self.proxy.tmaster
+
def setup(self):
self.master.clear_log()
self.master.state.clear()
@@ -150,13 +150,17 @@ class _Http2TestBase(object):
stream_id=1,
headers=[],
body=b'',
+ end_stream=None,
priority_exclusive=None,
priority_depends_on=None,
priority_weight=None):
+ if end_stream is None:
+ end_stream = (len(body) == 0)
+
h2_conn.send_headers(
stream_id=stream_id,
headers=headers,
- end_stream=(len(body) == 0),
+ end_stream=end_stream,
priority_exclusive=priority_exclusive,
priority_depends_on=priority_depends_on,
priority_weight=priority_weight,
@@ -376,6 +380,153 @@ class TestRequestWithPriority(_Http2Test):
@requires_alpn
+class TestPriority(_Http2Test):
+ priority_data = None
+
+ @classmethod
+ def handle_server_event(self, event, h2_conn, rfile, wfile):
+ if isinstance(event, h2.events.ConnectionTerminated):
+ return False
+ elif isinstance(event, h2.events.PriorityUpdated):
+ self.priority_data = (event.exclusive, event.depends_on, event.weight)
+ elif isinstance(event, h2.events.RequestReceived):
+ import warnings
+ with warnings.catch_warnings():
+ # Ignore UnicodeWarning:
+ # h2/utilities.py:64: UnicodeWarning: Unicode equal comparison
+ # failed to convert both arguments to Unicode - interpreting
+ # them as being unequal.
+ # elif header[0] in (b'cookie', u'cookie') and len(header[1]) < 20:
+
+ warnings.simplefilter("ignore")
+
+ headers = [(':status', '200')]
+ h2_conn.send_headers(event.stream_id, headers)
+ h2_conn.end_stream(event.stream_id)
+ wfile.write(h2_conn.data_to_send())
+ wfile.flush()
+ return True
+
+ def test_priority(self):
+ client, h2_conn = self._setup_connection()
+
+ h2_conn.prioritize(1, exclusive=True, depends_on=0, weight=42)
+ client.wfile.write(h2_conn.data_to_send())
+ client.wfile.flush()
+
+ self._send_request(
+ client.wfile,
+ h2_conn,
+ headers=[
+ (':authority', "127.0.0.1:%s" % self.server.server.address.port),
+ (':method', 'GET'),
+ (':scheme', 'https'),
+ (':path', '/'),
+ ],
+ )
+
+ done = False
+ while not done:
+ try:
+ raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+ events = h2_conn.receive_data(raw)
+ except HttpException:
+ print(traceback.format_exc())
+ assert False
+
+ client.wfile.write(h2_conn.data_to_send())
+ client.wfile.flush()
+
+ for event in events:
+ if isinstance(event, h2.events.StreamEnded):
+ done = True
+
+ h2_conn.close_connection()
+ client.wfile.write(h2_conn.data_to_send())
+ client.wfile.flush()
+
+ assert len(self.master.state.flows) == 1
+ assert self.priority_data == (True, 0, 42)
+
+
+@requires_alpn
+class TestPriorityWithExistingStream(_Http2Test):
+ priority_data = []
+
+ @classmethod
+ def handle_server_event(self, event, h2_conn, rfile, wfile):
+ if isinstance(event, h2.events.ConnectionTerminated):
+ return False
+ elif isinstance(event, h2.events.PriorityUpdated):
+ self.priority_data.append((event.exclusive, event.depends_on, event.weight))
+ elif isinstance(event, h2.events.RequestReceived):
+ assert not event.priority_updated
+
+ import warnings
+ with warnings.catch_warnings():
+ # Ignore UnicodeWarning:
+ # h2/utilities.py:64: UnicodeWarning: Unicode equal comparison
+ # failed to convert both arguments to Unicode - interpreting
+ # them as being unequal.
+ # elif header[0] in (b'cookie', u'cookie') and len(header[1]) < 20:
+
+ warnings.simplefilter("ignore")
+
+ headers = [(':status', '200')]
+ h2_conn.send_headers(event.stream_id, headers)
+ wfile.write(h2_conn.data_to_send())
+ wfile.flush()
+ elif isinstance(event, h2.events.StreamEnded):
+ h2_conn.end_stream(event.stream_id)
+ wfile.write(h2_conn.data_to_send())
+ wfile.flush()
+ return True
+
+ def test_priority_with_existing_stream(self):
+ client, h2_conn = self._setup_connection()
+
+ self._send_request(
+ client.wfile,
+ h2_conn,
+ headers=[
+ (':authority', "127.0.0.1:%s" % self.server.server.address.port),
+ (':method', 'GET'),
+ (':scheme', 'https'),
+ (':path', '/'),
+ ],
+ end_stream=False,
+ )
+
+ h2_conn.prioritize(1, exclusive=True, depends_on=0, weight=42)
+ h2_conn.end_stream(1)
+ client.wfile.write(h2_conn.data_to_send())
+ client.wfile.flush()
+
+ done = False
+ while not done:
+ try:
+ raw = b''.join(framereader.http2_read_raw_frame(client.rfile))
+ events = h2_conn.receive_data(raw)
+ except HttpException:
+ print(traceback.format_exc())
+ assert False
+
+ client.wfile.write(h2_conn.data_to_send())
+ client.wfile.flush()
+
+ for event in events:
+ if isinstance(event, h2.events.StreamEnded):
+ done = True
+
+ h2_conn.close_connection()
+ client.wfile.write(h2_conn.data_to_send())
+ client.wfile.flush()
+
+ assert len(self.master.state.flows) == 1
+ assert self.priority_data == [(True, 0, 42)]
+
+
+@requires_alpn
class TestStreamResetFromServer(_Http2Test):
@classmethod
diff --git a/test/netlib/http/http1/test_read.py b/test/netlib/http/http1/test_read.py
index 5285ac1d..c8a40ecb 100644
--- a/test/netlib/http/http1/test_read.py
+++ b/test/netlib/http/http1/test_read.py
@@ -1,6 +1,9 @@
from __future__ import absolute_import, print_function, division
+
from io import BytesIO
from mock import Mock
+import pytest
+
from netlib.exceptions import HttpException, HttpSyntaxException, HttpReadDisconnect, TcpDisconnect
from netlib.http import Headers
from netlib.http.http1.read import (
@@ -23,11 +26,18 @@ def test_get_header_tokens():
assert get_header_tokens(headers, "foo") == ["bar", "voing", "oink"]
-def test_read_request():
- rfile = BytesIO(b"GET / HTTP/1.1\r\n\r\nskip")
+@pytest.mark.parametrize("input", [
+ b"GET / HTTP/1.1\r\n\r\nskip",
+ b"GET / HTTP/1.1\r\n\r\nskip",
+ b"GET / HTTP/1.1\r\n\r\nskip",
+ b"GET / HTTP/1.1 \r\n\r\nskip",
+])
+def test_read_request(input):
+ rfile = BytesIO(input)
r = read_request(rfile)
assert r.method == "GET"
assert r.content == b""
+ assert r.http_version == "HTTP/1.1"
assert r.timestamp_end
assert rfile.read() == b"skip"
@@ -50,11 +60,19 @@ def test_read_request_head():
assert rfile.read() == b"skip"
-def test_read_response():
+@pytest.mark.parametrize("input", [
+ b"HTTP/1.1 418 I'm a teapot\r\n\r\nbody",
+ b"HTTP/1.1 418 I'm a teapot\r\n\r\nbody",
+ b"HTTP/1.1 418 I'm a teapot\r\n\r\nbody",
+ b"HTTP/1.1 418 I'm a teapot \r\n\r\nbody",
+])
+def test_read_response(input):
req = treq()
- rfile = BytesIO(b"HTTP/1.1 418 I'm a teapot\r\n\r\nbody")
+ rfile = BytesIO(input)
r = read_response(rfile, req)
+ assert r.http_version == "HTTP/1.1"
assert r.status_code == 418
+ assert r.reason == "I'm a teapot"
assert r.content == b"body"
assert r.timestamp_end
generated by cgit v1.2.3 (git 2.25.1) at 2025-08-14 10:31:09 +0000