Tests for POST

This commit is contained in:
Kovid Goyal 2015-05-19 21:34:26 +05:30
parent cdf9ab1c0e
commit 3c06608acd

View File

@ -6,7 +6,7 @@ from __future__ import (unicode_literals, division, absolute_import,
__license__ = 'GPL v3' __license__ = 'GPL v3'
__copyright__ = '2015, Kovid Goyal <kovid at kovidgoyal.net>' __copyright__ = '2015, Kovid Goyal <kovid at kovidgoyal.net>'
import textwrap, httplib import textwrap, httplib, socket
from io import BytesIO from io import BytesIO
from calibre.srv.tests.base import BaseTest, TestServer from calibre.srv.tests.base import BaseTest, TestServer
@ -80,11 +80,24 @@ class TestHTTP(BaseTest):
self.ae(r.status, httplib.NOT_FOUND) self.ae(r.status, httplib.NOT_FOUND)
self.ae(r.read(), 'Requested resource not found') self.ae(r.read(), 'Requested resource not found')
server.change_handler(lambda conn:conn.path[1]) server.change_handler(lambda conn:conn.path[1] + conn.input_reader.read().decode('ascii'))
# Test simple GET # Test simple GET
conn.request('GET', '/test') conn.request('GET', '/test')
self.ae(conn.getresponse().read(), 'test') self.ae(conn.getresponse().read(), 'test')
# Test POST with simple body
conn.request('POST', '/test', 'body')
r = conn.getresponse()
self.ae(r.status, httplib.CREATED)
self.ae(r.read(), 'testbody')
# Test POST with chunked transfer encoding
conn.request('POST', '/test', headers={'Transfer-Encoding': 'chunked'})
conn.send(b'4\r\nbody\r\n0\r\n\r\n')
r = conn.getresponse()
self.ae(r.status, httplib.CREATED)
self.ae(r.read(), 'testbody')
# Test pipelining # Test pipelining
responses = [] responses = []
for i in xrange(10): for i in xrange(10):
@ -97,4 +110,12 @@ class TestHTTP(BaseTest):
self.ae(r.read(), ('%d' % i).encode('ascii')) self.ae(r.read(), ('%d' % i).encode('ascii'))
conn._HTTPConnection__state = httplib._CS_IDLE conn._HTTPConnection__state = httplib._CS_IDLE
# Test closing
conn.request('GET', '/close', headers={'Connection':'close'})
r = conn.getresponse()
self.ae(r.status, 200), self.ae(r.read(), 'close')
conn.request('HEAD', '/close')
with self.assertRaises(socket.error):
conn.sock.send(b'xxx')
# }}} # }}}