- Jean-Paul's works for a framework used to run tests in running ERP5 instance

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@37926 20353a03-c40f-0410-a6d1-a30d3c3de9de

product/ERP5Type/tests/ERP5TypeLiveTestCase.py

+# -*- coding: utf-8 -*-
+##############################################################################
+#
+# Copyright (c) 2002-2010 Nexedi SA and Contributors. All Rights Reserved.
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsibility of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# guarantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+##############################################################################
+
+import unittest
+import os
+
+from Testing import ZopeTestCase
+from Testing.ZopeTestCase import PortalTestCase, user_name
+from Products.ERP5Type.tests.ProcessingNodeTestCase import ProcessingNodeTestCase
+from Products.ERP5Type.Globals import get_request
+import transaction
+
+from zLOG import LOG, DEBUG, INFO
+
+def profile_if_environ(environment_var_name):
+    if int(os.environ.get(environment_var_name, 0)):
+      def decorator(self, method):
+        def decorated():
+          self.runcall(method)
+        decorated.__name__ = method.__name__
+        decorated.__doc__ = method.__doc__
+        return decorated
+      return decorator
+    else:
+      # No profiling, return identity decorator
+      return lambda self, method: method
+
+class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase):
+    """ERP5TypeLiveTestCase is the default class for *all* tests
+    in ERP5. It is designed with the idea in mind that tests should
+    be run through the web. Command line based tests may be helpful
+    sometimes but should remain an exception because they hinder
+    productivity by adding an extra time to build the
+    environment (which is already built in live instances). 
+        
+    All other test classes should derive from ERP5TypeLiveTestCase.
+    
+    TODO: 
+    - An eplicit list of exceptions to live tests remains to be
+      defined. 
+    """
+
+    def shortDescription(self):
+      description = str(self)
+      doc = self._testMethodDoc
+      if doc and doc.split("\n")[0].strip():
+        description += ', ' + doc.split("\n")[0].strip()
+      return description
+
+    def getTitle(self):
+      """Returns the title of the test, for test reports.
+      """
+      return str(self.__class__)
+
+    def getPortalName(self):
+      """ Return the default ERP5 site id.
+      """
+      return 'erp5' # XXX-JPS For now a hack
+
+    def getPortal(self):
+      """Returns the portal object, i.e. the "fixture root".
+      """
+      return self.app[self.getPortalName()]
+
+    getPortalObject = getPortal
+
+    def login(self, user_name='ERP5TypeTestCase', quiet=0):
+      """
+      Most of the time, we need to login before doing anything
+      """
+      PortalTestCase.login(self, user_name)
+
+    def logout(self):
+      PortalTestCase.logout(self)
+      # clean up certain cache related REQUEST keys that might be associated
+      # with the logged in user
+      for key in ('_ec_cache', '_oai_cache'):
+        pass
+        #self.REQUEST.other.pop(key, None) # XXX
+
+    def _close(self):
+      '''Closes the ZODB connection.'''
+      transaction.abort()
+
+    # class-defined decorators for profiling.
+    # Depending on the environment variable, they return
+    # the same method, or a profiling wrapped call
+    _decorate_setUp = profile_if_environ('PROFILE_SETUP')
+    _decorate_testRun = profile_if_environ('PROFILE_TESTS')
+    _decorate_tearDown = profile_if_environ('PROFILE_TEARDOWN')
+
+    def __call__(self, *args, **kw):
+      # Pulling down the profiling from ZopeTestCase.profiler to allow
+      # overriding run()
+      # This cannot be done at instanciation because we need to
+      # wrap the bottom-most methods, e.g.
+      # SecurityTestCase.tearDown instead of ERP5TestCase.tearDown
+
+      self.setUp = self._decorate_setUp(self.setUp)
+      self.tearDown = self._decorate_tearDown(self.tearDown)
+
+      test_name = self._testMethodName
+      test_method = getattr(self, test_name)
+      setattr(self, test_name, self._decorate_testRun(test_method))
+
+      self.run(*args, **kw)
+
+    def _setup(self):
+        '''Configures the portal. Framework authors may
+           override.
+        '''
+        pass # Do nothing in a live test
+
+    def setUp(self):
+        '''Sets up the fixture. Do not override,
+           use the hooks instead.
+        '''
+        try:
+            self.beforeSetUp()
+            self.app = self._app()
+            self.portal = self._portal()
+            self._setup()
+            self.afterSetUp()
+        except:
+            self.beforeClear()
+            self._clear()
+            raise
+
+    def _app(self):
+        '''Returns the app object for a test.'''
+        request = get_request()
+        return request.PARENTS[-1]
+
+    def afterSetUp(self):
+      '''Called after setUp() has completed. This is
+         far and away the most useful hook.
+      '''
+      pass
+
+    def beforeSetUp(self):
+      '''Called after setUp() has completed. This is
+         far and away the most useful hook.
+      '''
+      pass
+
+    def beforeClear(self):
+      '''Called before _clear(). Subclasses should
+      use it to garbage collect objects which must not remain
+      in the system
+      '''
+      pass
+
+    def logMessage(self, message):
+      """
+        Shortcut function to log a message
+      """
+      ZopeTestCase._print('\n%s ' % message)
+      LOG('Testing ... ', DEBUG, message)
+
+    # Utility methods specific to ERP5Type
+    def getTemplateTool(self):
+      return getToolByName(self.getPortal(), 'portal_templates', None)
+
+    def getPreferenceTool(self) :
+      return getToolByName(self.getPortal(), 'portal_preferences', None)
+
+    def getTrashTool(self):
+      return getToolByName(self.getPortal(), 'portal_trash', None)
+
+    def getPasswordTool(self):
+      return getToolByName(self.getPortal(), 'portal_password', None)
+
+    def getSkinsTool(self):
+      return getToolByName(self.getPortal(), 'portal_skins', None)
+
+    def getCategoryTool(self):
+      return getToolByName(self.getPortal(), 'portal_categories', None)
+
+    def getWorkflowTool(self):
+      return getToolByName(self.getPortal(), 'portal_workflow', None)
+
+    def getCatalogTool(self):
+      return getToolByName(self.getPortal(), 'portal_catalog', None)
+
+    def getTypesTool(self):
+      return getToolByName(self.getPortal(), 'portal_types', None)
+    getTypeTool = getTypesTool
+
+    def getRuleTool(self):
+      return getattr(self.getPortal(), 'portal_rules', None)
+
+    def getClassTool(self):
+      return getattr(self.getPortal(), 'portal_classes', None)
+
+    def getSimulationTool(self):
+      return getToolByName(self.getPortal(), 'portal_simulation', None)
+
+    def getSQLConnection(self):
+      return getToolByName(self.getPortal(), 'erp5_sql_connection', None)
+
+    def getPortalId(self):
+      return self.getPortal().getId()
+
+    def getDomainTool(self):
+      return getToolByName(self.getPortal(), 'portal_domains', None)
+
+    def getAlarmTool(self):
+      return getattr(self.getPortal(), 'portal_alarms', None)
+
+    def getActivityTool(self):
+      return getattr(self.getPortal(), 'portal_activities', None)
+
+    def getArchiveTool(self):
+      return getattr(self.getPortal(), 'portal_archives', None)
+
+    def getCacheTool(self):
+      return getattr(self.getPortal(), 'portal_caches', None)
+
+    def getOrganisationModule(self):
+      return getattr(self.getPortal(), 'organisation_module',
+          getattr(self.getPortal(), 'organisation', None))
+
+    def getPersonModule(self):
+      return getattr(self.getPortal(), 'person_module',
+          getattr(self.getPortal(), 'person', None))
+
+    def getCurrencyModule(self):
+      return getattr(self.getPortal(), 'currency_module',
+          getattr(self.getPortal(), 'currency', None))
+
+    def validateRules(self):
+      """
+      try to validate all rules in rule_tool.
+      """
+      rule_tool = self.getRuleTool()
+      for rule in rule_tool.contentValues(
+          portal_type=rule_tool.getPortalRuleTypeList()):
+        if rule.getValidationState() != 'validated':
+          rule.validate()
+
+    def createSimpleUser(self, title, reference, function):
+      """
+        Helper function to create a Simple ERP5 User.
+        User password is the reference.
+        
+        XXX-JPS do wa have a "delete" method etc.
+      """
+      user = self.createUser(reference, person_kw=dict(title=title))
+      assignment = self.createUserAssignement(user, assignment_kw=dict(function=function))
+      return user
+
+    def createUser(self, reference, password=None, person_kw=None):
+      """
+        Create an ERP5 User.
+        Default password is the reference.
+        person_kw is passed as additional arguments when creating the person
+      """
+      if password is None:
+        password = reference
+      if person_kw is None:
+        person_kw = dict()
+
+      person = self.portal.person_module.newContent(portal_type='Person',
+                                                    reference=reference,
+                                                    password=password,
+                                                    **person_kw)
+      return person
+
+    def createUserAssignment(self, user, assignment_kw):
+      """
+        Create an assignment to user.
+      """
+      assignment = user.newContent(portal_type='Assignment', **assignment_kw)
+      assignment.open()
+      return assignment
+
+    def createUserAssignement(self, user, assignment_kw):
+      # BBB
+      warn('createUserAssignement is deprecated;'
+           'Use createUserAssignment instead',
+           DeprecationWarning)
+      return self.createUserAssignment(user, assignment_kw)
+
+    def failIfDifferentSet(self, a, b, msg=""):
+      if not msg:
+        msg='%r != %r' % (a, b)
+      for i in a:
+        self.failUnless(i in b, msg)
+      for i in b:
+        self.failUnless(i in a, msg)
+      self.assertEquals(len(a), len(b), msg)
+    assertSameSet = failIfDifferentSet
+
+    def assertWorkflowTransitionFails(self, object, workflow_id, transition_id,
+        error_message=None, state_variable='simulation_state'):
+      """
+        Check that passing given transition from given workflow on given object
+        raises ValidationFailed.
+        Do sanity checks (workflow history length increased by one, simulation
+        state unchanged).
+        If error_message is provided, it is asserted to be equal to the last
+        workflow history error message.
+      """
+      workflow_tool = self.getWorkflowTool()
+      reference_history_length = len(workflow_tool.getInfoFor(ob=object, name='history', wf_id=workflow_id))
+      state_method = 'get' + convertToUpperCase(state_variable)
+      method = getattr(object, state_method, None)
+      reference_workflow_state = method()
+      self.assertRaises(ValidationFailed, workflow_tool.doActionFor, object, transition_id, wf_id=workflow_id)
+      workflow_history = workflow_tool.getInfoFor(ob=object, name='history', wf_id=workflow_id)
+      self.assertEqual(len(workflow_history), reference_history_length + 1)
+      workflow_error_message = str(workflow_history[-1]['error_message'])
+      if error_message is not None:
+        self.assertEqual(workflow_error_message, error_message)
+      self.assertEqual(method(), reference_workflow_state)
+      return workflow_error_message
+
+    def tearDown(self):
+      '''Tears down the fixture. Do not override,
+         use the hooks instead.
+      '''
+      PortalTestCase.tearDown(self)
+
+    def beforeClose(self):
+      """
+      Clear "my activities"... how to do this ?
+      """
+      PortalTestCase.beforeClose(self)
+
+    def stepPdb(self, sequence=None, sequence_list=None):
+      """Invoke debugger"""
+      try: # try ipython if available
+        import IPython
+        IPython.Shell.IPShell(argv=[])
+        tracer = IPython.Debugger.Tracer()
+      except ImportError:
+        from pdb import set_trace as tracer
+      tracer()
+
+    def stepTic(self, **kw):
+      """
+      The is used to simulate the zope_tic_loop script
+      Each time this method is called, it simulates a call to tic
+      which invoke activities in the Activity Tool
+      """
+      if kw.get('sequence', None) is None:
+        # in case of using not in sequence commit transaction
+        transaction.commit()
+      self.tic()
+
+    def publish(self, path, basic=None, env=None, extra=None,
+                request_method='GET', stdin=None, handle_errors=True):
+        '''Publishes the object at 'path' returning a response object.'''
+
+        from ZPublisher.Response import Response
+        from ZPublisher.Test import publish_module
+
+        from AccessControl.SecurityManagement import getSecurityManager
+        from AccessControl.SecurityManagement import setSecurityManager
+        
+        # Save current security manager
+        sm = getSecurityManager()
+
+        # Commit the sandbox for good measure
+        transaction.commit()
+
+        if env is None:
+            env = {}
+        if extra is None:
+            extra = {}
+
+        request = self.app.REQUEST
+
+        env['SERVER_NAME'] = request['SERVER_NAME']
+        env['SERVER_PORT'] = request['SERVER_PORT']
+        env['HTTP_ACCEPT_CHARSET'] = request['HTTP_ACCEPT_CHARSET']
+        env['REQUEST_METHOD'] = request_method
+
+        p = path.split('?')
+        if len(p) == 1:
+            env['PATH_INFO'] = p[0]
+        elif len(p) == 2:
+            [env['PATH_INFO'], env['QUERY_STRING']] = p
+        else:
+            raise TypeError, ''
+
+        if basic:
+            env['HTTP_AUTHORIZATION'] = "Basic %s" % base64.encodestring(basic).replace('\012', '')
+
+        if stdin is None:
+            stdin = StringIO()
+
+        outstream = StringIO()
+        response = Response(stdout=outstream, stderr=sys.stderr)
+
+        publish_module('Zope2',
+                       response=response,
+                       stdin=stdin,
+                       environ=env,
+                       extra=extra,
+                       debug=not handle_errors,
+                      )
+
+        # Restore security manager
+        setSecurityManager(sm)
+
+        return ResponseWrapper(response, outstream, path)
+
+def runLiveTest(self, suite):
+  from Products.ERP5Type.tests.runUnitTest import ERP5TypeTestLoader
+  from Products.ERP5Type.tests import backportUnittest
+  from StringIO import StringIO
+  TestRunner = backportUnittest.TextTestRunner
+  unittest.TestLoader = ERP5TypeTestLoader
+  stream = StringIO()
+  output = StringIO()
+  output.write("**Running Live Test:\n")
+  result = TestRunner(stream=stream).run(suite)
+  output.write(stream.getvalue())
+  return output.getvalue()