[Zope-Checkins] CVS: Zope3/lib/python/Zope/Server/tests - __init__.py:1.1.2.1 testHTTPServer.py:1.1.2.1
Tres Seaver
tseaver@zope.com
Thu, 22 Nov 2001 00:41:36 -0500
Update of /cvs-repository/Zope3/lib/python/Zope/Server/tests
In directory cvs.zope.org:/tmp/cvs-serv3409/lib/python/Zope/Server/tests
Added Files:
Tag: Zope-3x-branch
__init__.py testHTTPServer.py
Log Message:
- Add unit tests for parts of Zope.Server.HTTPServer.
=== Added File Zope3/lib/python/Zope/Server/tests/__init__.py ===
# Copyright (c) 2001 Zope Corporation and Contributors. All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 1.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.
""" Unit tests for Zope.Server """
=== Added File Zope3/lib/python/Zope/Server/tests/testHTTPServer.py ===
# Copyright (c) 2001 Zope Corporation and Contributors. All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 1.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.
""" Unit tests for Zope.Server.HTTPResponse """
import unittest
import Zope.Publisher.HTTP.IPayload
from urlparse import urlparse
#
# Test shims.
#
class InitAny:
def __init__( self, **kw ):
self.__dict__.update( kw )
class FauxHandler( InitAny ):
data = request = None
def continue_request( self, data, request ):
self.data = data
self.request = request
class FauxRequest( InitAny ):
collector = DEFAULT_COLLECTOR = []
uri = 'http://www.zope.org/Resources;special?id=abc#aargh'
def split_uri( self ):
scheme, nethost, path, parms, query, fragment = urlparse( self.uri )
if parms:
parms = ';%s' % parms # Compensate for some stupidity
if query:
query = '?%s' % query # Compensate for some stupidity
return path, parms, query, fragment
class FauxChannel( InitAny ):
terminator = None
worked = 0
def set_terminator( self, value ):
self.terminator = value
def work( self ):
self.worked = 1
class FauxServer( InitAny ):
pass
class FauxRequestPayload:
__implements__ = Zope.Publisher.HTTP.IPayload.IRequestPayload
def processInputs(self, request):
"""
Processes request inputs.
"""
self.body = request.full_instream.read()
def getPublication(self, request):
"""
Returns the publication object.
"""
def debugInfo(self, request):
"""
Returns text containing debugging information.
"""
class FauxResponsePayload:
__implements__ = Zope.Publisher.HTTP.IPayload.IResponsePayload
def setBody(self, response, body):
"""
Sets the body of the response.
"""
def handleException(self, response, exc_info):
"""
Calls setBody() with an error response.
"""
#
# Test the collector.
#
class zhttp_collectorTest( unittest.TestCase ):
def setUp( self ):
self.handler = FauxHandler()
self.request = FauxRequest()
self.request.channel = self.channel = FauxChannel()
def tearDown( self ):
del self.channel
del self.request
def testZero( self ):
"""
Test building a collector with size = 0
"""
from Zope.Server.HTTPServer import zhttp_collector
collector = zhttp_collector( self.handler, self.request, 0 )
self.assertEqual( self.channel.terminator, 0 )
self.assertEqual( self.request.collector, collector )
self.failUnless( self.handler.data is None )
self.failUnless( self.handler.request is None )
collector.found_terminator()
self.assertEqual( self.handler.data.read(), '' )
self.failUnless( self.handler.request is self.request )
def testSmaller( self ):
from Zope.Server.HTTPServer import zhttp_collector
collector = zhttp_collector( self.handler, self.request, 15 )
self.assertEqual( self.channel.terminator, 15 )
self.assertEqual( self.request.collector, collector )
self.failUnless( self.handler.data is None )
self.failUnless( self.handler.request is None )
junk = 'x' * 15
collector.collect_incoming_data( junk )
collector.found_terminator()
self.assertEqual( self.handler.data.read(), junk )
self.failUnless( self.handler.request is self.request )
def testLarger( self ):
from Zope.Server.HTTPServer import zhttp_collector
collector = zhttp_collector( self.handler, self.request, 600000 )
self.assertEqual( self.channel.terminator, 600000 )
self.assertEqual( self.request.collector, collector )
self.failUnless( self.handler.data is None )
self.failUnless( self.handler.request is None )
junk = 'x' * 1000
for i in range( 600 ):
collector.collect_incoming_data( junk )
collector.found_terminator()
stuff = self.handler.data.read()
self.assertEqual( stuff, junk*600 )
self.failUnless( self.handler.request is self.request )
#
# Test the handler.
#
class zhttp_handlerTest( unittest.TestCase ):
def setUp( self ):
self.request_payload = FauxRequestPayload()
self.response_payload = FauxResponsePayload()
self.resolver = InitAny( cache={ '123.45.67.89'
: ( 0, 1, 'my.host.tla' )
} )
self.logger = InitAny( resolver=self.resolver )
self.server = FauxServer( port=9673
, server_name='localhost'
, SERVER_IDENT='XYZZY'
, logger=self.logger
)
self.channel = FauxChannel( server=self.server
, creation_time='2001/01/01 00:00:00 GMT'
, addr=('123.45.67.89', )
, queue=[]
)
self.request = FauxRequest( channel=self.channel
, command='GET'
, version='3.14'
, header= ( 'Foo: foo', 'Bar-baz: bar-baz' )
)
def testEmpty( self ):
from Zope.Server.HTTPServer import zhttp_handler
handler = zhttp_handler( self.request_payload, self.response_payload )
self.failUnless( handler.request_payload is self.request_payload )
self.failUnless( handler.response_payload is self.response_payload )
self.failIf( handler.env_override )
self.failIf( handler.hits )
self.failUnless( handler.match( FauxRequest( uri = '/' ) ) )
self.failUnless( handler.match( FauxRequest( uri = '/abc' ) ) )
self.failUnless( handler.match( FauxRequest( uri = '/abc/def' ) ) )
self.failIf( handler.match( FauxRequest( uri = '#' ) ) )
def testInit( self ):
from Zope.Server.HTTPServer import zhttp_handler
override = { 'foo' : 1, 'bar' : 2 }
handler = zhttp_handler( self.request_payload
, self.response_payload
, 'abc'
, override
)
self.assertEqual( handler.env_override, override )
self.failIf( handler.hits )
self.failIf( handler.match( FauxRequest( uri = '/' ) ) )
self.failUnless( handler.match( FauxRequest( uri = '/abc' ) ) )
self.failUnless( handler.match( FauxRequest( uri = '/abc/def' ) ) )
self.failIf( handler.match( FauxRequest( uri = '#' ) ) )
def testGetEnvironment( self ):
from Zope.Server.HTTPServer import zhttp_handler
#
# Test default URL, w/ no uri-base
#
req = self.request
req.command = 'sneer'
handler = zhttp_handler( self.request_payload, self.response_payload )
env = handler.get_environment( req )
self.assertEqual( env['REQUEST_METHOD'], 'SNEER' )
self.assertEqual( int( env['SERVER_PORT'] ), 9673 )
self.assertEqual( env['SERVER_NAME'], 'localhost' )
self.assertEqual( env['SERVER_SOFTWARE'], 'XYZZY' )
self.assertEqual( env['SERVER_PROTOCOL'], "HTTP/3.14" )
self.assertEqual( env['channel.creation_time']
, '2001/01/01 00:00:00 GMT' )
self.assertEqual( env['SCRIPT_NAME'], '' )
self.assertEqual( env['PATH_INFO'], '/Resources;special' )
self.assertEqual( env['QUERY_STRING'], 'id=abc' )
self.assertEqual( env['GATEWAY_INTERFACE'], 'CGI/1.1' )
self.assertEqual( env['REMOTE_ADDR'], '123.45.67.89' )
self.assertEqual( env['REMOTE_HOST'], 'my.host.tla' )
self.assertEqual( env['HTTP_FOO'], 'foo' )
self.assertEqual( env['HTTP_BAR_BAZ'], 'bar-baz' )
#
# Test with a different URL, and a uri-base.
#
req = FauxRequest( channel=self.channel
, command='glower'
, version='3.14'
, header= ()
, uri='http://www.zope.org/hmm/WikiCentral/DevWiki'
)
handler = zhttp_handler( self.request_payload
, self.response_payload
, uri_base='hmm'
)
env = handler.get_environment( req )
self.assertEqual( env['SCRIPT_NAME'], '/hmm' )
self.assertEqual( env['PATH_INFO'], '/WikiCentral/DevWiki' )
self.failIf( env.get( 'QUERY_STRING', None ) is not None )
self.failIf( env.get( 'HTTP_FOO', None ) is not None )
self.failIf( env.get( 'HTTP_BAR_BAZ', None ) is not None )
def testHandleNoLength( self ):
from Zope.Server.HTTPServer import zhttp_handler
self.channel.current_request = req = self.request
handler = zhttp_handler( self.request_payload, self.response_payload )
self.failIf( handler.hits.as_long() )
self.failIf( self.channel.worked )
handler.handle_request( req )
self.assertEqual( handler.hits.as_long(), 1 )
self.assertEqual( self.channel.current_request, None )
self.assertEqual( len( self.channel.queue ), 1 )
self.failUnless( self.channel.worked )
httpreq = self.channel.queue[0]
self.assertEqual( httpreq[ 'REQUEST_METHOD' ], 'GET' )
self.assertEqual( int( httpreq['SERVER_PORT'] ), 9673 )
self.assertEqual( httpreq['SERVER_NAME'], 'localhost' )
self.assertEqual( httpreq['SERVER_SOFTWARE'], 'XYZZY' )
self.assertEqual( httpreq['SERVER_PROTOCOL'], "HTTP/3.14" )
self.assertEqual( httpreq['SCRIPT_NAME'], '' )
self.assertEqual( httpreq['PATH_INFO'], '/Resources;special' )
self.assertEqual( httpreq['QUERY_STRING'], 'id=abc' )
self.assertEqual( httpreq['GATEWAY_INTERFACE'], 'CGI/1.1' )
self.assertEqual( httpreq['REMOTE_ADDR'], '123.45.67.89' )
self.assertEqual( httpreq['REMOTE_HOST'], 'my.host.tla' )
self.assertEqual( httpreq['HTTP_FOO'], 'foo' )
self.assertEqual( httpreq['HTTP_BAR_BAZ'], 'bar-baz' )
self.assertEqual( httpreq.response._http_version, req.version )
self.assertEqual( httpreq.response._server_version, 'XYZZY' )
def testHandleWithLength( self ):
from Zope.Server.HTTPServer import zhttp_handler
self.channel.current_request = req = self.request
req.header = ( 'Content-length: 100', )
handler = zhttp_handler( self.request_payload, self.response_payload )
self.failIf( handler.hits.as_long() )
self.failIf( self.channel.worked )
handler.handle_request( req )
self.assertEqual( handler.hits.as_long(), 1 )
self.assertEqual( self.channel.current_request, req )
self.assertEqual( len( self.channel.queue ), 0 )
self.failIf( self.channel.worked )
collector = getattr( req, 'collector', None )
self.failUnless( collector is not None )
junk = 'x' * 100
collector.collect_incoming_data( junk )
collector.found_terminator()
self.assertEqual( self.channel.current_request, None )
self.assertEqual( len( self.channel.queue ), 1 )
self.failUnless( self.channel.worked )
httpreq = self.channel.queue[0]
httpreq.processInputs()
self.assertEqual( self.request_payload.body, junk )