diff options
Diffstat (limited to 'project1/proj1_s4498062/webtests.py')
-rw-r--r-- | project1/proj1_s4498062/webtests.py | 374 |
1 files changed, 187 insertions, 187 deletions
diff --git a/project1/proj1_s4498062/webtests.py b/project1/proj1_s4498062/webtests.py index d047526..f2f6fbd 100644 --- a/project1/proj1_s4498062/webtests.py +++ b/project1/proj1_s4498062/webtests.py @@ -1,187 +1,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()
+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() |