# -*- coding: utf-8 -*-
# Derived from PloneTestCase in Plone.

#
# ERP5TypeTestCase
#

__version__ = '0.3.0'

import base64
import errno
import httplib
import md5
import os
import random
import re
import socket
import shutil
import sys
import time
import traceback
import urllib
from cStringIO import StringIO
from cPickle import dumps
from glob import glob
from warnings import warn
from ZTUtils import make_query

# XXX make sure that get_request works.
import Products.ERP5Type.Utils
from Products.ERP5Type import Globals

# store a copy of the original method
original_get_request = Globals.get_request
convertToUpperCase = Products.ERP5Type.Utils.convertToUpperCase

from Testing.ZopeTestCase.connections import registry
def get_context():
  if registry:
    return registry._conns[-1]

def get_request():
  request = original_get_request()
  if request is not None:
    return request
  current_app = get_context()
  if current_app is not None:
    return current_app.REQUEST

Products.ERP5Type.Utils.get_request = get_request
Globals.get_request = get_request

try:
  from zope.site.hooks import setSite
except ImportError:
  # BACK: Zope 2.8. setSite is somewhere else, and we can't use it anyway
  # since ERP5Site is not yet an ISite. Remove once we drop support for 2.8
  def setSite(site=None):
    pass

try:
  import itools.zope
  itools.zope.get_context = get_context
except ImportError:
  pass

import transaction
from Testing import ZopeTestCase
from Testing.ZopeTestCase import PortalTestCase, user_name
from Products.CMFCore.utils import getToolByName
from Products.DCWorkflow.DCWorkflow import ValidationFailed
from Products.ERP5Type.Base import _aq_reset
from Products.ERP5Type.Accessor.Constant import PropertyGetter as ConstantGetter
from zLOG import LOG, DEBUG

from Products.ERP5Type.tests.backportUnittest import SetupSiteError
from Products.ERP5Type.tests.utils import DummyMailHost, parseListeningAddress

# Quiet messages when installing products
install_product_quiet = 1
# Quiet messages when installing business templates
install_bt5_quiet = 0

import OFS.Application
OFS.Application.import_products()

# Std Zope Products
ZopeTestCase.installProduct('ExtFile', quiet=install_product_quiet)
ZopeTestCase.installProduct('Photo', quiet=install_product_quiet)
ZopeTestCase.installProduct('Formulator', quiet=install_product_quiet)
ZopeTestCase.installProduct('FCKeditor', quiet=install_product_quiet)
ZopeTestCase.installProduct('ZSQLMethods', quiet=install_product_quiet)
ZopeTestCase.installProduct('ZMySQLDA', quiet=install_product_quiet)
ZopeTestCase.installProduct('ZSQLCatalog', quiet=install_product_quiet)
ZopeTestCase.installProduct('ZMailIn', quiet=install_product_quiet)
ZopeTestCase.installProduct('ZGDChart', quiet=install_product_quiet)
ZopeTestCase.installProduct('ZCTextIndex', quiet=install_product_quiet)
ZopeTestCase.installProduct('MailHost', quiet=install_product_quiet)
ZopeTestCase.installProduct('PageTemplates', quiet=install_product_quiet)
ZopeTestCase.installProduct('PythonScripts', quiet=install_product_quiet)
ZopeTestCase.installProduct('ExternalMethod', quiet=install_product_quiet)
ZopeTestCase.installProduct('Sessions', quiet=install_product_quiet)

try:
  from Testing.ZopeTestCase.layer import onsetup
  # On Zope 2.12, installProduct() is a delayed call, so imports that depend on
  # products being "initialize"d should only be called "on setup". The
  # decorator above delays calls to the decorated function until after Product
  # initialization.
  # Also, we need Five to wire all our CMF dependencies.
  ZopeTestCase.installProduct('Five', quiet=install_product_quiet)
except ImportError:
  # On Zope 2.8 the call does not have to be delayed as product installation
  # happens immediately
  def onsetup(decorated):
    return decorated

