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
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
|
# (c) 2005 Ian Bicking, Clark C. Evans and contributors
# This module is part of the Python Paste Project and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
import time
import random
import os
import tempfile
try:
# Python 3
from email.utils import parsedate_tz, mktime_tz
except ImportError:
# Python 2
from rfc822 import parsedate_tz, mktime_tz
import six
from paste import fileapp
from paste.fileapp import *
from paste.fixture import *
# NOTE(haypo): don't use string.letters because the order of lower and upper
# case letters changes when locale.setlocale() is called for the first time
LETTERS = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
def test_data():
harness = TestApp(DataApp(b'mycontent'))
res = harness.get("/")
assert 'application/octet-stream' == res.header('content-type')
assert '9' == res.header('content-length')
assert "<Response 200 OK 'mycontent'>" == repr(res)
harness.app.set_content(b"bingles")
assert "<Response 200 OK 'bingles'>" == repr(harness.get("/"))
def test_cache():
def build(*args,**kwargs):
app = DataApp(b"SomeContent")
app.cache_control(*args,**kwargs)
return TestApp(app).get("/")
res = build()
assert 'public' == res.header('cache-control')
assert not res.header('expires',None)
res = build(private=True)
assert 'private' == res.header('cache-control')
assert mktime_tz(parsedate_tz(res.header('expires'))) < time.time()
res = build(no_cache=True)
assert 'no-cache' == res.header('cache-control')
assert mktime_tz(parsedate_tz(res.header('expires'))) < time.time()
res = build(max_age=60,s_maxage=30)
assert 'public, max-age=60, s-maxage=30' == res.header('cache-control')
expires = mktime_tz(parsedate_tz(res.header('expires')))
assert expires > time.time()+58 and expires < time.time()+61
res = build(private=True, max_age=60, no_transform=True, no_store=True)
assert 'private, no-store, no-transform, max-age=60' == \
res.header('cache-control')
expires = mktime_tz(parsedate_tz(res.header('expires')))
assert mktime_tz(parsedate_tz(res.header('expires'))) < time.time()
def test_disposition():
def build(*args,**kwargs):
app = DataApp(b"SomeContent")
app.content_disposition(*args,**kwargs)
return TestApp(app).get("/")
res = build()
assert 'attachment' == res.header('content-disposition')
assert 'application/octet-stream' == res.header('content-type')
res = build(filename="bing.txt")
assert 'attachment; filename="bing.txt"' == \
res.header('content-disposition')
assert 'text/plain' == res.header('content-type')
res = build(inline=True)
assert 'inline' == res.header('content-disposition')
assert 'application/octet-stream' == res.header('content-type')
res = build(inline=True, filename="/some/path/bing.txt")
assert 'inline; filename="bing.txt"' == \
res.header('content-disposition')
assert 'text/plain' == res.header('content-type')
try:
res = build(inline=True,attachment=True)
except AssertionError:
pass
else:
assert False, "should be an exception"
def test_modified():
harness = TestApp(DataApp(b'mycontent'))
res = harness.get("/")
assert "<Response 200 OK 'mycontent'>" == repr(res)
last_modified = res.header('last-modified')
res = harness.get("/",headers={'if-modified-since': last_modified})
assert "<Response 304 Not Modified ''>" == repr(res)
res = harness.get("/",headers={'if-modified-since': last_modified + \
'; length=1506'})
assert "<Response 304 Not Modified ''>" == repr(res)
res = harness.get("/",status=400,
headers={'if-modified-since': 'garbage'})
assert 400 == res.status and b"ill-formed timestamp" in res.body
res = harness.get("/",status=400,
headers={'if-modified-since':
'Thu, 22 Dec 3030 01:01:01 GMT'})
assert 400 == res.status and b"check your system clock" in res.body
def test_file():
tempfile = "test_fileapp.%s.txt" % (random.random())
content = LETTERS * 20
if six.PY3:
content = content.encode('utf8')
with open(tempfile, "wb") as fp:
fp.write(content)
try:
app = fileapp.FileApp(tempfile)
res = TestApp(app).get("/")
assert len(content) == int(res.header('content-length'))
assert 'text/plain' == res.header('content-type')
assert content == res.body
assert content == app.content # this is cashed
lastmod = res.header('last-modified')
print("updating", tempfile)
file = open(tempfile,"a+")
file.write("0123456789")
file.close()
res = TestApp(app).get("/",headers={'Cache-Control': 'max-age=0'})
assert len(content)+10 == int(res.header('content-length'))
assert 'text/plain' == res.header('content-type')
assert content + b"0123456789" == res.body
assert app.content # we are still cached
file = open(tempfile,"a+")
file.write("X" * fileapp.CACHE_SIZE) # exceed the cashe size
file.write("YZ")
file.close()
res = TestApp(app).get("/",headers={'Cache-Control': 'max-age=0'})
newsize = fileapp.CACHE_SIZE + len(content)+12
assert newsize == int(res.header('content-length'))
assert newsize == len(res.body)
assert res.body.startswith(content) and res.body.endswith(b'XYZ')
assert not app.content # we are no longer cached
finally:
os.unlink(tempfile)
def test_dir():
tmpdir = tempfile.mkdtemp()
try:
tmpfile = os.path.join(tmpdir, 'file')
tmpsubdir = os.path.join(tmpdir, 'dir')
fp = open(tmpfile, 'w')
fp.write('abcd')
fp.close()
os.mkdir(tmpsubdir)
try:
app = fileapp.DirectoryApp(tmpdir)
for path in ['/', '', '//', '/..', '/.', '/../..']:
assert TestApp(app).get(path, status=403).status == 403, ValueError(path)
for path in ['/~', '/foo', '/dir', '/dir/']:
assert TestApp(app).get(path, status=404).status == 404, ValueError(path)
assert TestApp(app).get('/file').body == b'abcd'
finally:
os.remove(tmpfile)
os.rmdir(tmpsubdir)
finally:
os.rmdir(tmpdir)
def _excercize_range(build,content):
# full content request, but using ranges'
res = build("bytes=0-%d" % (len(content)-1))
assert res.header('accept-ranges') == 'bytes'
assert res.body == content
assert res.header('content-length') == str(len(content))
res = build("bytes=-%d" % (len(content)-1))
assert res.body == content
assert res.header('content-length') == str(len(content))
res = build("bytes=0-")
assert res.body == content
assert res.header('content-length') == str(len(content))
# partial content requests
res = build("bytes=0-9", status=206)
assert res.body == content[:10]
assert res.header('content-length') == '10'
res = build("bytes=%d-" % (len(content)-1), status=206)
assert res.body == b'Z'
assert res.header('content-length') == '1'
res = build("bytes=%d-%d" % (3,17), status=206)
assert res.body == content[3:18]
assert res.header('content-length') == '15'
def test_range():
content = LETTERS * 5
if six.PY3:
content = content.encode('utf8')
def build(range, status=206):
app = DataApp(content)
return TestApp(app).get("/",headers={'Range': range}, status=status)
_excercize_range(build,content)
build('bytes=0-%d' % (len(content)+1), 416)
def test_file_range():
tempfile = "test_fileapp.%s.txt" % (random.random())
content = LETTERS * (1+(fileapp.CACHE_SIZE // len(LETTERS)))
if six.PY3:
content = content.encode('utf8')
assert len(content) > fileapp.CACHE_SIZE
with open(tempfile, "wb") as fp:
fp.write(content)
try:
def build(range, status=206):
app = fileapp.FileApp(tempfile)
return TestApp(app).get("/",headers={'Range': range},
status=status)
_excercize_range(build,content)
for size in (13,len(LETTERS), len(LETTERS)-1):
fileapp.BLOCK_SIZE = size
_excercize_range(build,content)
finally:
os.unlink(tempfile)
def test_file_cache():
filename = os.path.join(os.path.dirname(__file__),
'urlparser_data', 'secured.txt')
app = TestApp(fileapp.FileApp(filename))
res = app.get('/')
etag = res.header('ETag')
last_mod = res.header('Last-Modified')
res = app.get('/', headers={'If-Modified-Since': last_mod},
status=304)
res = app.get('/', headers={'If-None-Match': etag},
status=304)
res = app.get('/', headers={'If-None-Match': 'asdf'},
status=200)
res = app.get('/', headers={'If-Modified-Since': 'Sat, 1 Jan 2005 12:00:00 GMT'},
status=200)
res = app.get('/', headers={'If-Modified-Since': last_mod + '; length=100'},
status=304)
res = app.get('/', headers={'If-Modified-Since': 'invalid date'},
status=400)
def test_methods():
filename = os.path.join(os.path.dirname(__file__),
'urlparser_data', 'secured.txt')
app = TestApp(fileapp.FileApp(filename))
get_res = app.get('')
res = app.get('', extra_environ={'REQUEST_METHOD': 'HEAD'})
assert res.headers == get_res.headers
assert not res.body
app.post('', status=405) # Method Not Allowed
|
