Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_no_browser():
with mock.patch("shutil.which") as which:
which.return_value = False
b = browser.Browser()
with taddons.context() as tctx:
b.start()
assert tctx.master.has_log("platform is not supported")
async def test_simple():
with taddons.context(loadcore=False) as tctx:
a = tctx.master.addons
assert len(a) == 0
a.add(TAddon("one"))
assert a.get("one")
assert not a.get("two")
assert len(a) == 1
a.clear()
assert len(a) == 0
assert not a.chain
a.add(TAddon("one"))
a.trigger("running")
a.trigger("response")
assert await tctx.master.await_log("not callable")
def test_set():
sa = core.Core()
with taddons.context(loadcore=False) as tctx:
assert tctx.master.options.server
tctx.command(sa.set, "server", "false")
assert not tctx.master.options.server
with pytest.raises(exceptions.CommandError):
tctx.command(sa.set, "nonexistent")
def test_configure():
up = upstream_auth.UpstreamAuth()
with taddons.context() as tctx:
tctx.configure(up, upstream_auth="test:test")
assert up.auth == b"Basic" + b" " + base64.b64encode(b"test:test")
tctx.configure(up, upstream_auth="test:")
assert up.auth == b"Basic" + b" " + base64.b64encode(b"test:")
tctx.configure(up, upstream_auth=None)
assert not up.auth
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_concurrent(self, tdata):
with taddons.context() as tctx:
sc = tctx.script(
tdata.path(
"mitmproxy/data/addonscripts/concurrent_decorator.py"
)
)
f1, f2 = tflow.tflow(), tflow.tflow()
tctx.cycle(sc, f1)
tctx.cycle(sc, f2)
start = time.time()
while time.time() - start < 5:
if f1.reply.state == f2.reply.state == "committed":
return
raise ValueError("Script never acked")
def test_cut():
c = cut.Cut()
with taddons.context():
tflows = [tflow.tflow(resp=True)]
assert c.cut(tflows, ["request.method"]) == [["GET"]]
assert c.cut(tflows, ["request.scheme"]) == [["http"]]
assert c.cut(tflows, ["request.host"]) == [["address"]]
assert c.cut(tflows, ["request.port"]) == [["22"]]
assert c.cut(tflows, ["request.path"]) == [["/path"]]
assert c.cut(tflows, ["request.url"]) == [["http://address:22/path"]]
assert c.cut(tflows, ["request.content"]) == [[b"content"]]
assert c.cut(tflows, ["request.header[header]"]) == [["qvalue"]]
assert c.cut(tflows, ["request.header[unknown]"]) == [[""]]
assert c.cut(tflows, ["response.status_code"]) == [["200"]]
assert c.cut(tflows, ["response.reason"]) == [["OK"]]
assert c.cut(tflows, ["response.content"]) == [[b"message"]]
assert c.cut(tflows, ["response.header[header-response]"]) == [["svalue"]]
assert c.cut(tflows, ["moo"]) == [[""]]
def test_request(self):
sc = stickycookie.StickyCookie()
with taddons.context(sc) as tctx:
tctx.configure(sc, stickycookie=".*")
f = self._response(sc, "SSID=mooo", "www.google.com")
assert "cookie" not in f.request.headers
sc.request(f)
assert "cookie" in f.request.headers
async def test_nonexistent_file(self):
rf = readfile.ReadFile()
with taddons.context(rf) as tctx:
with pytest.raises(exceptions.FlowReadException):
await rf.load_flows_from_path("nonexistent")
assert await tctx.master.await_log("nonexistent")
def test_simple():
sa = streambodies.StreamBodies()
with taddons.context(sa) as tctx:
with pytest.raises(exceptions.OptionsError):
tctx.configure(sa, stream_large_bodies = "invalid")
tctx.configure(sa, stream_large_bodies = "10")
f = tflow.tflow()
f.request.content = b""
f.request.headers["Content-Length"] = "1024"
assert not f.request.stream
sa.requestheaders(f)
assert f.request.stream
f = tflow.tflow(resp=True)
f.response.content = b""
f.response.headers["Content-Length"] = "1024"
assert not f.response.stream
sa.responseheaders(f)
def test_output(self, outfile, expected_out, expected_err, capfd):
t = termlog.TermLog(outfile=outfile)
with taddons.context(t) as tctx:
tctx.options.termlog_verbosity = "info"
tctx.configure(t)
t.log(log.LogEntry("one", "info"))
t.log(log.LogEntry("two", "debug"))
t.log(log.LogEntry("three", "warn"))
t.log(log.LogEntry("four", "error"))
out, err = capfd.readouterr()
assert out.strip().splitlines() == expected_out
assert err.strip().splitlines() == expected_err