try:
  # Workaround iHotFix patch that doesn't work with
  # ZopeTestCase REQUESTs
  ZopeTestCase.installProduct('iHotfix', quiet=install_product_quiet)
  from Products import iHotfix
  from types import UnicodeType
  # revert monkey patchs from iHotfix
  iHotfix.get_request = get_request

  originalStringIO = iHotfix.originalStringIO
  class UnicodeSafeStringIO(originalStringIO):
    """StringIO like class which never fails with unicode."""
    def write(self, s):
      if isinstance(s, UnicodeType):
        s = s.encode('utf8', 'repr')
      originalStringIO.write(self, s)
  # iHotFix will patch PageTemplate StringIO with
  iHotfix.iHotfixStringIO = UnicodeSafeStringIO
except ImportError:
  pass
ZopeTestCase.installProduct('Localizer', quiet=install_product_quiet)
try:
  # Workaround Localizer >= 1.2 patch that doesn't work with
  # ZopeTestCase REQUESTs (it's the same as iHotFix
  from Products.Localizer import patches
  # revert monkey patches from Localizer
  patches.get_request = get_request
except ImportError:
  pass

try:
  from Products.Localizer import patches
  # originalStringIO has been removed from recent Localizer versions
  from Products.Localizer.patches import originalStringIO
  class UnicodeSafeStringIO(patches.originalStringIO):
    """StringIO like class which never fails with unicode."""
    def write(self, s):
      if isinstance(s, unicode):
        s = s.encode('utf8', 'repr')
      patches.originalStringIO.write(self, s)
  # Localizer will patch PageTemplate StringIO with
  patches.LocalizerStringIO = UnicodeSafeStringIO
except ImportError:
  pass

from Products.ERP5Type.tests.ProcessingNodeTestCase import \
  ProcessingNodeTestCase, patchActivityTool
onsetup(patchActivityTool)()

ZopeTestCase.installProduct('TimerService', quiet=install_product_quiet)

# CMF
ZopeTestCase.installProduct('CMFCore', quiet=install_product_quiet)
ZopeTestCase.installProduct('CMFDefault', quiet=install_product_quiet)
ZopeTestCase.installProduct('CMFTopic', quiet=install_product_quiet)
ZopeTestCase.installProduct('DCWorkflow', quiet=install_product_quiet)
ZopeTestCase.installProduct('CMFCalendar', quiet=install_product_quiet)

# Based on CMF
ZopeTestCase.installProduct('CMFPhoto', quiet=install_product_quiet)
ZopeTestCase.installProduct('BTreeFolder2', quiet=install_product_quiet)
ZopeTestCase.installProduct('CMFReportTool', quiet=install_product_quiet) # Not required by ERP5Type but required by ERP5Form
ZopeTestCase.installProduct('TranslationService', quiet=install_product_quiet)
ZopeTestCase.installProduct('PortalTransforms', quiet=install_product_quiet)
ZopeTestCase.installProduct('MimetypesRegistry', quiet=install_product_quiet)

# Security Stuff
ZopeTestCase.installProduct('PluggableAuthService', quiet=install_product_quiet)
ZopeTestCase.installProduct('ERP5Security', quiet=install_product_quiet)

# Debugging
ZopeTestCase.installProduct('VerboseSecurity', quiet=install_product_quiet)
ZopeTestCase.installProduct('Zelenium', quiet=install_product_quiet)

# ERP5 - ERP5Type product is installed last so that
#        initializeProductDocumentRegistry is only called
#        after all products which need to register their Document 
#        classes can register them by invoking updateGlobals in __init__
ZopeTestCase.installProduct('CMFActivity', quiet=install_product_quiet)
ZopeTestCase.installProduct('ERP5Catalog', quiet=install_product_quiet)
ZopeTestCase.installProduct('ERP5Form', quiet=install_product_quiet)
ZopeTestCase.installProduct('ERP5', quiet=install_product_quiet)
ZopeTestCase.installProduct('ERP5SyncML', quiet=install_product_quiet)
ZopeTestCase.installProduct('ERP5Type', quiet=install_product_quiet)
ZopeTestCase.installProduct('CMFCategory', quiet=install_product_quiet)
ZopeTestCase.installProduct('ZMySQLDDA', quiet=install_product_quiet)

