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
116
117
118
119
120
121
122
123
|
import cStringIO
import tempfile
import os
import time
import shutil
from contextlib import contextmanager
from netlib import tcp, utils, odict, http
def treader(bytes):
"""
Construct a tcp.Read object from bytes.
"""
fp = cStringIO.StringIO(bytes)
return tcp.Reader(fp)
@contextmanager
def tmpdir(*args, **kwargs):
orig_workdir = os.getcwd()
temp_workdir = tempfile.mkdtemp(*args, **kwargs)
os.chdir(temp_workdir)
yield temp_workdir
os.chdir(orig_workdir)
shutil.rmtree(temp_workdir)
def raises(exc, obj, *args, **kwargs):
"""
Assert that a callable raises a specified exception.
:exc An exception class or a string. If a class, assert that an
exception of this type is raised. If a string, assert that the string
occurs in the string representation of the exception, based on a
case-insenstivie match.
:obj A callable object.
:args Arguments to be passsed to the callable.
:kwargs Arguments to be passed to the callable.
"""
try:
ret = obj(*args, **kwargs)
except Exception as v:
if isinstance(exc, basestring):
if exc.lower() in str(v).lower():
return
else:
raise AssertionError(
"Expected %s, but caught %s" % (
repr(str(exc)), v
)
)
else:
if isinstance(v, exc):
return
else:
raise AssertionError(
"Expected %s, but caught %s %s" % (
exc.__name__, v.__class__.__name__, str(v)
)
)
raise AssertionError("No exception raised. Return value: {}".format(ret))
test_data = utils.Data(__name__)
def treq(content="content", scheme="http", host="address", port=22):
"""
@return: libmproxy.protocol.http.HTTPRequest
"""
headers = odict.ODictCaseless()
headers["header"] = ["qvalue"]
req = http.Request(
"relative",
"GET",
scheme,
host,
port,
"/path",
(1, 1),
headers,
content,
None,
None,
)
return req
def treq_absolute(content="content"):
"""
@return: libmproxy.protocol.http.HTTPRequest
"""
r = treq(content)
r.form_in = r.form_out = "absolute"
r.host = "address"
r.port = 22
r.scheme = "http"
return r
def tresp(content="message"):
"""
@return: libmproxy.protocol.http.HTTPResponse
"""
headers = odict.ODictCaseless()
headers["header_response"] = ["svalue"]
resp = http.semantics.Response(
(1, 1),
200,
"OK",
headers,
content,
timestamp_start=time.time(),
timestamp_end=time.time(),
)
return resp
|
