-
-
Notifications
You must be signed in to change notification settings - Fork 3.9k
/
test_modifybody.py
107 lines (94 loc) · 3.56 KB
/
test_modifybody.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
import pytest
from mitmproxy.addons import modifybody
from mitmproxy.addons import proxyserver
from mitmproxy.test import taddons
from mitmproxy.test import tflow
from mitmproxy.test.tutils import tresp
class TestModifyBody:
def test_configure(self):
mb = modifybody.ModifyBody()
with taddons.context(mb) as tctx:
tctx.configure(mb, modify_body=["one/two/three"])
with pytest.raises(Exception, match="Cannot parse modify_body"):
tctx.configure(mb, modify_body=["/"])
def test_warn_conflict(self, caplog):
caplog.set_level("DEBUG")
mb = modifybody.ModifyBody()
with taddons.context(mb, proxyserver.Proxyserver()) as tctx:
tctx.configure(mb, stream_large_bodies="3m", modify_body=["one/two/three"])
assert "Streamed bodies will not be modified" in caplog.text
def test_simple(self):
mb = modifybody.ModifyBody()
with taddons.context(mb) as tctx:
tctx.configure(
mb,
modify_body=[
"/~q/foo/bar",
"/~s/foo/bar",
],
)
f = tflow.tflow()
f.request.content = b"foo"
mb.request(f)
assert f.request.content == b"bar"
f = tflow.tflow(resp=True)
f.response.content = b"foo"
mb.response(f)
assert f.response.content == b"bar"
@pytest.mark.parametrize("take", [True, False])
def test_taken(self, take):
mb = modifybody.ModifyBody()
with taddons.context(mb) as tctx:
tctx.configure(mb, modify_body=["/foo/bar"])
f = tflow.tflow()
f.request.content = b"foo"
if take:
f.response = tresp()
mb.request(f)
assert (f.request.content == b"bar") ^ take
f = tflow.tflow(resp=True)
f.response.content = b"foo"
if take:
f.kill()
mb.response(f)
assert (f.response.content == b"bar") ^ take
def test_order(self):
mb = modifybody.ModifyBody()
with taddons.context(mb) as tctx:
tctx.configure(
mb,
modify_body=[
"/foo/bar",
"/bar/baz",
"/foo/oh noes!",
"/bar/oh noes!",
],
)
f = tflow.tflow()
f.request.content = b"foo"
mb.request(f)
assert f.request.content == b"baz"
class TestModifyBodyFile:
def test_simple(self, tmpdir):
mb = modifybody.ModifyBody()
with taddons.context(mb) as tctx:
tmpfile = tmpdir.join("replacement")
tmpfile.write("bar")
tctx.configure(mb, modify_body=["/~q/foo/@" + str(tmpfile)])
f = tflow.tflow()
f.request.content = b"foo"
mb.request(f)
assert f.request.content == b"bar"
async def test_nonexistent(self, tmpdir, caplog):
mb = modifybody.ModifyBody()
with taddons.context(mb) as tctx:
with pytest.raises(Exception, match="Invalid file path"):
tctx.configure(mb, modify_body=["/~q/foo/@nonexistent"])
tmpfile = tmpdir.join("replacement")
tmpfile.write("bar")
tctx.configure(mb, modify_body=["/~q/foo/@" + str(tmpfile)])
tmpfile.remove()
f = tflow.tflow()
f.request.content = b"foo"
mb.request(f)
assert "Could not read" in caplog.text