ZopeTestCase.installProduct('ParsedXML', quiet=install_product_quiet)

# Install everything else which looks like related to ERP5
from OFS.Application import get_products
from App.config import getConfiguration

instancehome = getConfiguration().instancehome
for priority, product_name, index, product_dir in get_products():
  # XXX very heuristic
  if os.path.isdir(os.path.join(product_dir, product_name, 'Document')) \
     or os.path.isdir(os.path.join(product_dir, product_name, 'PropertySheet')) \
     or os.path.isdir(os.path.join(product_dir, product_name, 'Constraint')) \
     or os.path.isdir(os.path.join(product_dir, product_name, 'Tool')):
    ZopeTestCase.installProduct(product_name, quiet=install_product_quiet)

from AccessControl.SecurityManagement import newSecurityManager, noSecurityManager

from Acquisition import aq_base

portal_name = 'erp5_portal'

# we keep a reference to all sites for which setup failed the first time, to
# prevent replaying the same failing setup step for each test.
failed_portal_installation = {}

# have we installed business templates ?
# this is a mapping 'list of business template -> boolean
setup_done = {}

def _getConnectionStringDict():
  """Returns the connection strings used for this test.
  """
  connection_string_dict = {}
  default = os.environ.get('erp5_sql_connection_string')
  for connection in ('erp5_sql_connection_string',
                     'erp5_sql_deferred_connection_string',
                     # default value for transactionless is derived from value
                     # for cmf_activity, so process it last
                     'cmf_activity_sql_connection_string'):
    connection_string = os.environ.get(connection, default)
    if connection_string:
      connection_string_dict[connection] = connection_string
  connection = 'erp5_sql_transactionless_connection_string'
  if os.environ.get(connection, connection_string):
    connection_string_dict[connection] = \
      os.environ.get(connection, '-' + connection_string)
  return connection_string_dict

