mirror of
https://github.com/mitmproxy/mitmproxy.git
synced 2025-02-07 22:16:57 +00:00
Merge pull request #1008 from dufferzafar/test-examples
Some tests for examples
This commit is contained in:
commit
1bfc27c3f6
@ -14,7 +14,7 @@ def response(context, flow):
|
||||
if flow.request.host in context.iframe_url:
|
||||
return
|
||||
with decoded(flow.response): # Remove content encoding (gzip, ...)
|
||||
html = BeautifulSoup(flow.response.content)
|
||||
html = BeautifulSoup(flow.response.content, "lxml")
|
||||
if html.body:
|
||||
iframe = html.new_tag(
|
||||
"iframe",
|
||||
|
@ -1,5 +1,5 @@
|
||||
def request(context, flow):
|
||||
if "application/x-www-form-urlencoded" in flow.request.headers.get("content-type", ""):
|
||||
form = flow.request.get_form_urlencoded()
|
||||
form = flow.request.urlencoded_form
|
||||
if form is not None:
|
||||
form["mitmproxy"] = ["rocks"]
|
||||
flow.request.set_form_urlencoded(form)
|
||||
flow.request.urlencoded_form = form
|
||||
|
@ -1,6 +1,5 @@
|
||||
|
||||
def request(context, flow):
|
||||
q = flow.request.get_query()
|
||||
q = flow.request.query
|
||||
if q:
|
||||
q["mitmproxy"] = ["rocks"]
|
||||
flow.request.set_query(q)
|
||||
flow.request.query = q
|
||||
|
@ -1,11 +1,42 @@
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
from contextlib import contextmanager
|
||||
|
||||
from mitmproxy import utils, script
|
||||
from mitmproxy.proxy import config
|
||||
from . import tservers
|
||||
from netlib import tutils as netutils
|
||||
from netlib.http import Headers
|
||||
from . import tservers, tutils
|
||||
|
||||
example_dir = utils.Data(__name__).path("../../examples")
|
||||
|
||||
|
||||
class DummyContext(object):
|
||||
"""Emulate script.ScriptContext() functionality."""
|
||||
|
||||
contentview = None
|
||||
|
||||
def log(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def add_contentview(self, view_obj):
|
||||
self.contentview = view_obj
|
||||
|
||||
def remove_contentview(self, view_obj):
|
||||
self.contentview = None
|
||||
|
||||
|
||||
@contextmanager
|
||||
def example(command):
|
||||
command = os.path.join(example_dir, command)
|
||||
ctx = DummyContext()
|
||||
s = script.Script(command, ctx)
|
||||
yield s
|
||||
s.unload()
|
||||
|
||||
|
||||
def test_load_scripts():
|
||||
example_dir = utils.Data(__name__).path("../../examples")
|
||||
scripts = glob.glob("%s/*.py" % example_dir)
|
||||
|
||||
tmaster = tservers.TestMaster(config.ProxyConfig())
|
||||
@ -28,3 +59,87 @@ def test_load_scripts():
|
||||
raise
|
||||
else:
|
||||
s.unload()
|
||||
|
||||
|
||||
def test_add_header():
|
||||
flow = tutils.tflow(resp=netutils.tresp())
|
||||
with example("add_header.py") as ex:
|
||||
ex.run("response", flow)
|
||||
assert flow.response.headers["newheader"] == "foo"
|
||||
|
||||
|
||||
def test_custom_contentviews():
|
||||
with example("custom_contentviews.py") as ex:
|
||||
pig = ex.ctx.contentview
|
||||
_, fmt = pig("<html>test!</html>")
|
||||
assert any('esttay!' in val[0][1] for val in fmt)
|
||||
assert not pig("gobbledygook")
|
||||
|
||||
|
||||
def test_iframe_injector():
|
||||
with tutils.raises(script.ScriptException):
|
||||
with example("iframe_injector.py") as ex:
|
||||
pass
|
||||
|
||||
flow = tutils.tflow(resp=netutils.tresp(content="<html>mitmproxy</html>"))
|
||||
with example("iframe_injector.py http://example.org/evil_iframe") as ex:
|
||||
ex.run("response", flow)
|
||||
content = flow.response.content
|
||||
assert 'iframe' in content and 'evil_iframe' in content
|
||||
|
||||
|
||||
def test_modify_form():
|
||||
form_header = Headers(content_type="application/x-www-form-urlencoded")
|
||||
flow = tutils.tflow(req=netutils.treq(headers=form_header))
|
||||
with example("modify_form.py") as ex:
|
||||
ex.run("request", flow)
|
||||
assert flow.request.urlencoded_form["mitmproxy"] == ["rocks"]
|
||||
|
||||
|
||||
def test_modify_querystring():
|
||||
flow = tutils.tflow(req=netutils.treq(path="/search?q=term"))
|
||||
with example("modify_querystring.py") as ex:
|
||||
ex.run("request", flow)
|
||||
assert flow.request.query["mitmproxy"] == ["rocks"]
|
||||
|
||||
|
||||
def test_modify_response_body():
|
||||
with tutils.raises(script.ScriptException):
|
||||
with example("modify_response_body.py") as ex:
|
||||
pass
|
||||
|
||||
flow = tutils.tflow(resp=netutils.tresp(content="I <3 mitmproxy"))
|
||||
with example("modify_response_body.py mitmproxy rocks") as ex:
|
||||
assert ex.ctx.old == "mitmproxy" and ex.ctx.new == "rocks"
|
||||
ex.run("response", flow)
|
||||
assert flow.response.content == "I <3 rocks"
|
||||
|
||||
|
||||
def test_redirect_requests():
|
||||
flow = tutils.tflow(req=netutils.treq(host="example.org"))
|
||||
with example("redirect_requests.py") as ex:
|
||||
ex.run("request", flow)
|
||||
assert flow.request.host == "mitmproxy.org"
|
||||
|
||||
|
||||
def test_har_extractor():
|
||||
with tutils.raises(script.ScriptException):
|
||||
with example("har_extractor.py") as ex:
|
||||
pass
|
||||
|
||||
times = dict(
|
||||
timestamp_start=746203272,
|
||||
timestamp_end=746203272,
|
||||
)
|
||||
|
||||
flow = tutils.tflow(
|
||||
req=netutils.treq(**times),
|
||||
resp=netutils.tresp(**times)
|
||||
)
|
||||
|
||||
with example("har_extractor.py -") as ex:
|
||||
ex.run("response", flow)
|
||||
|
||||
with open(tutils.test_data.path("data/har_extractor.har")) as fp:
|
||||
test_data = json.load(fp)
|
||||
assert json.loads(ex.ctx.HARLog.json()) == test_data["test_response"]
|
||||
|
@ -1,4 +1,3 @@
|
||||
import json
|
||||
from textwrap import dedent
|
||||
|
||||
import netlib.tutils
|
||||
|
@ -1,37 +0,0 @@
|
||||
import json
|
||||
import netlib.tutils
|
||||
from . import tutils
|
||||
|
||||
from examples import har_extractor
|
||||
|
||||
|
||||
class Context(object):
|
||||
pass
|
||||
|
||||
|
||||
trequest = netlib.tutils.treq(
|
||||
timestamp_start=746203272,
|
||||
timestamp_end=746203272,
|
||||
)
|
||||
|
||||
tresponse = netlib.tutils.tresp(
|
||||
timestamp_start=746203272,
|
||||
timestamp_end=746203272,
|
||||
)
|
||||
|
||||
|
||||
def test_start():
|
||||
tutils.raises(ValueError, har_extractor.start, Context(), [])
|
||||
|
||||
|
||||
def test_response():
|
||||
ctx = Context()
|
||||
ctx.HARLog = har_extractor._HARLog([])
|
||||
ctx.seen_server = set()
|
||||
|
||||
fl = tutils.tflow(req=trequest, resp=tresponse)
|
||||
har_extractor.response(ctx, fl)
|
||||
|
||||
with open(tutils.test_data.path("data/har_extractor.har")) as fp:
|
||||
test_data = json.load(fp)
|
||||
assert json.loads(ctx.HARLog.json()) == test_data["test_response"]
|
Loading…
x
Reference in New Issue
Block a user