Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2025-02-01 15:55:28 +00:00
Code Issues Projects Releases Wiki Activity

intercept: tests++

Browse Source
This commit is contained in:
Maximilian Hils 2020-08-27 11:37:35 +02:00
parent 28f04c93fd
commit 327e933faf
2 changed files with 24 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
mitmproxy/addons/intercept.py
View File

@ -30,7 +30,7 @@ class Intercept:
ctx.options.intercept_active = False ctx.options.intercept_active = False
def should_intercept(self, f: flow.Flow) -> bool: def should_intercept(self, f: flow.Flow) -> bool:
return ( return bool(
ctx.options.intercept_active ctx.options.intercept_active
and self.filt and self.filt
and self.filt(f) and self.filt(f)
@ -39,6 +39,7 @@ class Intercept:
def process_flow(self, f: flow.Flow) -> None: def process_flow(self, f: flow.Flow) -> None:
if self.should_intercept(f): if self.should_intercept(f):
assert f.reply
if f.reply.state != "start": if f.reply.state != "start":
return ctx.log.debug("Cannot intercept request that is already taken by another addon.") return ctx.log.debug("Cannot intercept request that is already taken by another addon.")
f.intercept() f.intercept()

25
test/mitmproxy/addons/test_intercept.py
View File

@ -43,12 +43,31 @@ def test_simple():
tctx.cycle(r, f) tctx.cycle(r, f)
assert f.intercepted assert f.intercepted
def test_tcp():
r = intercept.Intercept()
with taddons.context(r) as tctx:
tctx.configure(r, intercept="~tcp")
f = tflow.ttcpflow()
tctx.cycle(r, f)
assert f.intercepted
tctx.configure(r, intercept_active=False) tctx.configure(r, intercept_active=False)
f = tflow.ttcpflow() f = tflow.ttcpflow()
tctx.cycle(r, f) tctx.cycle(r, f)
assert not f.intercepted assert not f.intercepted
tctx.configure(r, intercept_active=True)
f = tflow.ttcpflow() def test_already_taken():
tctx.cycle(r, f) r = intercept.Intercept()
with taddons.context(r) as tctx:
tctx.configure(r, intercept="~q")
f = tflow.tflow()
tctx.invoke(r, "request", f)
assert f.intercepted assert f.intercepted
f = tflow.tflow()
f.reply.take()
tctx.invoke(r, "request", f)
assert not f.intercepted