def _getConversionServerDict():
  """ Returns a dict with hostname and port for Conversion Server (Oood)
  """
  conversion_server_hostname = os.environ.get('conversion_server_hostname', 
                                              'localhost')
  conversion_server_port = os.environ.get('conversion_server_port',
                                          '8008')
  return dict(hostname=conversion_server_hostname, 
              port=int(conversion_server_port))


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 ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
    """TestCase for ERP5 based tests.

    This TestCase setups an ERP5Site and installs business templates.
    """

    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 dummy_test(self):
      ZopeTestCase._print('All tests are skipped when --save option is passed '
                          'with --update_business_templates or without --load')

    def getRevision(self):
      try:
        import pysvn
        return pysvn.Client().info('%s/Products/ERP5'
                    % os.environ['INSTANCE_HOME']).revision.number
      except:
        return None

    def getTitle(self):
      """Returns the title of the test, for test reports.
      """
      return str(self.__class__)

    def getPortalName(self):
      """
        Return the name of a portal for this test case.
        This is necessary for each test case to use a different portal built by
        different business templates.
        The test runner can set `erp5_tests_portal_id` environment variable
        to force a portal id.
      """
      forced_portal_id = os.environ.get('erp5_tests_portal_id')
      if forced_portal_id:
        return str(forced_portal_id)
      m = md5.new()
      m.update(repr(self.getBusinessTemplateList())+ self.getTitle())
      uid = m.hexdigest()

      return portal_name + '_' + uid

    def getPortal(self):
      """Returns the portal object, i.e. the "fixture root".

      It also does some initialization, as if the portal was accessed for the
      first time for the current request.
      For performance reason, this should be used in only 3 places:
      'setUpERP5Site', 'tic' and 'PortalTestCase._portal'
      """
      portal = self.app[self.getPortalName()]
      # Make sure skins are correctly set-up (it's not implicitly set up
      # by Acquisition on Zope 2.12 as it is on 2.8)
      portal.setupCurrentSkin(portal.REQUEST)
      self.REQUEST = portal.REQUEST
      setSite(portal)
      return portal

    getPortalObject = getPortal

    def _app(self):
      '''Opens a ZODB connection and returns the app object.
      
      We override it to patch HTTP_ACCEPT_CHARSET into REQUEST to get the zpt
      unicode conflict resolver to work properly'''
      app = PortalTestCase._app(self)
      app.REQUEST['HTTP_ACCEPT_CHARSET'] = 'utf-8'
      return app

    def enableLightInstall(self):
      """
      You can override this. Return if we should do a light install (1) or not (0)
      """
      return 1

    def enableActivityTool(self):
      """
      You can override this. Return if we should create (1) or not (0) an activity tool
      """
      return 1

    def enableHotReindexing(self):
      """
      You can override this. Return if we should create (1) or not (0) an activity tool
      """
      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
    def _getBTPathAndIdList(template_list):
      bootstrap_path = os.environ.get('erp5_tests_bootstrap_path') or \
        ERP5Site.getBootstrapDirectory()
      bt5_path = os.environ.get('erp5_tests_bt5_path')
      if bt5_path:
        bt5_path_list = bt5_path.split(',')
      else:
        bt5_path = os.path.join(os.environ['INSTANCE_HOME'], 'bt5')
        bt5_path_list = bt5_path, os.path.join(bt5_path, '*')

      def search(path, template):
        urltype, url = urllib.splittype(path + '/' + template)
        if urltype == 'http':
          host, selector = urllib.splithost(url)
          user_passwd, host = urllib.splituser(host)
          host = urllib.unquote(host)
          h = httplib.HTTP(host)
          h.putrequest('HEAD', selector)
          h.putheader('Host', host)
          if user_passwd:
            h.putheader('Authorization',
                        'Basic %s' % base64.b64encode(user_passwd).strip())
          h.endheaders()
          errcode, errmsg, headers = h.getreply()
          if errcode == 200:
            return urltype + ':' + url
        else:
          path_list = glob(os.path.join(path, template))
          if path_list:
            return path_list[0]

      not_found_list = []
      new_template_list = []
      for template in template_list:
        id = template.split('/')[-1]
        for path in bt5_path_list:
          path = search(path, template) or search(path, template + '.bt5')
          if path:
            break
        else:
          path = os.path.join(bootstrap_path, template)
          if not os.path.exists(path):
            not_found_list.append(template)
            continue
        new_template_list.append((path, id))
      if not_found_list:
        raise RuntimeError("Following BT can't be found on your system : %s"
                           % ', '.join(not_found_list))
      return new_template_list

    def manuallyInstallBusinessTemplate(self, *template_list):
      new_template_list = self._getBTPathAndIdList(template_list)
      light_install = self.enableLightInstall()
      self._installBusinessTemplateList(new_template_list,
                                        light_install=light_install)
      self.tic()

    def uninstallBusinessTemplate(self, *template_list):
      template_list = set(template_list)
      uninstalled_list = []
      portal = self.portal
      for bt in portal.portal_templates.getInstalledBusinessTemplateList():
        bt_title = bt.getTitle()
        if bt_title in template_list:
          bt.uninstall(remove_translations=True)
          uninstalled_list.append(bt_title)
      if uninstalled_list:
        getattr(portal, 'ERP5Site_updateTranslationTable', lambda: None)()
      self.stepTic()
      return uninstalled_list

    def setUp(self):
      '''Sets up the fixture. Do not override,
         use the hooks instead.
      '''
      # This is a workaround for the overwriting problem in Testing/__init__.py
      # in Zope.  So this overwrites them again to revert the changes made by
      # Testing.
      # XXX: Leo: Is this still true? We need to reevaluate how we get 
      # information from our environment.
      try:
        import App.config
      except ImportError:
        os.environ['INSTANCE_HOME'] = INSTANCE_HOME =\
                            os.environ['COPY_OF_INSTANCE_HOME']
        os.environ['SOFTWARE_HOME'] = SOFTWARE_HOME =\
                            os.environ['COPY_OF_SOFTWARE_HOME']
      else:
        cfg = App.config.getConfiguration()
        cfg.instancehome = os.environ['COPY_OF_INSTANCE_HOME']
        App.config.setConfiguration(cfg)

      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
      ti = self.getTypesTool().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
      _aq_reset()

    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 _installBusinessTemplateList(self, business_template_list,
                                     light_install=True,
                                     quiet=True):
      template_tool = self.portal.portal_templates
      update_business_templates = os.environ.get('update_business_templates') is not None
      BusinessTemplate_getModifiedObject = aq_base(
        getattr(self.portal, 'BusinessTemplate_getModifiedObject', None))

      # Add some business templates
      for url, bt_title in business_template_list:
        start = time.time()
        get_install_kw = False
        if bt_title in [x.getTitle() for x in template_tool.getInstalledBusinessTemplateList()]:
          if update_business_templates:
            if not quiet:
              ZopeTestCase._print('Updating %s business template ... ' % bt_title)
            if BusinessTemplate_getModifiedObject is not None:
              get_install_kw = True
          else:
            continue
        else:
          if not quiet:
            ZopeTestCase._print('Adding %s business template ... ' % bt_title)
        bt = template_tool.download(url)
        if not quiet:
          ZopeTestCase._print('(imported in %.3fs) ' % (time.time() - start))
        install_kw = None
        if get_install_kw:
          install_kw = {}
          listbox_object_list = BusinessTemplate_getModifiedObject.__of__(bt)()
          for listbox_line in listbox_object_list:
            install_kw[listbox_line.object_id] = listbox_line.choice_item_list[0][1]
        bt.install(light_install=light_install,
                   object_to_update=install_kw,
                   update_translation=1)
        # Release locks
        transaction.commit()
        if not quiet:
          ZopeTestCase._print('done (%.3fs)\n' % (time.time() - start))

    def setUpERP5Site(self,
                     business_template_list=(),
                     quiet=0,
                     light_install=1,
                     create_activities=1,
                     hot_reindexing=1,
                     erp5_catalog_storage='erp5_mysql_innodb_catalog',
                     use_dummy_mail_host=0):
      '''
        Creates an ERP5 site.
        business_template_list must be specified correctly
        (e.g. '("erp5_base", )').
      '''
      portal_name = self.getPortalName()
      title = self.getTitle()
      from Products.ERP5Type.Base import _aq_reset
      if portal_name in failed_portal_installation:
        raise SetupSiteError(
            'Installation of %s already failed, giving up' % portal_name)
      try:
        self.app = app = self._app()
        app.test_portal_name = portal_name

        global setup_done
        portal = app._getOb(portal_name, None)
        if portal is None or not setup_done.get(tuple(business_template_list)):
          setup_done[tuple(business_template_list)] = 1
          business_template_list = \
            self._getBTPathAndIdList(business_template_list)
          try:
            _start = time.time()
            # Add user and log in
            if not quiet:
              ZopeTestCase._print('\nAdding ERP5TypeTestCase user ... \n')
            uf = app.acl_users
            uf._doAddUser('ERP5TypeTestCase', '', ['Manager', 'Member', 'Assignee',
                          'Assignor', 'Author', 'Auditor', 'Associate'], [])
            user = uf.getUserById('ERP5TypeTestCase').__of__(uf)
            newSecurityManager(None, user)
            # Add ERP5 Site
            reindex = 1
            if hot_reindexing:
              setattr(app, 'isIndexable', 0)
              reindex = 0

            if portal is None:
              if not quiet:
                ZopeTestCase._print('Adding %s ERP5 Site ... ' % portal_name)

              extra_constructor_kw = _getConnectionStringDict()
              # manage_addERP5Site does not accept the following 2 arguments
              for k in ('erp5_sql_deferred_connection_string',
                        'erp5_sql_transactionless_connection_string'):
                extra_constructor_kw.pop(k, None)
              email_from_address = os.environ.get('email_from_address')
              if email_from_address is not None:
                extra_constructor_kw['email_from_address'] = email_from_address

              factory = app.manage_addProduct['ERP5']
              factory.manage_addERP5Site(portal_name,
                                       erp5_catalog_storage=erp5_catalog_storage,
                                       light_install=light_install,
                                       reindex=reindex,
                                       create_activities=create_activities,
                                       **extra_constructor_kw )
              sql = extra_constructor_kw.get('erp5_sql_connection_string')
              if sql:
                app[portal_name]._setProperty('erp5_site_global_id',
                                              base64.standard_b64encode(sql))
              if not quiet:
                ZopeTestCase._print('done (%.3fs)\n' % (time.time() - _start))
              # Release locks
              transaction.commit()
            self.portal = portal = self.getPortal()

            if len(setup_done) == 1: # make sure it is run only once
              try:
                from Products import DeadlockDebugger
              except ImportError:
                pass
              self.serverhost, self.serverport = self.startZServer()
              self._registerNode(distributing=1, processing=1)

            self._updateConnectionStrings()
            self._recreateCatalog()
            self._installBusinessTemplateList(business_template_list,
                                              light_install=light_install,
                                              quiet=quiet)
            self._updateConversionServerConfiguration()
            # Create a Manager user at the Portal level
            uf = self.getPortal().acl_users
            uf._doAddUser('ERP5TypeTestCase', '', ['Manager', 'Member', 'Assignee',
                            'Assignor', 'Author', 'Auditor', 'Associate'], [])
            user = uf.getUserById('ERP5TypeTestCase').__of__(uf)

            if use_dummy_mail_host:
              self._setUpDummyMailHost()

            setup_once = getattr(self, 'setUpOnce', None)
            if setup_once is not None and \
                   not getattr(portal, 'set_up_once_called', 0):
              portal.set_up_once_called = 1
              if not quiet:
                ZopeTestCase._print('Executing setUpOnce ... ')
                start = time.time()
              setup_once()
              if not quiet:
                ZopeTestCase._print('done (%.3fs)\n' % (time.time() - start))

            # Enable reindexing
            # Do hot reindexing # Does not work
            if hot_reindexing:
              setattr(app,'isIndexable', 1)
              portal.portal_catalog.manage_hotReindexAll()

            transaction.commit()
            self.tic(not quiet)

            # Reset aq dynamic, so all unit tests will start again
            _aq_reset()

            # Log out
            if not quiet:
              ZopeTestCase._print('Logout ... \n')
            noSecurityManager()
            if not quiet:
              ZopeTestCase._print('done (%.3fs)\n' % (time.time()-_start,))
              ZopeTestCase._print('Running Unit tests of %s\n' % title)
          except:
            transaction.abort()
            raise
          else:
            transaction.commit()
            del self.portal, self.app
            ZopeTestCase.close(app)
      except:
        f = StringIO()
        traceback.print_exc(file=f)
        ZopeTestCase._print(f.getvalue())
        f.close()
        failed_portal_installation[portal_name] = 1
        ZopeTestCase._print('Ran Unit test of %s (installation failed)\n'
                            % title) # run_unit_test depends on this string.
        raise

    def tearDown(self):
      '''Tears down the fixture. Do not override,
         use the hooks instead.
      '''
      PortalTestCase.tearDown(self)
      setSite() # undo site configuration from self.getPortal()

    def beforeClose(self):
      PortalTestCase.beforeClose(self)
      try:
        portal_activities = self.portal.portal_activities
        # Drop remaining activities.
        count = portal_activities.countMessage()
        portal_activities.manageClearActivities(keep=False)
        if count:
          LOG('Products.ERP5Type.tests.ERP5TypeTestCase.beforeClose', DEBUG,
              'dropped %d left-over activity messages' % (count,))
        transaction.commit()
      except AttributeError:
        pass

    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 copyInputFileToImportFolder(self, relative_path):
      """
      Copies a file located in $TESTFILEDIR/input/ to
      import/ folder of test instance and returns the
      full path.
      If files already exists, overwrites it.
      """
      test_path = os.path.dirname(__file__)

      source_path = os.path.join(test_path, 'input', relative_path)
      self.assertTrue(os.path.exists(source_path))

      import_path = os.path.join(instancehome, 'import')
      if not os.path.exists(import_path):
        if os.path.islink(import_path):
          # broken symlink
          os.unlink(import_path)
        os.mkdir(import_path)

      shutil.copy(source_path, import_path)
      return import_path

    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
