summaryrefslogtreecommitdiff
path: root/project1/proj1_s4498062/webtests.py
diff options
context:
space:
mode:
Diffstat (limited to 'project1/proj1_s4498062/webtests.py')
-rw-r--r--project1/proj1_s4498062/webtests.py374
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()