[Zope-CVS] CVS: Packages/FunctionalTests/FunctionalTests - Invocation.py:1.1 Request.py:1.1 Result.py:1.1 interfaces.py:1.1 FTRunner.py:1.3 Framework.py:1.3
Tres Seaver
tseaver@zope.com
Tue, 20 May 2003 21:05:07 -0400
Update of /cvs-repository/Packages/FunctionalTests/FunctionalTests
In directory cvs.zope.org:/tmp/cvs-serv3224/FunctionalTests
Modified Files:
FTRunner.py Framework.py
Added Files:
Invocation.py Request.py Result.py interfaces.py
Log Message:
- Refactor over-large Framework.py:
o Split out Request, Result, and Invocation classes / helpers
into separate modlues.
o Begin process of "scarecrowifying" the framework interfaces.
o Clean up docstrings to match newer standards.
o Fix long-broken composite result test, uncovered via refactoring.
=== Added File Packages/FunctionalTests/FunctionalTests/Invocation.py ===
""" Classes: InvocationBase, HTTPRequestInvocation, ZEORequestInvocation
$Id: Invocation.py,v 1.1 2003/05/21 01:05:05 tseaver Exp $
"""
import time
from interfaces import IRequestInvocation
class InvocationBase:
""" Base class for implementing IRequestInvocation.
"""
_start_time = None
_stop_time = None
def __init__( self, request ):
self._request = request
def getRequest( self ):
""" See IRequestInvocation.
"""
return self._request
def beginRequest( self ):
""" See IRequestInvocation.
"""
self._start_time = time.time()
def endRequest( self ):
""" See IRequestInvocation.
"""
self._stop_time = time.time()
def dump( self ):
""" See IRequestInvocation.
"""
raise NotImplementedError
def getResult( self ):
""" See IRequestInvocation.
"""
raise NotImplementedError
def getElapsedTime( self ):
""" See IRequestInvocation.
"""
return self._stop_time and self._stop_time - self._start_time or 0.0
class HTTPRequestInvocation( InvocationBase ):
__implements__ = ( IRequestInvocation, )
_reply_code = 200
_reply_message = None
_reply_headers = None
_reply_payload = None
def update( self, reply_code, reply_message, reply_headers, reply_payload ):
""" See IRequestInvocation.
"""
self._reply_code = reply_code
self._reply_headers = reply_headers
self._reply_message = reply_message
self._reply_payload = reply_payload
def dump( self ):
""" See IRequestInvocation.
"""
return ( 'HTTP Request: result = %s\n%s\n\n%s'
% ( self.getReplyStatus()
, self._reply_headers
, self._reply_payload
)
)
def getReplyCode( self ):
""" See IRequestInvocation.
"""
return self._reply_code
getResult = getReplyCode
def getReplyStatus( self ):
""" See IRequestInvocation.
"""
return '%d %s' % ( self._reply_code, self._reply_message )
def getReplyPayload( self ):
""" See IRequestInvocation.
"""
return self._reply_payload
def getRedirect( self ):
""" See IRequestInvocation.
o Return None if no redirect was issued.
"""
return self._reply_headers.get( 'Location' )
def getCookies( self ):
""" See IRequestInvocation.
o Each object should describe one cookie set by the response.
"""
result = []
for h in filter( lambda x: x.lower().startswith( 'set-cookie:' )
, self._reply_headers.headers ):
cookie = Cookie.SmartCookie()
cookie.load( h )
result.append( cookie )
return result
class ZEORequestInvocation( InvocationBase ):
__implements__ = ( IRequestInvocation, )
_result = None
def update( self, result ):
""" See IRequestInvocation.
"""
self._result = result
def dump( self ):
""" See IRequestInvocation.
"""
return 'ZEO Request: result = %s' % self._result
def getResult( self ):
""" See IRequestInvocation.
"""
return self._result
=== Added File Packages/FunctionalTests/FunctionalTests/Request.py === (516/616 lines abridged)
""" Classes: RequestError, HTTPRequest, ZEORequest
Functions: buildRequest
$Id
"""
import re
import httplib
import urlparse
from Invocation import HTTPRequestInvocation
from Invocation import ZEORequestInvocation
class RequestError( Exception ):
pass
class _RequestBase: # XXX scarecrow me?
""" Common base class for requests.
"""
def __init__( self, name ):
self._name = name
self._expected_result = 1
self._expected_time = 1.0 # second
#
# Accessors
#
def getName( self ):
""" Return the name of this request within the scenario.
"""
return self._name
def getExpectedResult( self ):
""" Return the expected result (return value, HTTP request code,
etc.)
"""
return self._expected_result
def getExpectedTime( self ):
""" Return the maximum expected time to complete the request.
"""
return self._expected_time
def __call__( self, result ):
[-=- -=- -=- 516 lines omitted -=- -=- -=-]
for fk in field_keys:
r.addField( cp.get( section, fk ) )
if 'expected_result' in options:
r.setExpectedResult( cp.getint( section, 'expected_result' ) )
if 'expected_redirect' in options:
r.setExpectedRedirect( cp.get( section, 'expected_redirect' ) )
if 'expected_time' in options:
r.setExpectedTime( cp.getfloat( section, 'expected_time' ) )
if 'expected_cookies' in options:
r.setExpectedCookies( cp.get( section, 'expected_cookies' ) )
if 'flush_cookies' in options:
r.setExpectedCookies( None )
return r
def _buildZEORequest( cp, section ):
""" Construct a ZEORequest, using the values in 'cp' and 'section'.
"""
r = ZEORequest( section
, cp.get( section, 'Filename' )
, cp.get( section, 'Function' )
)
options = cp.options( section )
if 'expected_result' in options:
r.setExpectedResult( cp.getint( section, 'expected_result' ) )
if 'expected_time' in options:
r.setExpectedTime( cp.getfloat( section, 'expected_time' ) )
return r
def buildRequest( cp, section ):
""" Construct a Request, using the values in 'cp' and 'section'.
"""
if 'sleep' in cp.options( section ):
return _buildSleepRequest( cp, section )
elif 'url' in cp.options( section ):
return _buildHTTPRequest( cp, section )
else:
return _buildZEORequest( cp, section )
=== Added File Packages/FunctionalTests/FunctionalTests/Result.py ===
""" Classes: Result
$Id: Result.py,v 1.1 2003/05/21 01:05:05 tseaver Exp $
"""
from interfaces import IResult
class Result:
""" Hold on to the results of running a test or suite of tests.
"""
__implements__ = ( IResult, )
def __init__( self
, test
, application=None
, defaults={}
, time_requests=1
, check_responses=1
, check_redirects=1
, check_content=1
, check_elapsed_times=1
):
self._test = test
self._application = application
self._defaults = {}
self._defaults.update( defaults )
self._state = {}
self._time_requests = time_requests
self._check_responses = check_responses
self._check_redirects = check_redirects
self._check_content = check_content
self._check_elapsed_times = check_elapsed_times and time_requests
self._invocations = []
self._errors = []
self._fatal_errors = []
self._children = []
self._cookies = []
self._parent = None
#
# Queries
#
def getTest( self ):
""" See IResult.
"""
return self._test
def getApplication( self ):
""" See IResult.
"""
return self._application
def getDefaults( self ):
""" See IResult.
"""
result = {}
result.update( self._defaults )
return result
def getParent( self ):
""" See IResult.
"""
return self._parent
def setParent( self, parent ):
""" See IResult.
"""
self._parent = parent
def getCookies( self ):
""" See IResult.
"""
if self._cookies:
return self._cookies
#import pdb; pdb.set_trace()
if self._parent:
return self._parent.getCookies()
return []
def setCookies( self, cookies ):
""" See IResult.
"""
self._cookies = cookies
if self._parent:
#import pdb; pdb.set_trace()
self._parent.setCookies( cookies )
def setStateValue( self, key, value ):
""" See IResult.
"""
self._state[ key ] = value
def getStateValue( self, key ):
""" See IResult.
"""
return self._state[ key ]
def timeRequests( self ):
""" See IResult.
"""
return self._time_requests
def checkResponses( self ):
""" See IResult.
"""
return self._check_responses
def checkRedirects( self ):
""" See IResult.
"""
return self._check_redirects
def checkContent( self ):
""" See IResult.
"""
return self._check_content
def checkElapsedTimes( self ):
""" See IResult.
"""
return self._check_elapsed_times
def listInvocations( self, roll_up=1 ):
""" See IResult.
o If 'roll_up', return transitive closure, else only return
those we manage directly.
"""
result = []
result.extend( self._invocations )
if roll_up:
for child in self.listChildren():
result.extend( child.listInvocations( roll_up ) )
return result
def listErrors( self, fatal_only=0, roll_up=1 ):
""" See IResult.
"""
result = []
if fatal_only:
result.extend( self._fatal_errors )
else:
result.extend( self._fatal_errors )
result.extend( self._errors )
if roll_up:
for child in self.listChildren():
result.extend( child.listErrors( fatal_only, roll_up ) )
return result
def listChildren( self ):
""" See IResult.
"""
return tuple( self._children )
def listDumpedResults( self ):
result = []
for invocation in self._invocations:
result.append( invocation.dump() )
return result
#
# Mutators
#
def addInvocation( self, invocation, request ):
""" See IResult.
"""
self._invocations.append( invocation )
self._validateInvocation( invocation, request )
def newChild( self, test ):
""" See IResult.
"""
klass = self.__class__
child = klass( test
, self.getApplication()
, self.getDefaults()
, self.timeRequests()
, self.checkResponses()
, self.checkElapsedTimes()
)
child.setParent( self )
self._children.append( child )
return child
def logError( self, msg ):
""" See IResult.
"""
self._errors.append( msg )
def logFatalError( self, msg ):
""" See IResult.
"""
self._fatal_errors.append( msg )
def __call__( self ):
return len( self._fatal_errors ) + len( self._errors ) == 0
#
# Utilities
#
def _validateInvocation( self
, invocation
, request
, force_check_response=0
, force_check_redirects=0
, force_check_content=0
, force_check_time=0
):
""" Test the result against the required conditions.
o The 'force_*' arguments override the values stored as attributes.
"""
if self.checkResponses() or force_check_response:
self._validateResponse( invocation, request )
if self.checkRedirects() or force_check_redirects:
self._validateRedirect( invocation, request )
if self.checkContent() or force_check_content:
self._validatePayload( invocation, request )
if self.checkElapsedTimes() or force_check_time:
self._validateElapsedTime( invocation, request )
def _validateResponse( self, invocation, request ):
""" Check that the response code is correct.
o Log problems to self._fatal_errors.
"""
expected = request.getExpectedResult()
got = invocation.getResult()
if got != expected:
msg = '[%-10s] response: %s != %s' % ( request.getName()
, got, expected )
self.logFatalError( msg )
def _validateRedirect( self, invocation, request ):
""" Check that we got the expected redirect response.
o Log problems to self._fatal_errors.
"""
expected = request.getExpectedRedirect()
if expected != None:
expected = expected.strip()
got = invocation.getRedirect()
if got != expected:
msg = '[%-10s] response: %s != %s' % ( request.getName()
, got, expected )
self.logFatalError( msg )
def _validatePayload( self, invocation, request ):
""" Apply specified check on the response payload.
o Log problems to self._fatal_errors / self._errors
"""
def _validateElapsedTime( self, invocation, request ):
""" Did the request complete fast enough?
o Log problems to self._errors.
"""
expected = request.getExpectedTime()
got = invocation.getElapsedTime()
if got > expected:
msg = '[%-10s] elapsed: %s != %s' % ( request.getName()
, got, expected )
self.logError( msg )
=== Added File Packages/FunctionalTests/FunctionalTests/interfaces.py ===
""" Interfaces: IRequestInvocation
$Id: interfaces.py,v 1.1 2003/05/21 01:05:05 tseaver Exp $
"""
try:
from Interface import Interface
except ImportError:
class Interface: pass
class IRequest( Interface ):
""" Define an abstract request, apart from any invocation.
"""
class IResult( Interface ):
""" Represent the result of invoking one or more related requests.
o Holds onto lists of invocations and sub-results.
o Implements GoF Composite pattern.
"""
#
# Accessors
#
def getTest( self ):
""" Return our test object.
"""
def getApplication( self ):
""" Return the root object of Zope.
"""
def getDefaults( self ):
""" Return our test object.
"""
def getParent( self ):
""" Return our parent request, or None.
"""
def getCookies( self ):
""" Return cookies we have accumulated during the request.
"""
def timeRequests( self ):
""" Should we capture elapsed times?
"""
def checkResponses( self ):
""" Should we validate response values?
"""
def checkRedirects( self ):
""" Should we validate redirect locations?
"""
def checkContent( self ):
""" Should we validate content?
"""
def checkElapsedTimes( self ):
""" Should we validate elapsed times?
"""
def listInvocations( self, roll_up=1 ):
""" Return the list of invocations we manage.
o If 'roll_up', return transitive closure, else only return
those we manage directly.
"""
def listErrors( self, fatal_only=0, roll_up=1 ):
""" Return a list of fatal errors.
"""
def listChildren( self ):
""" Return our sub-results.
"""
def listDumpedResults( self ):
""" Return a list of verbose invocation dumps.
"""
def __call__( self ):
""" Test the validity of the request.
o It is valid if it has not yet seen any errors.
"""
#
# Mutators
#
def setParent( self, parent ):
""" Setup our parent request.
"""
def setCookies( self, cookies ):
""" Store our cookies.
"""
def setStateValue( self, key, value ):
""" Remember a value (e.g., across invocations)
"""
def getStateValue( self, key ):
""" Recall a value stored using 'setStateValue'.
"""
def addInvocation( self, invocation, request ):
""" Remember a request invocation other than setup / postcondition.
"""
def newChild( self, test ):
""" Create, remember, and return a sub-result.
"""
def logError( self, msg ):
""" Record a fatal error.
"""
def logFatalError( self, msg ):
""" Record a fatal error.
"""
class IRequestInvocation( Interface ):
""" Record particular request invocations.
"""
def getRequest( self ):
""" Return the request for which we were invoked.
"""
def beginRequest( self ):
""" Record the start of the request invocation.
"""
def endRequest( self ):
""" Record the end of the request invocation.
"""
def dump( self ):
""" Return a string describing the invocation.
"""
def getResult( self ):
""" Return a value representing the result of the invocation.
"""
def getElapsedTime( self ):
""" Return the time taken by this invocation.
"""
=== Packages/FunctionalTests/FunctionalTests/FTRunner.py 1.2 => 1.3 ===
--- Packages/FunctionalTests/FunctionalTests/FTRunner.py:1.2 Tue May 20 09:21:25 2003
+++ Packages/FunctionalTests/FunctionalTests/FTRunner.py Tue May 20 21:05:05 2003
@@ -307,7 +307,7 @@
, request.getURI()[:55]
), # note comma
- print ' %0.3f' % invocation.getInterval()
+ print ' %0.3f' % invocation.getElapsedTime()
def printResult( self, result ):
=== Packages/FunctionalTests/FunctionalTests/Framework.py 1.2 => 1.3 === (937/1037 lines abridged)
--- Packages/FunctionalTests/FunctionalTests/Framework.py:1.2 Tue May 20 09:21:25 2003
+++ Packages/FunctionalTests/FunctionalTests/Framework.py Tue May 20 21:05:05 2003
@@ -1,1000 +1,16 @@
-import string
import os
import re
-import time
-import httplib
-import urllib
-import urlparse
-import base64
-import Cookie
-import MimeWriter
-import StringIO
import ConfigParserExt
-#
-# Result classes
-#
-
-class Result:
- """
- Hold on to the results of running a test or suite of tests.
- Implements the GoF Composite pattern.
- """
- def __init__( self
- , test
- , application=None
- , defaults={}
- , time_requests=1
- , check_responses=1
- , check_redirects=1
- , check_content=1
- , check_elapsed_times=1
- ):
-
- self._test = test
- self._application = application
- self._defaults = {}
- self._defaults.update( defaults )
- self._state = {}
- self._time_requests = time_requests
- self._check_responses = check_responses
- self._check_redirects = check_redirects
- self._check_content = check_content
- self._check_elapsed_times = check_elapsed_times and time_requests
- self._invocations = []
- self._errors = []
- self._fatal_errors = []
[-=- -=- -=- 937 lines omitted -=- -=- -=-]
-
- if 'expected_result' in options:
- r.setExpectedResult( cp.getint( section, 'expected_result' ) )
-
- if 'expected_time' in options:
- r.setExpectedTime( cp.getfloat( section, 'expected_time' ) )
-
- return r
-
-def _buildRequest( cp, section ):
- """
- Construct a Request, using the values in 'cp' and 'section'.
- """
- if 'sleep' in cp.options( section ):
- return _buildSleepRequest( cp, section )
- elif 'url' in cp.options( section ):
- return _buildHTTPRequest( cp, section )
- else:
- return _buildZEORequest( cp, section )
#
# Execution classes
@@ -1279,7 +295,7 @@
, 'load_sequence' ) )
if _SETUP_SECTION in cp.listSpecialSections( 0 ):
- test.setSetup( _buildRequest( cp, _SETUP_SECTION ) )
+ test.setSetup( buildRequest( cp, _SETUP_SECTION ) )
for section in cp.listOtherSections():
options = cp.options( section )
@@ -1288,15 +304,15 @@
else:
repeat_count = 1
if isScenario:
- test.addRequest( _buildRequest( cp, section ), repeat_count )
+ test.addRequest( buildRequest( cp, section ), repeat_count )
else:
child = buildTest( cp.get( section, 'file' ), defaults )
test.addChild( child, repeat_count )
if isScenario and _POSTCONDITION_SECTION in cp.listSpecialSections( 0 ):
- test.setPostcondition( _buildRequest( cp, _POSTCONDITION_SECTION ) )
+ test.setPostcondition( buildRequest( cp, _POSTCONDITION_SECTION ) )
if _TEARDOWN_SECTION in cp.listSpecialSections( 0 ):
- test.setTeardown( _buildRequest( cp, _TEARDOWN_SECTION ) )
+ test.setTeardown( buildRequest( cp, _TEARDOWN_SECTION ) )
return test