Skip to content

E
erp5
Commit ed4e67bc authored by Sebastien Robin's avatar Sebastien Robin
Browse files
Options

create mixin class for code common between live tests 

and command line tests

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@42528 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent 44470bd3
Hide whitespace changes
Inline Side-by-side
Showing with 438 additions and 718 deletions
product/ERP5Type/tests/ERP5TypeLiveTestCase.py
View file @ ed4e67bc
...@@ -35,6 +35,7 @@ from Testing.ZopeTestCase import PortalTestCase, user_name ...@@ -35,6 +35,7 @@ from Testing.ZopeTestCase import PortalTestCase, user_name
from Products.CMFCore.utils import getToolByName from Products.CMFCore.utils import getToolByName
from Products.ERP5Type.tests.ProcessingNodeTestCase import ProcessingNodeTestCase from Products.ERP5Type.tests.ProcessingNodeTestCase import ProcessingNodeTestCase
from Products.ERP5Type.Globals import get_request from Products.ERP5Type.Globals import get_request
from ERP5TypeTestCase import ERP5TypeTestCaseMixin
import transaction import transaction
from zLOG import LOG, DEBUG, INFO from zLOG import LOG, DEBUG, INFO
...@@ -58,7 +59,7 @@ from Products.ERP5Type.tests import ProcessingNodeTestCase as\ ...@@ -58,7 +59,7 @@ from Products.ERP5Type.tests import ProcessingNodeTestCase as\
ProcessingNodeTestCaseModule ProcessingNodeTestCaseModule
ProcessingNodeTestCaseModule.patchActivityTool = lambda: None ProcessingNodeTestCaseModule.patchActivityTool = lambda: None
class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase): class ERP5TypeLiveTestCase(ERP5TypeTestMixin):
"""ERP5TypeLiveTestCase is the default class for *all* tests """ERP5TypeLiveTestCase is the default class for *all* tests
in ERP5. It is designed with the idea in mind that tests should 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 be run through the web. Command line based tests may be helpful
...@@ -73,18 +74,6 @@ class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -73,18 +74,6 @@ class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase):
defined. 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): def getPortalName(self):
""" Return the default ERP5 site id. """ Return the default ERP5 site id.
""" """
...@@ -102,47 +91,18 @@ class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -102,47 +91,18 @@ class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase):
getPortalObject = getPortal getPortalObject = getPortal
def login(self, user_name='ERP5TypeTestCase', quiet=0): #def logout(self):
""" # PortalTestCase.logout(self)
Most of the time, we need to login before doing anything # # clean up certain cache related REQUEST keys that might be associated
""" # # with the logged in user
PortalTestCase.login(self, user_name) # for key in ('_ec_cache', '_oai_cache'):
# pass
def logout(self): # #self.REQUEST.other.pop(key, None) # XXX
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): def _close(self):
'''Closes the ZODB connection.''' '''Closes the ZODB connection.'''
transaction.abort() 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): def _setup(self):
'''Configures the portal. Framework authors may '''Configures the portal. Framework authors may
override. override.
...@@ -188,171 +148,6 @@ class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -188,171 +148,6 @@ class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase):
''' '''
pass 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): def tearDown(self):
'''Tears down the fixture. Do not override, '''Tears down the fixture. Do not override,
use the hooks instead. use the hooks instead.
...@@ -365,85 +160,6 @@ class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -365,85 +160,6 @@ class ERP5TypeLiveTestCase(ProcessingNodeTestCase, PortalTestCase):
""" """
PortalTestCase.beforeClose(self) 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(test_list, verbosity=1, stream=None, **kw): def runLiveTest(test_list, verbosity=1, stream=None, **kw):
from Products.ERP5Type.tests.runUnitTest import DebugTestResult from Products.ERP5Type.tests.runUnitTest import DebugTestResult
from Products.ERP5Type.tests.runUnitTest import ERP5TypeTestLoader from Products.ERP5Type.tests.runUnitTest import ERP5TypeTestLoader
......
product/ERP5Type/tests/ERP5TypeTestCase.py
View file @ ed4e67bc
...@@ -274,10 +274,8 @@ def profile_if_environ(environment_var_name): ...@@ -274,10 +274,8 @@ def profile_if_environ(environment_var_name):
# No profiling, return identity decorator # No profiling, return identity decorator
return lambda self, method: method return lambda self, method: method
class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): class ERP5TypeTestMixin(ProcessingNodeTestCase, PortalTestCase):
"""TestCase for ERP5 based tests. """Mixin class for ERP5 based tests.
This TestCase setups an ERP5Site and installs business templates.
""" """
def shortDescription(self): def shortDescription(self):
...@@ -287,10 +285,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -287,10 +285,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
description += ', ' + doc.split("\n")[0].strip() description += ', ' + doc.split("\n")[0].strip()
return description return description
def dummy_test(self):
ZopeTestCase._print('All tests are skipped when --save option is passed '
'with --update_business_templates or without --load')
def getRevision(self): def getRevision(self):
erp5_path = os.path.join(instancehome, 'Products', 'ERP5') erp5_path = os.path.join(instancehome, 'Products', 'ERP5')
try: try:
...@@ -304,6 +298,340 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -304,6 +298,340 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
""" """
return str(self.__class__) return str(self.__class__)
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'):
self.REQUEST.other.pop(key, None)
def _setupUser(self):
'''Creates the default user.'''
uf = self.portal.acl_users
# do nothing if the user already exists
if not uf.getUser(user_name):
uf._doAddUser(user_name, 'secret', ['Member'], [])
def getDefaultSitePreferenceId(self):
"""Default id, usefull method to override
"""
return "default_site_preference"
# 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 _addPropertySheet(self, portal_type_name,
property_sheet_name='TestPropertySheet',
property_sheet_code=None):
"""Utility method to add a property sheet to a type information.
You might be interested in the higer level method _addProperty
This method registers all added property sheets, to be able to remove
them in tearDown.
"""
# install the 'real' class tool
class_tool = self.getClassTool()
if property_sheet_code is not None:
class_tool.newPropertySheet(property_sheet_name)
# XXX need to commit the transaction at this point, because class tool
# files are no longer available to the current transaction.
transaction.commit()
class_tool.editPropertySheet(property_sheet_name, property_sheet_code)
transaction.commit()
class_tool.importPropertySheet(property_sheet_name)
# We set the property sheet on the portal type
types_tool = self.getTypesTool()
ti = types_tool.getTypeInfo(portal_type_name)
property_sheet_set = set(ti.getTypePropertySheetList())
property_sheet_set.add(property_sheet_name)
ti._setTypePropertySheetList(list(property_sheet_set))
# remember that we added a property sheet for tear down
self._added_property_sheets.setdefault(
portal_type_name, []).append(property_sheet_name)
# reset aq_dynamic cache
types_tool.resetDynamicDocuments()
def getRule(self, **kw):
return self.portal.portal_rules.searchFolder(
sort_on='version', sort_order='descending', **kw)[0].getObject()
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.
"""
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)
self.assertEquals(set(a), set(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 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()
getPortalObject = getPortal
# 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 logMessage(self, message):
"""
Shortcut function to log a message
"""
ZopeTestCase._print('\n%s ' % message)
LOG('Testing ... ', DEBUG, message)
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)
class ERP5TypeTestCase(ERP5TypeTestMixin):
"""TestCase for ERP5 based tests.
This TestCase setups an ERP5Site and installs business templates.
"""
def dummy_test(self):
ZopeTestCase._print('All tests are skipped when --save option is passed '
'with --update_business_templates or without --load')
def getPortalName(self): def getPortalName(self):
""" """
Return the name of a portal for this test case. Return the name of a portal for this test case.
...@@ -337,8 +665,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -337,8 +665,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
setSite(portal) setSite(portal)
return portal return portal
getPortalObject = getPortal
def _app(self): def _app(self):
'''Opens a ZODB connection and returns the app object. '''Opens a ZODB connection and returns the app object.
...@@ -366,49 +692,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -366,49 +692,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
""" """
return 0 return 0
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'):
self.REQUEST.other.pop(key, None)
def _setupUser(self):
'''Creates the default user.'''
uf = self.portal.acl_users
# do nothing if the user already exists
if not uf.getUser(user_name):
uf._doAddUser(user_name, 'secret', ['Member'], [])
# 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)
@staticmethod @staticmethod
def _getBTPathAndIdList(template_list): def _getBTPathAndIdList(template_list):
bootstrap_path = os.environ.get('erp5_tests_bootstrap_path') or \ bootstrap_path = os.environ.get('erp5_tests_bootstrap_path') or \
...@@ -479,313 +762,112 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -479,313 +762,112 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
if uninstalled_list: if uninstalled_list:
getattr(portal, 'ERP5Site_updateTranslationTable', lambda: None)() getattr(portal, 'ERP5Site_updateTranslationTable', lambda: None)()
self.stepTic() self.stepTic()
return uninstalled_list return uninstalled_list
def setUp(self):
'''Sets up the fixture. Do not override,
use the hooks instead.
'''
from Products.CMFActivity.ActivityRuntimeEnvironment import BaseMessage
# Activities in unit tests shall never fail.
# Let's be a litte tolerant for the moment.
BaseMessage.max_retry = property(lambda self:
self.activity_kw.get('max_retry', 1))
use_dummy_mail_host = os.environ.get('use_dummy_mail_host', 0)
template_list = self.getBusinessTemplateList()
erp5_catalog_storage = os.environ.get('erp5_catalog_storage',
'erp5_mysql_innodb_catalog')
update_business_templates = os.environ.get('update_business_templates') is not None
erp5_load_data_fs = int(os.environ.get('erp5_load_data_fs', 0))
if update_business_templates and erp5_load_data_fs:
update_only = os.environ.get('update_only', None)
template_list = (erp5_catalog_storage, 'erp5_core',
'erp5_xhtml_style') + tuple(template_list)
# Update only specified business templates, regular expression
# can be used.
if update_only is not None:
update_only_list = update_only.split(',')
matching_template_list = []
# First parse the template list in order to keep same order
for business_template in template_list:
for expression in update_only_list:
if re.search(expression, business_template):
matching_template_list.append(business_template)
template_list = matching_template_list
# keep a mapping type info name -> property sheet list, to remove them in
# tear down.
self._added_property_sheets = {}
light_install = self.enableLightInstall()
create_activities = self.enableActivityTool()
hot_reindexing = self.enableHotReindexing()
self.setUpERP5Site(business_template_list=template_list,
light_install=light_install,
create_activities=create_activities,
quiet=install_bt5_quiet,
hot_reindexing=hot_reindexing,
erp5_catalog_storage=erp5_catalog_storage,
use_dummy_mail_host=use_dummy_mail_host)
PortalTestCase.setUp(self)
def afterSetUp(self):
'''Called after setUp() has completed. This is
far and away the most useful hook.
'''
pass
def getBusinessTemplateList(self):
"""
You must override this. Return the list of business templates.
"""
return ()
def logMessage(self, message):
"""
Shortcut function to log a message
"""
ZopeTestCase._print('\n%s ' % message)
LOG('Testing ... ', DEBUG, message)
def _updateConnectionStrings(self):
"""Update connection strings with values passed by the testRunner
"""
# update connection strings
for connection_string_name, connection_string in\
_getConnectionStringDict().items():
connection_name = connection_string_name.replace('_string', '')
getattr(self.portal, connection_name).edit('', connection_string)
def _setUpDummyMailHost(self):
"""Replace Original Mail Host by Dummy Mail Host.
"""
if 'MailHost' in self.portal.objectIds():
self.portal.manage_delObjects(['MailHost'])
self.portal._setObject('MailHost', DummyMailHost('MailHost'))
def _updateConversionServerConfiguration(self):
"""Update conversion server (Oood) at default site preferences.
"""
conversion_dict = _getConversionServerDict()
preference = self.portal.portal_preferences[
self.getDefaultSitePreferenceId()]
preference._setPreferredOoodocServerAddress(conversion_dict['hostname'])
preference._setPreferredOoodocServerPortNumber(conversion_dict['port'])
def getDefaultSitePreferenceId(self):
"""Default id, usefull method to override
"""
return "default_site_preference"
def _recreateCatalog(self, quiet=0):
"""Clear activities and catalog and recatalog everything.
Test runner can set `erp5_tests_recreate_catalog` environnement variable,
in that case we have to clear catalog. """
if int(os.environ.get('erp5_tests_recreate_catalog', 0)):
try:
_start = time.time()
if not quiet:
ZopeTestCase._print('\nRecreating catalog ... ')
portal = self.getPortal()
portal.portal_activities.manageClearActivities()
portal.portal_catalog.manage_catalogClear()
transaction.commit()
portal.ERP5Site_reindexAll()
transaction.commit()
self.tic()
if not quiet:
ZopeTestCase._print('done (%.3fs)\n' % (time.time() - _start,))
finally:
os.environ['erp5_tests_recreate_catalog'] = '0'
# 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 _addPropertySheet(self, portal_type_name,
property_sheet_name='TestPropertySheet',
property_sheet_code=None):
"""Utility method to add a property sheet to a type information.
You might be interested in the higer level method _addProperty
This method registers all added property sheets, to be able to remove
them in tearDown.
"""
# install the 'real' class tool
class_tool = self.getClassTool()
if property_sheet_code is not None:
class_tool.newPropertySheet(property_sheet_name)
# XXX need to commit the transaction at this point, because class tool
# files are no longer available to the current transaction.
transaction.commit()
class_tool.editPropertySheet(property_sheet_name, property_sheet_code)
transaction.commit()
class_tool.importPropertySheet(property_sheet_name)
# We set the property sheet on the portal type def setUp(self):
types_tool = self.getTypesTool() '''Sets up the fixture. Do not override,
ti = types_tool.getTypeInfo(portal_type_name) use the hooks instead.
property_sheet_set = set(ti.getTypePropertySheetList()) '''
property_sheet_set.add(property_sheet_name) from Products.CMFActivity.ActivityRuntimeEnvironment import BaseMessage
ti._setTypePropertySheetList(list(property_sheet_set)) # Activities in unit tests shall never fail.
# Let's be a litte tolerant for the moment.
BaseMessage.max_retry = property(lambda self:
self.activity_kw.get('max_retry', 1))
# remember that we added a property sheet for tear down use_dummy_mail_host = os.environ.get('use_dummy_mail_host', 0)
self._added_property_sheets.setdefault( template_list = self.getBusinessTemplateList()
portal_type_name, []).append(property_sheet_name) erp5_catalog_storage = os.environ.get('erp5_catalog_storage',
# reset aq_dynamic cache 'erp5_mysql_innodb_catalog')
types_tool.resetDynamicDocuments() update_business_templates = os.environ.get('update_business_templates') is not None
erp5_load_data_fs = int(os.environ.get('erp5_load_data_fs', 0))
if update_business_templates and erp5_load_data_fs:
update_only = os.environ.get('update_only', None)
template_list = (erp5_catalog_storage, 'erp5_core',
'erp5_xhtml_style') + tuple(template_list)
# Update only specified business templates, regular expression
# can be used.
if update_only is not None:
update_only_list = update_only.split(',')
matching_template_list = []
# First parse the template list in order to keep same order
for business_template in template_list:
for expression in update_only_list:
if re.search(expression, business_template):
matching_template_list.append(business_template)
template_list = matching_template_list
def getRule(self, **kw): # keep a mapping type info name -> property sheet list, to remove them in
return self.portal.portal_rules.searchFolder( # tear down.
sort_on='version', sort_order='descending', **kw)[0].getObject() self._added_property_sheets = {}
light_install = self.enableLightInstall()
create_activities = self.enableActivityTool()
hot_reindexing = self.enableHotReindexing()
self.setUpERP5Site(business_template_list=template_list,
light_install=light_install,
create_activities=create_activities,
quiet=install_bt5_quiet,
hot_reindexing=hot_reindexing,
erp5_catalog_storage=erp5_catalog_storage,
use_dummy_mail_host=use_dummy_mail_host)
PortalTestCase.setUp(self)
def validateRules(self): def afterSetUp(self):
""" '''Called after setUp() has completed. This is
try to validate all rules in rule_tool. far and away the most useful hook.
""" '''
rule_tool = self.getRuleTool() pass
for rule in rule_tool.contentValues(
portal_type=rule_tool.getPortalRuleTypeList()):
if rule.getValidationState() != 'validated':
rule.validate()
def createSimpleUser(self, title, reference, function): def getBusinessTemplateList(self):
""" """
Helper function to create a Simple ERP5 User. You must override this. Return the list of business templates.
User password is the reference.
""" """
user = self.createUser(reference, person_kw=dict(title=title)) return ()
assignment = self.createUserAssignement(user, assignment_kw=dict(function=function))
return user
def createUser(self, reference, password=None, person_kw=None): def _updateConnectionStrings(self):
""" """Update connection strings with values passed by the testRunner
Create an ERP5 User.
Default password is the reference.
person_kw is passed as additional arguments when creating the person
""" """
if password is None: # update connection strings
password = reference for connection_string_name, connection_string in\
if person_kw is None: _getConnectionStringDict().items():
person_kw = dict() connection_name = connection_string_name.replace('_string', '')
getattr(self.portal, connection_name).edit('', connection_string)
person = self.portal.person_module.newContent(portal_type='Person',
reference=reference,
password=password,
**person_kw)
return person
def createUserAssignment(self, user, assignment_kw): def _setUpDummyMailHost(self):
""" """Replace Original Mail Host by Dummy Mail Host.
Create an assignment to user.
""" """
assignment = user.newContent(portal_type='Assignment', **assignment_kw) if 'MailHost' in self.portal.objectIds():
assignment.open() self.portal.manage_delObjects(['MailHost'])
return assignment self.portal._setObject('MailHost', DummyMailHost('MailHost'))
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)
self.assertEquals(set(a), set(b), msg)
assertSameSet = failIfDifferentSet
def assertWorkflowTransitionFails(self, object, workflow_id, transition_id, def _updateConversionServerConfiguration(self):
error_message=None, state_variable='simulation_state'): """Update conversion server (Oood) at default site preferences.
"""
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() conversion_dict = _getConversionServerDict()
reference_history_length = len(workflow_tool.getInfoFor(ob=object, name='history', wf_id=workflow_id)) preference = self.portal.portal_preferences[
state_method = 'get' + convertToUpperCase(state_variable) self.getDefaultSitePreferenceId()]
method = getattr(object, state_method, None) preference._setPreferredOoodocServerAddress(conversion_dict['hostname'])
reference_workflow_state = method() preference._setPreferredOoodocServerPortNumber(conversion_dict['port'])
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) def _recreateCatalog(self, quiet=0):
self.assertEqual(len(workflow_history), reference_history_length + 1) """Clear activities and catalog and recatalog everything.
workflow_error_message = str(workflow_history[-1]['error_message']) Test runner can set `erp5_tests_recreate_catalog` environnement variable,
if error_message is not None: in that case we have to clear catalog. """
self.assertEqual(workflow_error_message, error_message) if int(os.environ.get('erp5_tests_recreate_catalog', 0)):
self.assertEqual(method(), reference_workflow_state) try:
return workflow_error_message _start = time.time()
if not quiet:
ZopeTestCase._print('\nRecreating catalog ... ')
portal = self.getPortal()
portal.portal_activities.manageClearActivities()
portal.portal_catalog.manage_catalogClear()
transaction.commit()
portal.ERP5Site_reindexAll()
transaction.commit()
self.tic()
if not quiet:
ZopeTestCase._print('done (%.3fs)\n' % (time.time() - _start,))
finally:
os.environ['erp5_tests_recreate_catalog'] = '0'
def _installBusinessTemplateList(self, business_template_list, def _installBusinessTemplateList(self, business_template_list,
light_install=True, light_install=True,
...@@ -1003,27 +1085,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -1003,27 +1085,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
break break
PortalTestCase.tearDown(self) PortalTestCase.tearDown(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 importObjectFromFile(self, container, relative_path, **kw): def importObjectFromFile(self, container, relative_path, **kw):
"""Import an object from a file located in $TESTFILEDIR/input/""" """Import an object from a file located in $TESTFILEDIR/input/"""
test_path = os.path.dirname(__file__) test_path = os.path.dirname(__file__)
...@@ -1033,63 +1094,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -1033,63 +1094,6 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
obj.manage_afterClone(obj) obj.manage_afterClone(obj)
return obj return obj
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)
from Products.ERP5 import ERP5Site from Products.ERP5 import ERP5Site
ERP5Site.getBootstrapBusinessTemplateUrl = lambda bt_title: \ ERP5Site.getBootstrapBusinessTemplateUrl = lambda bt_title: \
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7