ERP5Site.getBootstrapBusinessTemplateUrl = lambda bt_title: \
  ERP5TypeTestCase._getBTPathAndIdList((bt_title,))[0][0]


class ResponseWrapper:
    '''Decorates a response object with additional introspective methods.'''

    _headers_separator_re = re.compile('(?:\r?\n){2}')

    def __init__(self, response, outstream, path):
        self._response = response
        self._outstream = outstream
        self._path = path

    def __getattr__(self, name):
        return getattr(self._response, name)

    def getOutput(self):
        '''Returns the complete output, headers and all.'''
        return self._outstream.getvalue()

    def getBody(self):
        '''Returns the page body, i.e. the output par headers.'''
        output = self.getOutput()
        try:
            headers, body = self._headers_separator_re.split(output, 1)
            return body
        except ValueError:
            # not enough values to unpack: no body
            return None

    def getPath(self):
        '''Returns the path used by the request.'''
        return self._path

    def getHeader(self, name):
        '''Returns the value of a response header.'''
        return self.headers.get(name.lower())

    def getCookie(self, name):
        '''Returns a response cookie.'''
        return self.cookies.get(name)

class ERP5ReportTestCase(ERP5TypeTestCase):
  """Base class for testing ERP5 Reports
  """
  def getReportSectionList(self, context, report_name):
    """Get the list of report sections in a report called on context."""
    report = getattr(context, report_name)
    if hasattr(report, 'report_method'):
      report_method = getattr(context, report.report_method)
      return report_method()
    else:
      report_item_list = []
      for reportbox in [field for field in report.get_fields()
                        if field.getRecursiveTemplateField().meta_type == 'ReportBox']:
        report_item_list.extend(reportbox.render())
      return report_item_list

  def getListBoxLineList(self, report_section):
    """Render the listbox in a report section, return None if no listbox exists
    in the report_section.
    """
    result = None
    here = report_section.getObject(self.portal)
    report_section.pushReport(self.portal)
    form = getattr(here, report_section.getFormId())
    self.portal.REQUEST['here'] = here
    if form.has_field('listbox'):
      result = form.listbox.get_value('default',
                                      render_format='list',
                                      REQUEST=self.portal.REQUEST)
    report_section.popReport(self.portal)
    return result

  def checkLineProperties(self, line, **kw):
    """Check properties of a report line.
    """
    diff_list = []
    for k, v in kw.items():
      if v != line.getColumnProperty(k):
        diff_list.append('`%s`: expected: %r actual: %r' %
                                (k, v, line.getColumnProperty(k)))
    if diff_list:
      self.fail('Lines differs:\n' + '\n'.join(diff_list))

