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
108
109
110
111
112
113
114
115
|
from libmproxy import console, proxy, filt, flow
import utils
import libpry
class uFlow(libpry.AutoTree):
def test_run_script(self):
f = utils.tflow()
f.response = utils.tresp()
f.request = f.response.request
f, se = f.run_script("scripts/a")
assert "DEBUG" == se.strip()
assert f.request.host == "TESTOK"
def test_run_script_err(self):
f = utils.tflow()
f.response = utils.tresp()
f.request = f.response.request
libpry.raises("returned error", f.run_script,"scripts/err_return")
libpry.raises("invalid response", f.run_script,"scripts/err_data")
libpry.raises("no such file", f.run_script,"nonexistent")
libpry.raises("permission denied", f.run_script,"scripts/nonexecutable")
def test_match(self):
f = utils.tflow()
f.response = utils.tresp()
f.request = f.response.request
assert not f.match(filt.parse("~b test"))
def test_backup(self):
f = utils.tflow()
assert not f.modified()
f.backup()
assert f.modified()
f.revert()
def test_getset_state(self):
f = utils.tflow()
state = f.get_state()
assert f == flow.Flow.from_state(state)
f.response = utils.tresp()
f.request = f.response.request
state = f.get_state()
assert f == flow.Flow.from_state(state)
def test_simple(self):
f = utils.tflow()
assert console.format_flow(f, True)
assert console.format_flow(f, False)
f.request = utils.treq()
assert console.format_flow(f, True)
assert console.format_flow(f, False)
f.response = utils.tresp()
f.response.headers["content-type"] = ["text/html"]
assert console.format_flow(f, True)
assert console.format_flow(f, False)
f.response.code = 404
assert console.format_flow(f, True)
assert console.format_flow(f, False)
f.focus = True
assert console.format_flow(f, True)
assert console.format_flow(f, False)
f.connection = flow.ReplayConnection()
assert console.format_flow(f, True)
assert console.format_flow(f, False)
f.response = None
assert console.format_flow(f, True)
assert console.format_flow(f, False)
f.error = proxy.Error(200, "test")
assert console.format_flow(f, True)
assert console.format_flow(f, False)
def test_kill(self):
f = utils.tflow()
f.request = utils.treq()
f.intercept()
assert not f.request.acked
f.kill()
assert f.request.acked
f.intercept()
f.response = utils.tresp()
f.request = f.response.request
f.request.ack()
assert not f.response.acked
f.kill()
assert f.response.acked
def test_accept_intercept(self):
f = utils.tflow()
f.request = utils.treq()
f.intercept()
assert not f.request.acked
f.accept_intercept()
assert f.request.acked
f.response = utils.tresp()
f.request = f.response.request
f.intercept()
f.request.ack()
assert not f.response.acked
f.accept_intercept()
assert f.response.acked
def test_serialization(self):
f = flow.Flow(None)
f.request = utils.treq()
tests = [
uFlow()
]
|
