Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_stdin(self, stdin, data, corrupt_data):
rf = readfile.ReadFileStdin()
with taddons.context(rf):
with asynctest.patch('mitmproxy.master.Master.load_flow') as mck:
stdin.buffer = data
assert not mck.awaited
await rf.load_flows(stdin.buffer)
assert mck.awaited
stdin.buffer = corrupt_data
with pytest.raises(exceptions.FlowReadException):
await rf.load_flows(stdin.buffer)
def show(flow_detail, flows):
d = dumper.Dumper()
with taddons.context() as ctx:
ctx.configure(d, flow_detail=flow_detail)
for f in flows:
ctx.cycle(d, f)
def test_configure(self, tdata):
up = proxyauth.ProxyAuth()
with taddons.context(up) as ctx:
with pytest.raises(exceptions.OptionsError):
ctx.configure(up, proxyauth="foo")
ctx.configure(up, proxyauth="foo:bar")
assert up.singleuser == ["foo", "bar"]
ctx.configure(up, proxyauth=None)
assert up.singleuser is None
ctx.configure(up, proxyauth="any")
assert up.nonanonymous
ctx.configure(up, proxyauth=None)
assert not up.nonanonymous
with mock.patch('ldap3.Server', return_value="ldap://fake_server:389 - cleartext"):
with mock.patch('ldap3.Connection', return_value="test"):
def test_simple():
r = intercept.Intercept()
with taddons.context(options=options.Options()) as tctx:
assert not r.filt
tctx.configure(r, intercept="~q")
assert r.filt
assert tctx.options.intercept_active
with pytest.raises(exceptions.OptionsError):
tctx.configure(r, intercept="~~")
tctx.configure(r, intercept=None)
assert not r.filt
assert not tctx.options.intercept_active
tctx.configure(r, intercept="~s")
f = tflow.tflow(resp=True)
tctx.cycle(r, f)
assert f.intercepted
def test_duplicate():
v = view.View()
with taddons.context():
f = [
tflow.tflow(),
tflow.tflow(),
]
v.add(f)
assert len(v) == 2
v.duplicate(f)
assert len(v) == 4
assert v.focus.index == 2
def test_server_playback_kill():
s = serverplayback.ServerPlayback()
with taddons.context(s) as tctx:
tctx.configure(
s,
server_replay_refresh=True,
server_replay_kill_extra=True
)
f = tflow.tflow()
f.response = mitmproxy.test.tutils.tresp(content=f.request.content)
s.load_flows([f])
f = tflow.tflow()
f.request.host = "nonexistent"
tctx.cycle(s, f)
assert f.reply.value == exceptions.Kill
async def test_concurrent_err(self, tdata):
with taddons.context() as tctx:
tctx.script(
tdata.path(
"mitmproxy/data/addonscripts/concurrent_decorator_err.py"
)
)
assert await tctx.master.await_log("decorator not supported")
def test_movement():
v = view.View()
with taddons.context():
v.go(0)
v.add([
tflow.tflow(),
tflow.tflow(),
tflow.tflow(),
tflow.tflow(),
tflow.tflow(),
])
assert v.focus.index == 0
v.go(-1)
assert v.focus.index == 4
v.go(0)
assert v.focus.index == 0
v.go(1)
assert v.focus.index == 1
v.go(999)
def test_unknown():
with taddons.context() as tctx:
b = mitmproxy.types._UnknownType()
assert b.is_valid(tctx.master.commands, mitmproxy.types.Unknown, "foo") is False
assert b.is_valid(tctx.master.commands, mitmproxy.types.Unknown, 1) is False
assert b.completion(tctx.master.commands, mitmproxy.types.Unknown, "") == []
assert b.parse(tctx.master.commands, mitmproxy.types.Unknown, "foo") == "foo"
def test_configure():
d = dumper.Dumper()
with taddons.context(d) as ctx:
ctx.configure(d, dumper_filter="~b foo")
assert d.filter
f = tflow.tflow(resp=True)
assert not d.match(f)
f.response.content = b"foo"
assert d.match(f)
ctx.configure(d, dumper_filter=None)
assert not d.filter
with pytest.raises(exceptions.OptionsError):
ctx.configure(d, dumper_filter="~~")
assert not d.filter