def dummy_setUp(self):
  '''
  This one is overloaded so that it does not execute beforeSetUp and afterSetUp
  from the original tests, which would write to the FileStorage when --save is
  enabled
  '''
  try:
      self.beforeSetUp()
      self.app = self._app()
      self.portal = self._portal()
      self._setup()
  except:
      self._clear()
      raise

def dummy_tearDown(self):
  '''
  This one is overloaded so that it does not execute beforeTearDown from
  the original tests, which would write to the FileStorage when --save
  is enabled
  '''
  self._clear(1)

class ZEOServerTestCase(ERP5TypeTestCase):
  """TestCase class to run a ZEO storage

  Main method is 'asyncore_loop' (inherited) since there is nothing to do
  except processing I/O.
  """

  def setUp(self):
    # Start ZEO storage and send address to parent process if any.
    from Zope2.custom_zodb import zeo_client, Storage
    from ZEO.StorageServer import StorageServer
    storage = {'1': Storage}
    for host_port in parseListeningAddress(os.environ.get('zeo_server')):
      try:
        self.zeo_server = StorageServer(host_port, storage)
        break
      except socket.error, e:
        if e[0] != errno.EADDRINUSE:
          raise
    if zeo_client:
      os.write(zeo_client, repr(host_port))
      os.close(zeo_client)
    ZopeTestCase._print("\nZEO Storage started at %s:%s ... " % host_port)

  def tearDown(self):
    self.zeo_server.close_server()


