############################################################################## # # Copyright (c) 2003 Zope Corporation and Contributors. # All Rights Reserved. # # This software is subject to the provisions of the Zope Public License, # Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution. # THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED # WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED # WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS # FOR A PARTICULAR PURPOSE. # ############################################################################## """Test general ZServer machinery.""" from ZServer.HTTPResponse import ZServerHTTPResponse from ZServer.FTPResponse import FTPResponse from ZServer.PCGIServer import PCGIResponse from ZServer.FCGIServer import FCGIResponse from ZPublisher.Iterators import IStreamIterator import unittest from cStringIO import StringIO class ZServerResponseTestCase(unittest.TestCase): """Test ZServer response objects.""" def test_http_response_write_unicode(self): response = ZServerHTTPResponse() self.assertRaises(TypeError, response.write, u'bad') def test_ftp_response_write_unicode(self): response = FTPResponse() self.assertRaises(TypeError, response.write, u'bad') def test_pcgi_response_write_unicode(self): response = PCGIResponse() self.assertRaises(TypeError, response.write, u'bad') def test_fcgi_response_write_unicode(self): response = FCGIResponse() self.assertRaises(TypeError, response.write, u'bad') def test_setBodyIterator(self): channel = DummyChannel() one = ZServerHTTPResponse(stdout=channel) one.setHeader('content-length', 5) one.setBody(test_streamiterator()) one.outputBody() all = channel.all() lines = all.split('\r\n') self.assertEqual(lines[-2], '') # end of headers self.assertEqual(lines[-1], 'hello') # payload def test_setBodyIteratorFailsWithoutContentLength(self): one = ZServerHTTPResponse(stdout=DummyChannel()) self.assertRaises(AssertionError, one.setBody, test_streamiterator()) class DummyChannel: def __init__(self): self.out = StringIO() def all(self): self.out.seek(0) return self.out.read() def read(self): pass def write(self, data, len=None): try: if isinstance(data, str): self.out.write(data) return except TypeError: pass while 1: s = data.more() if not s: break self.out.write(s) class test_streamiterator: __implements__ = IStreamIterator data = "hello" done = 0 def next(self): if not self.done: self.done = 1 return self.data raise StopIteration class ZServerHTTPResponseTestCase(unittest.TestCase): """Test ZServer HTTPResponse object""" def _makeOne(self): return ZServerHTTPResponse() def testToString(self): response = self._makeOne() response.headers = { 'content-type': 'text/plain', 'all-lower-case': 'foo', 'Title-Cased': 'bar', 'mixed-CasED': 'spam', 'multilined': 'eggs\n\tham'} response.accumulated_headers = 'foo-bar: bar\n\tbaz\nFoo-bar: monty\n' response.cookies = dict(foo=dict(value='bar')) response.body = 'A body\nwith multiple lines\n' result = str(response) headers, body = result.rsplit('\r\n\r\n') self.assertEqual(body, response.body) self.assertTrue(headers.startswith('HTTP/1.0 200 OK\r\n')) # 15 header lines all delimited by \r\n self.assertEqual( ['\n' in line for line in headers.split('\r\n')], 15 * [False]) self.assertTrue('Multilined: eggs\r\n\tham\r\n' in headers) self.assertTrue('Foo-Bar: bar\r\n\tbaz\r\n' in headers) def _assertResponsesAreEqual(self, got, expected): got = got.split('\r\n') # Sort the headers into alphabetical order. headers = got[1:got.index('')] headers.sort() got[1:len(headers)+1] = headers # Compare line by line. for n in range(len(expected)): if expected[n].endswith('...'): m = len(expected[n]) - 3 self.assertEqual(got[n][:m], expected[n][:m]) else: self.assertEqual(got[n], expected[n]) self.assertEqual(len(got), len(expected)) def test_emptyResponse(self): # Empty repsonses have no Content-Length. response = self._makeOne() self._assertResponsesAreEqual(str(response), ('HTTP/1.0 204 No Content', 'Connection: close', 'Date: ...', 'Server: ...', '', '')) def test_304(self): # Now we set the status to 304. 304 responses, according to RFC 2616, # should not have a content-length header. __str__ should not add it # back in if it is missing. response = self._makeOne() response.setStatus(304) self._assertResponsesAreEqual(str(response), ('HTTP/1.0 304 Not Modified', 'Connection: close', 'Date: ...', 'Server: ...', '', '')) def test_304ContentLength(self): # __str__ should strip out Content-Length response = self._makeOne() response.setStatus(304) response.setHeader('content-length', '123') self._assertResponsesAreEqual(str(response), ('HTTP/1.0 304 Not Modified', 'Connection: close', 'Date: ...', 'Server: ...', '', '')) def test_304ContentType(self): # __str__ should strip out Content-Type response = self._makeOne() response.setStatus(304) response.setHeader('content-type', 'text/plain') self._assertResponsesAreEqual(str(response), ('HTTP/1.0 304 Not Modified', 'Connection: close', 'Date: ...', 'Server: ...', '', '')) def test_304ExplicitKeepAlive(self): # Explicit keep-alive connection header for HTTP 1.0. response = self._makeOne() response._http_connection = 'keep-alive' response.setStatus(304) self._assertResponsesAreEqual(str(response), ('HTTP/1.0 304 Not Modified', 'Connection: Keep-Alive', 'Date: ...', 'Server: ...', '', '')) def test_304ImplicitKeepAlive(self): # Keep-alive is implicit for HTTP 1.1. response = self._makeOne() response._http_version = '1.1' response._http_connection = 'keep-alive' response.setStatus(304) self._assertResponsesAreEqual(str(response), ('HTTP/1.1 304 Not Modified', 'Date: ...', 'Server: ...', '', '')) def test_contentLength(self): # Check that __str__ adds in the correct Content-Length header. response = self._makeOne() response._http_version = '1.1' response._http_connection = 'keep-alive' response.body = '123456789' response.setHeader('Content-Type', 'text/plain') self._assertResponsesAreEqual(str(response), ('HTTP/1.1 200 OK', 'Content-Length: 9', 'Content-Type: text/plain', 'Date: ...', 'Server: ...', '', '123456789')) def test_emptyBody(self): # Check that a response with an empty message body returns a # Content-Length of 0. A common example of this is a 302 redirect. response = self._makeOne() response._http_version = '1.1' response._http_connection = 'keep-alive' response.redirect('somewhere') self._assertResponsesAreEqual(str(response), ('HTTP/1.1 302 Moved Temporarily', 'Content-Length: 0', 'Date: ...', 'Location: somewhere', 'Server: ...', '', '')) def test_HEAD(self): # A response to a HEAD request will have a non zero content # length and an empty body. response = self._makeOne() response._http_version = '1.1' response._http_connection = 'keep-alive' response.setHeader('Content-Type', 'text/plain') response.setHeader('Content-Length', 123) self._assertResponsesAreEqual(str(response), ('HTTP/1.1 200 OK', 'Content-Length: 123', 'Content-Type: text/plain', 'Date: ...', 'Server: ...', '', '')) def test_suite(): suite = unittest.TestSuite() suite.addTests(( unittest.makeSuite(ZServerResponseTestCase), unittest.makeSuite(ZServerHTTPResponseTestCase) )) return suite if __name__ == "__main__": unittest.main(defaultTest="test_suite")