add the tests

This commit is contained in:
Yorick van Pelt 2015-04-03 00:27:57 +02:00
parent 0bf033d4d4
commit aed59bdeb8

View file

@ -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:])