diff --git a/webtests/tests.py b/webtests/tests.py index 83568ca..f672afd 100644 --- a/webtests/tests.py +++ b/webtests/tests.py @@ -1,57 +1,175 @@ import sys import unittest import socket +from time import sleep -def http_req(url, headers = []): - h = "".join([hdr + '\r\n' for hdr in ["Connection: close"] + headers]) + +# basic stuff for http testing + +# make simple http requests +def http_req(url, headers = [], close = True): + if close: + start = ["Connection: close"] + else: + start = [] + h = "".join([hdr + '\r\n' for hdr in start + headers]) return "GET %s HTTP/1.1\r\n%s\r\n" % (url, h) +# wrapper around sockets for blocking http parsing +class http_client(object): + def __init__(self, addr): + self.s = socket.create_connection(addr) + self.buf = "" + def closed(self): + # probably errors when there is data available + return len(self.s.recv(1, socket.MSG_DONTWAIT)) == 0 + def buf_empty(self): + return len(self.buf) == 0 + def recvall(self): + """call recv until the connection is closed""" + while 1: + s = self.s.recv(8192) + if not s: + b = self.buf + self.buf = "" + return b + self.buf += s + def recv_one(self): + """call recv until there is at least one parsed response""" + while 1: + s = self.s.recv(8192) + if not s: + (code, headers, content, self.buf) = parse_resp(self.buf) + return (code, headers, content, self.buf) + self.buf += s + if '\r\n\r\n' in self.buf: + resp = parse_resp(self.buf) + if resp: + (code, headers, content, self.buf) = resp + return (code, headers, content, self.buf) + def http_req(self, url, *args, **kwargs): + self.s.send(http_req(url, *args, **kwargs)) +def parse_resp(resp): + assert resp.startswith("HTTP/1.1 ") + code = int(resp[9:12]) + hdr, data = resp.split('\r\n\r\n', 1) + header_strs = hdr.split('\r\n')[1:] + headers = {h.split(': ', 1)[0]: h.split(': ', 1)[1] for h in header_strs} + # Content-Length isn't mandatory + # but my server always sends it, so I can use it for tests + length = int(headers['Content-Length']) + content = data[:length] + if len(content) != length: return None + rest = data[length:] + return (code, headers, content, rest) -def recvall(sock): - """call recv until the connection is closed""" - d = "" - while 1: - s = sock.recv(8192) - if not s: return d - d += s +def req_url(url, headers = []): + cli = http_client(('localhost', port)) + cli.http_req(url, headers) + return cli.recvall() +def get_url_resp(url, headers = []): + resp = req_url(url, headers) + (code, headers, content, rest) = parse_resp(resp) + assert len(rest) == 0 + return (code, headers, content) # GET for an existing single resource class GETSingles(unittest.TestCase): # in YOUR local area NOW def test_single_existing(self): - s = socket.create_connection(('localhost', 8080)) - s.send(http_req('/')) - resp = recvall(s) - assert resp.startswith("HTTP/1.1 200 OK\r\n") - # TODO: check content + (code, _, data) = get_url_resp('/test.txt') + assert code == 200 + assert data == "test\n" # GET for a single resource that doesn't exist def test_single_notexisting(self): - s = socket.create_connection(('localhost', 8080)) - s.send(http_req('/this_is_not_the_path/you/are/looking/for')) - resp = recvall(s) - assert resp.startswith("HTTP/1.1 404 Not Found\r\n") + (code, _, data) = get_url_resp('/404') + assert code == 404 + # GET for a directory with an existing index.html file + def test_index_html(self): + (code, _, data) = get_url_resp('/index/') + assert code == 200 + # GET for a directory with non-existing index.html file + def test_no_index_html(self): + (code, _, data) = get_url_resp('/no_index/') + assert code == 404 +class GETMultis(unittest.TestCase): + # GET for an existing single resource followed by a GET for that same resource, + # with caching utilized on the client/tester side + def test_cache(self): + (code, headers, data) = get_url_resp('/test_img.png') + assert code == 200 + (code , headers, data) = get_url_resp('/test_img.png', ["If-None-Match: " + headers['ETag']]) + assert len(data) == 0 + assert code == 304 + # multiple GETs over the same (persistent) connection with the last GET + # prompting closing the connection, the connection should be closed + def test_persistent(self): + cli = http_client(('localhost', port)) + cli.http_req('/test.txt', close = False) + (code, headers, data, rest) = cli.recv_one() + assert code == 200 + assert data == "test\n" + cli.http_req('/test.txt', close = True) + (code, headers, data, rest) = cli.recv_one() + assert code == 200 + assert data == "test\n" + assert cli.buf_empty() + assert cli.closed() + # multiple GETs over the same (persistent) connection, followed by a wait during + # which the connection times out, the connection should be closed + def test_persistent_timeout(self): + cli = http_client(('localhost', port)) + cli.http_req('/test.txt', close = False) + (code, headers, data, rest) = cli.recv_one() + assert code == 200 + assert data == "test\n" -# GET for an existing single resource followed by a GET for that same resource, -# with caching utilized on the client/tester side -# GET for a directory with an existing index.html file -# GET for a directory with non-existing index.html file + cli.http_req('/test.txt', close = False) + (code, headers, data, rest) = cli.recv_one() + assert code == 200 + assert data == "test\n" + assert cli.buf_empty() + # now wait for timeout (10 secs + safety margin) + sleep(10.5) + assert cli.closed() + # multiple GETs, some of which are parallel (think of the situation when your + # browser is fetching a composite resource), the responses should be sent in an + # orderly fashion + def test_parallel(self): + # test multiple connections at the same time + cli1 = http_client(('localhost', port)) + cli2 = http_client(('localhost', port)) + # test pipelining too + cli1.http_req('/test.txt', close = False) + cli1.http_req('/test.txt', close = True) -# multiple GETs over the same (persistent) connection with the last GET prompting closing the con- -# nection, the connection should be closed -# multiple GETs over the same (persistent) connection, followed by a wait during which the connection -# times out, the connection should be closed -# multiple GETs, some of which are parallel (think of the situation when your browser is fetching a -# composite resource), the responses should be sent in an orderly fashion + cli2.http_req('/test.txt', close = True) -#In each case, you should test the response code after every step, as well as the content (if any is expected -#or if none is expected). + (code, headers, data, _) = cli1.recv_one() + assert code == 200 + assert data == "test\n" + + (code, headers, data, _) = cli2.recv_one() + assert code == 200 + assert data == "test\n" + assert cli2.buf_empty() + assert cli2.closed() + + (code, headers, data, rest) = cli1.recv_one() + assert code == 200 + assert data == "test\n" + assert cli1.buf_empty() + assert cli1.closed() + +#In each case, you should test the response code after every step, as well as +#the content (if any is expected or if none is expected). if __name__ == '__main__': try: + global port port = int(sys.argv[1]) except (ValueError, IndexError): print "invalid port given" else: - # raise NotImplementedError() unittest.main(argv=sys.argv[1:])