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
|
import os.path
import unittest
import socket
import sys
import time
from webhttp.config import config
import webhttp.message
import webhttp.parser
portnr = 8001
class TestGetRequests(unittest.TestCase):
"""Test cases for GET requests"""
def setUp(self):
"""Prepare for testing"""
self.client_skt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client_skt.connect(("localhost", portnr))
self.parser = webhttp.parser.ResponseParser()
self.default_headers = [
('Host', 'localhost:%d' % portnr),
('Connection', 'close')
]
self.default_headers_pers = [
('Host', 'localhost:%d' % portnr),
('Connection', 'keep-alive')
]
config().read(os.path.expanduser('~/.webpy.ini'))
def tearDown(self):
"""Clean up after testing"""
try:
self.client_skt.shutdown(socket.SHUT_RDWR)
self.client_skt.close()
except:
pass
def refresh_socket(self):
try:
self.client_skt.shutdown(socket.SHUT_RDWR)
self.client_skt.close()
except:
pass
self.client_skt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client_skt.connect(("localhost", portnr))
def request(self, method, uri, headers, persistent=False):
request = webhttp.message.Request()
request.method = method
request.uri = uri
for name, value in headers:
request.set_header(name, value)
sent = self.client_skt.send(str(request))
if sent == 0:
return None
message = self.client_skt.recv(1024)
if not persistent:
self.refresh_socket()
if message == '':
return None
else:
response = self.parser.parse_response(message)
return response
def test_existing_file(self):
"""GET for a single resource that exists"""
response = self.request(
'GET', '/test/index.html', self.default_headers)
self.assertEqual(response.code, 200)
self.assertTrue(response.body)
def test_nonexistant_file(self):
"""GET for a single resource that does not exist"""
response = self.request(
'GET', '/test/nonexistant.html', self.default_headers)
self.assertEqual(response.code, 404)
self.assertTrue(response.body)
def test_caching(self):
"""GET for an existing single resource followed by a GET for that same
resource with caching utilized on the client/tester side
"""
response = self.request('GET', '/test', self.default_headers)
response = self.request('GET', '/test', self.default_headers + \
[('If-None-Match', 'invalid-etag')])
self.assertEqual(response.code, 200, 'If-None-Match returns 200')
response = self.request('GET', '/test', self.default_headers + \
[('If-None-Match', response.get_header('ETag'))])
self.assertEqual(response.code, 304, 'If-None-Match returns 304')
response = self.request('GET', '/test', self.default_headers + \
[('If-Match', response.get_header('ETag'))])
self.assertEqual(response.code, 200, 'If-Match returns 200')
response = self.request('GET', '/test', self.default_headers + \
[('If-Match', 'invalid-etag')])
self.assertEqual(response.code, 304, 'If-Match returns 304')
def test_extisting_index_file(self):
"""GET for a directory with an existing index.html file"""
self.assertEqual(self.request('GET', '/test', self.default_headers),
self.request('GET', '/test/index.html', self.default_headers))
def test_nonexistant_index_file(self):
"""GET for a directory with a non-existant index.html file"""
response = self.request('GET', '/test/no-index', self.default_headers)
self.assertEqual(response.code, 403)
self.assertTrue(response.body)
def test_error_page(self):
r1 = self.request('GET', '/test/nonexistant', self.default_headers)
r2 = self.request('GET', config('error404'), self.default_headers)
self.assertEqual(r1.body, r2.body)
def test_persistent_close(self):
"""Multiple GETs over the same (persistent) connection with the last
GET prompting closing the connection, the connection should be closed.
"""
try:
self.request('GET', '/test', self.default_headers_pers, True)
self.request('GET', '/test', self.default_headers_pers, True)
self.request('GET', '/test', self.default_headers_pers, True)
except socket.error:
self.fail('Persistent connection closed prematurely')
self.request('GET', '/test', self.default_headers, True)
self.assertIsNone(self.request('GET', '/test',
self.default_headers_pers, True))
self.refresh_socket()
def test_persistent_timeout(self):
"""Multiple GETs over the same (persistent) connection, followed by a
wait during which the connection times out, the connection should be
closed.
"""
self.request('GET', '/test', self.default_headers_pers, True)
time.sleep(20)
self.assertIsNone(self.request('GET', '/test',
self.default_headers_pers, True))
self.refresh_socket()
def test_encoding(self):
"""GET which requests an existing resource using gzip encodign, which
is accepted by the server.
"""
r1 = self.request('GET', '/test', self.default_headers + \
[('Accept-Encoding', 'gzip;q=1, identity;q=0')])
self.assertEqual(r1.get_header('Content-Encoding'), 'gzip')
r1.decompress()
r2 = self.request('GET', '/test', self.default_headers + \
[('Accept-Encoding', '')])
self.assertEqual(r1.body, r2.body)
def test_doubledot(self):
response = self.request('GET', '/../test', self.default_headers)
self.assertEquals(response.code, 403)
if __name__ == "__main__":
# Parse command line arguments
import argparse
parser = argparse.ArgumentParser(description="HTTP Tests")
parser.add_argument("-p", "--port", type=int, default=8001)
# Arguments for the unittest framework
parser.add_argument('unittest_args', nargs='*')
args = parser.parse_args()
portnr = args.port
# Only pass the unittest arguments to unittest
sys.argv[1:] = args.unittest_args
# Start test suite
unittest.main()
|