class lazy_func_prop(object):
  """Descriptor to delay the compilations of Python Scripts
     until some of their attributes are accessed.
  """
  default_dict = {}
  def __init__(self, name, default):
    self.name = name
    self.default_dict[name] = default
  def __get__(self, instance, owner):
    if self.name not in instance.__dict__:
      instance.__dict__.update(self.default_dict)
      instance._orig_compile()
    return instance.__dict__[self.name]
  def __set__(self, instance, value):
    instance.__dict__[self.name] = value
  def __delete__(self, instance):
    del instance.__dict__[self.name]

@onsetup
def optimize():
  '''Significantly reduces portal creation time.'''
  def __init__(self, text):
    # Don't compile expressions on creation
    self.text = text
  from Products.CMFCore.Expression import Expression
  Expression.__init__ = __init__

  # Delay the compilations of Python Scripts until they are really executed.
  from Products.PythonScripts.PythonScript import PythonScript
  # In the future, Python Scripts will be exported without those 2 attributes:
  PythonScript.func_code = lazy_func_prop('func_code', None)
  PythonScript.func_defaults = lazy_func_prop('func_defaults', None)

  PythonScript._orig_compile = PythonScript._compile
  def _compile(self):
    # mark the script as being not compiled
    for name in lazy_func_prop.default_dict:
      self.__dict__.pop(name, None)
  PythonScript._compile = _compile
  PythonScript_exec = PythonScript._exec
  def _exec(self, *args):
    self.func_code # trigger compilation if needed
    return PythonScript_exec(self, *args)
  PythonScript._exec = _exec
  from Acquisition import aq_parent
  def _makeFunction(self, dummy=0): # CMFCore.FSPythonScript uses dummy arg.
    self.ZCacheable_invalidate()
    self.__dict__.update(lazy_func_prop.default_dict)
    self._orig_compile()
    if not (aq_parent(self) is None or hasattr(self, '_filepath')):
      # It needs a _filepath, and has an acquisition wrapper.
      self._filepath = self.get_filepath()
  PythonScript._makeFunction = _makeFunction

  # XXX Previous implementation of this 'optimize' function requires that
  #     Python Scripts always contain up-to-date data for 'func_code' and
  #     'func_defaults' properties, so make sure we always export them.
  #     This compatibility code is not required for normal ERP5 instances
  #     because scripts are compiled at BT installation.
  from Products.ERP5Type.Document.BusinessTemplate import BaseTemplateItem
  BaseTemplateItem_removeProperties = BaseTemplateItem.removeProperties
  def removeProperties(self, obj, export):
    if export and isinstance(obj, PythonScript):
      obj.func_code # trigger compilation if needed
    return BaseTemplateItem_removeProperties(self, obj, export)
  BaseTemplateItem.removeProperties = removeProperties

  # Do not reindex portal types sub objects by default
  # We will probably disable reindexing for other types later
  full_indexing_set = set(os.environ.get('enable_full_indexing', '').split(','))
  if not 'portal_types' in full_indexing_set:
    from Products.ERP5Type.Document.RoleInformation import RoleInformation
    RoleInformation.isIndexable = ConstantGetter('isIndexable', value=False)
    from Products.ERP5Type.Document.ActionInformation import ActionInformation
    ActionInformation.isIndexable = ConstantGetter('isIndexable', value=False)

optimize()

@onsetup
def fortify():
  '''Add some extra checks that we don't have at runtime, not to slow down the
  system.
  '''
  # check that we don't store persistent objects in cache
  from Products.ERP5Type.CachePlugins.BaseCache import CacheEntry
  CacheEntry__init__ = CacheEntry.__init__
  def __init__(self, value, *args, **kw):
    # this will raise TypeError if you try to cache a persistent object
    dumps(value)
    CacheEntry__init__(self, value, *args, **kw)
  CacheEntry.__init__ = __init__
  # randomize priorities of activities in a deterministic way
  seed = os.environ.get("random_activity_priority")
  if seed is not None:
    ZopeTestCase._print("RNG seed for priorities of activities is %r\n" % seed)
    rng = random.Random(seed)
    from Products.CMFActivity.ActivityTool import Message
    Message__init__ = Message.__init__
    def __init__(self, obj, active_process, activity_kw, *args, **kw):
      activity_kw['priority'] = rng.randint(-128, 127)
      Message__init__(self, obj, active_process, activity_kw, *args, **kw)
    Message.__init__ = __init__


fortify()
