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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
|
"""Tests for mechanize._response.seek_wrapper and friends."""
import copy
import cStringIO
from unittest import TestCase
class TestUnSeekable:
def __init__(self, text):
self._file = cStringIO.StringIO(text)
self.log = []
def tell(self): return self._file.tell()
def seek(self, offset, whence=0): assert False
def read(self, size=-1):
self.log.append(("read", size))
return self._file.read(size)
def readline(self, size=-1):
self.log.append(("readline", size))
return self._file.readline(size)
def readlines(self, sizehint=-1):
self.log.append(("readlines", sizehint))
return self._file.readlines(sizehint)
class TestUnSeekableResponse(TestUnSeekable):
def __init__(self, text, headers):
TestUnSeekable.__init__(self, text)
self.code = 200
self.msg = "OK"
self.headers = headers
self.url = "http://example.com/"
def geturl(self):
return self.url
def info(self):
return self.headers
def close(self):
pass
class SeekableTests(TestCase):
text = """\
The quick brown fox
jumps over the lazy
dog.
"""
text_lines = map(lambda l: l+"\n", text.split("\n")[:-1])
def testSeekable(self):
from mechanize._response import seek_wrapper
text = self.text
for ii in range(1, 6):
fh = TestUnSeekable(text)
sfh = seek_wrapper(fh)
test = getattr(self, "_test%d" % ii)
test(sfh)
# copies have independent seek positions
fh = TestUnSeekable(text)
sfh = seek_wrapper(fh)
self._testCopy(sfh)
def _testCopy(self, sfh):
sfh2 = copy.copy(sfh)
sfh.read(10)
text = self.text
self.assertEqual(sfh2.read(10), text[:10])
sfh2.seek(5)
self.assertEqual(sfh.read(10), text[10:20])
self.assertEqual(sfh2.read(10), text[5:15])
sfh.seek(0)
sfh2.seek(0)
return sfh2
def _test1(self, sfh):
text = self.text
text_lines = self.text_lines
assert sfh.read(10) == text[:10] # calls fh.read
assert sfh.log[-1] == ("read", 10) # .log delegated to fh
sfh.seek(0) # doesn't call fh.seek
assert sfh.read(10) == text[:10] # doesn't call fh.read
assert len(sfh.log) == 1
sfh.seek(0)
assert sfh.read(5) == text[:5] # read only part of cached data
assert len(sfh.log) == 1
sfh.seek(0)
assert sfh.read(25) == text[:25] # calls fh.read
assert sfh.log[1] == ("read", 15)
lines = []
sfh.seek(-1, 1)
while 1:
l = sfh.readline()
if l == "": break
lines.append(l)
assert lines == ["s over the lazy\n"]+text_lines[2:]
assert sfh.log[2:] == [("readline", -1)]*5
sfh.seek(0)
lines = []
while 1:
l = sfh.readline()
if l == "": break
lines.append(l)
assert lines == text_lines
def _test2(self, sfh):
text = self.text
sfh.read(5)
sfh.seek(0)
assert sfh.read() == text
assert sfh.read() == ""
sfh.seek(0)
assert sfh.read() == text
sfh.seek(0)
assert sfh.readline(5) == "The q"
assert sfh.read() == text[5:]
sfh.seek(0)
assert sfh.readline(5) == "The q"
assert sfh.readline() == "uick brown fox\n"
def _test3(self, sfh):
text_lines = self.text_lines
sfh.read(25)
sfh.seek(-1, 1)
self.assertEqual(sfh.readlines(), ["s over the lazy\n"]+text_lines[2:])
sfh.seek(0)
assert sfh.readlines() == text_lines
def _test4(self, sfh):
text_lines = self.text_lines
count = 0
limit = 10
while count < limit:
if count == 5:
self.assertRaises(StopIteration, sfh.next)
break
else:
sfh.next() == text_lines[count]
count = count + 1
else:
assert False, "StopIteration not raised"
def _test5(self, sfh):
text = self.text
sfh.read(10)
sfh.seek(5)
self.assert_(sfh.invariant())
sfh.seek(0, 2)
self.assert_(sfh.invariant())
sfh.seek(0)
self.assertEqual(sfh.read(), text)
def testResponseSeekWrapper(self):
from mechanize import response_seek_wrapper
hdrs = {"Content-type": "text/html"}
r = TestUnSeekableResponse(self.text, hdrs)
rsw = response_seek_wrapper(r)
rsw2 = self._testCopy(rsw)
self.assert_(rsw is not rsw2)
self.assertEqual(rsw.info(), rsw2.info())
self.assert_(rsw.info() is not rsw2.info())
# should be able to close already-closed object
rsw2.close()
rsw2.close()
def testSetResponseData(self):
from mechanize import response_seek_wrapper
r = TestUnSeekableResponse(self.text, {'blah': 'yawn'})
rsw = response_seek_wrapper(r)
rsw.set_data("""\
A Seeming somwhat more than View;
That doth instruct the Mind
In Things that ly behind,
""")
self.assertEqual(rsw.read(9), "A Seeming")
self.assertEqual(rsw.read(13), " somwhat more")
rsw.seek(0)
self.assertEqual(rsw.read(9), "A Seeming")
self.assertEqual(rsw.readline(), " somwhat more than View;\n")
rsw.seek(0)
self.assertEqual(rsw.readline(), "A Seeming somwhat more than View;\n")
rsw.seek(-1, 1)
self.assertEqual(rsw.read(7), "\n That")
r = TestUnSeekableResponse(self.text, {'blah': 'yawn'})
rsw = response_seek_wrapper(r)
rsw.set_data(self.text)
self._test2(rsw)
rsw.seek(0)
self._test4(rsw)
def testGetResponseData(self):
from mechanize import response_seek_wrapper
r = TestUnSeekableResponse(self.text, {'blah': 'yawn'})
rsw = response_seek_wrapper(r)
self.assertEqual(rsw.get_data(), self.text)
self._test2(rsw)
rsw.seek(0)
self._test4(rsw)
if __name__ == "__main__":
import unittest
unittest.main()
|
