[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