TestEGovMixin.py

##############################################################################
#
# Copyright (c) 2008 Nexedi SARL and Contributors. All Rights Reserved.
#                  Fabien Morin <fabien@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from Testing.ZopeTestCase.PortalTestCase import PortalTestCase
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.SecurityTestCase import SecurityTestCase
from AccessControl.SecurityManagement import getSecurityManager
from Products.ERP5Type.tests.utils import DummyMailHost
from AccessControl import Unauthorized
from Testing import ZopeTestCase
from Products.ERP5Type.tests.Sequence import Step, Sequence, SequenceList
from zLOG import LOG
import transaction
import random
import email
from email.header import decode_header, make_header
from email.utils import parseaddr

class TestEGovMixin(SecurityTestCase):
  """Usefull methods for eGov Unit Tests."""
  
  # define all username corresponding to all roles used in eGov
  assignor_login = 'chef'
  assignee_login = 'agent'
  assignee_login_2 = 'agent_2'
  associate_login = 'agent_requested'

  organisation_1_login = 'societe_a'
  organisation_2_login = 'societe_b'

  all_username_list = ( assignor_login,
                        assignee_login,
                        assignee_login_2,
                        #associate_login,
                        organisation_1_login,
                        organisation_2_login)

  all_role_list = ( 'Manager',
                    'Assignor',
                    'Assignee',
                    'Author',
                    'Associate',
                    'Auditor',)

  #Permissions
  VIEW = 'View'
  ACCESS = 'Access contents information'
  ADD = 'Add portal content'
  MODIFY = 'Modify portal content'
  DELETE = 'Delete objects'


  # use modified method to render a more verbose output
  def play(self, context, sequence=None, sequence_number=0, quiet=0):
    if sequence is None:
      for idx, step in enumerate(self._step_list):
        step.play(context, sequence=self, quiet=quiet)
        # commit transaction after each step
        transaction.commit()
  Sequence.play = play

  def play(self, context, sequence=None, quiet=0):
    method_name = 'step' + self._method_name
    method = getattr(context,method_name)
    # We can in same cases replay many times the same step,
    # or not playing it at all
    nb_replay = random.randrange(0,self._max_replay+1)
    if self._required:
      if nb_replay==0:
        nb_replay=1
    for i in range(0,nb_replay):
      if not quiet:
        ZopeTestCase._print('\n  Playing step %s' % self._method_name)
        ZopeTestCase._print('\n    -> %s' % method.__doc__)
        LOG('Step.play', 0, '  Playing step %s' % self._method_name)
        LOG('Step.play', 0, '    -> %s' % method.__doc__)
      method(sequence=sequence)
  Step.play = play

  def playSequence(self, sequence_string, quiet=0) :
    ZopeTestCase._print('\n\n\n---------------------------------------------------------------------')
    ZopeTestCase._print('\nStarting New Sequence %s :' % self._TestCase__testMethodName)
    ZopeTestCase._print('\n * %s... \n' % self._TestCase__testMethodDoc)
    LOG('Sequence.play', 0, 'Starting New Sequence %s :' % self._TestCase__testMethodName)
    LOG('Sequence.play', 0, ' * %s... \n' % self._TestCase__testMethodDoc)
    sequence_list = SequenceList()
    sequence_list.addSequenceString(sequence_string)
    sequence_list.play(self, quiet=quiet)

  def getBusinessTemplateList(self):
    """return list of business templates to be installed. """
    return ( 'erp5_base',)

  def afterSetUp(self):
    """
      Method called before the launch of the test to initialize some data
    """
    self.createManagerAndLogin()

    # add a dummy mailhost not to send real messages
    if 'MailHost' in self.portal.objectIds():
      self.portal.manage_delObjects(['MailHost'])
    self.portal._setObject('MailHost', DummyMailHost('MailHost'))

    # remove all message in the message_table because
    # the previous test might have failed
    message_list = self.getPortal().portal_activities.getMessageList()
    for message in message_list:
      self.getPortal().portal_activities.manageCancel(message.object_path,
                                                      message.method_id)
    self.createUsers()
    self.createOrganisations()

    # XXX quick hack not to have mysql database pre-fill.
    self.portal.__class__.DeclarationTVA_zGetSIGTASInformation \
        = lambda x,**kw: []

    transaction.commit()
    self.tic()
    
  def beforeTearDown(self):
    """Clean up."""
    for module in self.portal.objectValues(spec=('ERP5 Folder',)):
      # we want to keep some IDs
      module.manage_delObjects([x for x in module.objectIds()
                                if x not in ('EUR',)])
    transaction.commit()
    self.tic()

  def getUserFolder(self) :
    return getattr(self.getPortal(), 'acl_users', None)

  loginAsUser = PortalTestCase.login

  diff_list = lambda self,x,y: [i for i in x if i not in y] 

  def createManagerAndLogin(self):
    """
      Create a simple user in user_folder with manager rights.
      This user will be used to initialize data in the method afterSetup
    """
    self.getUserFolder()._doAddUser('manager', 'manager', self.all_role_list, 
                                    [])
    self.login('manager')
  
  def createOneUser(self, username, function=None, group=None):
     """Create one person that will be users."""
     person_module = self.getPersonModule()
     user = person_module.newContent(
                              portal_type='Person',
                              reference=username,
                              title=username,
                              id=username,
                              password='secret')
     assignment = user.newContent(portal_type='Assignment')
     if function is not None:
       assignment.setFunction(function)
       self.assertNotEqual(assignment.getFunctionValue(), None)
     if group is not None:
       assignment.setGroup(group)
       self.assertNotEqual(assignment.getGroupValue(), None)
     assignment.open()

  def createUsers(self):
    """Create persons that will be users."""
    module = self.getPersonModule()
    if len(module.getObjectIds()) == 0:
      # create users
      self.createOneUser(self.assignor_login, 'function/section/chef', 
          'group/dgid/di/cge')
      self.createOneUser(self.assignee_login, 'function/impots/inspecteur', 
          'group/dgid/di/cge')
      self.createOneUser(self.assignee_login_2, 'function/impots/inspecteur', 
          'group/dgid/di/cge')
      self.createOneUser(self.associate_login, 'function/section/chef', 
          'group/dgid/di/csf/bf')

      # make this available to catalog
      transaction.commit()
      self.tic()

  def createOneOrganisation(self, username, role=None, function=None, 
                            group=None):
    """Create one organisation that will be user."""
    organisation_module = self.getOrganisationModule()
    user = organisation_module.newContent(
                             portal_type='Organisation',
                             title=username,
                             id=username,
                             reference=username,
                             password='secret')
    user.setRole(role)
    user.setFunction(function)
    user.setGroup(group)

    self.assertEqual(user.getRole(), role)
    self.assertEqual(user.getFunction(), function)
    self.assertEqual(user.getGroup(), group)
    self.assertEqual(user.getReference(), username)
  
  def createOrganisations(self):
    """Create organisations that will be users."""
    module = self.getOrganisationModule()
    if len(module.getObjectIds()) == 0:
      self.createOneOrganisation(self.organisation_1_login, 
          role='entreprise/siege')
      self.createOneOrganisation(self.organisation_2_login, 
          role='entreprise/siege')

      # make this available to catalog
      transaction.commit()
      self.tic()

  def checkRights(self, object_list, security_mapping, username):
    self.loginAsUser(username)
    user = getSecurityManager().getUser()
    if type(object_list) != type([]):
      object_list = [object_list,]
    for object in object_list:
      for permission, has in security_mapping.items():
        if user.has_permission(permission, object) and not has:
          self.fail('%s Permission should be Unauthorized on %s' % \
                                                ( permission,
                                                  object.getRelativeUrl()))
        if not(user.has_permission(permission, object)) and has:
          self.fail('%s Permission should be Authorized on %s' % \
                                                ( permission,
                                                  object.getRelativeUrl()))

  def checkTransition(self, object_list, possible_transition_list, 
                      not_possible_transition_list, username):
    
    if type(object_list) != type([]):
      object_list = [object_list,]
    for object in object_list:
      for transition in possible_transition_list:
        self.failUnlessUserCanPassWorkflowTransition(username, transition, 
                                                     object)
      for transition in not_possible_transition_list:
        self.failIfUserCanPassWorkflowTransition(username, transition, object)

  # Copied from ERP5Type/patches/CMFMailIn.py
  def decode_email(self, file):
    # Prepare result
    theMail = {
      'attachment_list': [],
      'body': '',
      # Place all the email header in the headers dictionary in theMail
      'headers': {}
    }
    # Get Message
    msg = email.message_from_string(file)
    # Back up original file
    theMail['__original__'] = file
    # Recode headers to UTF-8 if needed
    for key, value in msg.items():
      decoded_value_list = decode_header(value)
      unicode_value = make_header(decoded_value_list)
      new_value = unicode_value.__unicode__().encode('utf-8')
      theMail['headers'][key.lower()] = new_value
    # Filter mail addresses
    for header in ('resent-to', 'resent-from', 'resent-cc', 'resent-sender',
                   'to', 'from', 'cc', 'sender', 'reply-to'):
      header_field = theMail['headers'].get(header)
      if header_field:
          theMail['headers'][header] = parseaddr(header_field)[1]
    # Get attachments
    body_found = 0
    for part in msg.walk():
      content_type = part.get_content_type()
      file_name = part.get_filename()
      # multipart/* are just containers
      # XXX Check if data is None ?
      if content_type.startswith('multipart'):
        continue
      # message/rfc822 contains attached email message
      # next 'part' will be the message itself
      # so we ignore this one to avoid doubling
      elif content_type == 'message/rfc822':
        continue
      elif content_type in ("text/plain", "text/html"):
        charset = part.get_content_charset()
        payload = part.get_payload(decode=True)
        #LOG('CMFMailIn -> ',0,'charset: %s, payload: %s' % (charset,payload))
        if charset:
          payload = unicode(payload, charset).encode('utf-8')
        if body_found:
          # Keep the content type
          theMail['attachment_list'].append((file_name,
                                             content_type, payload))
        else:
          theMail['body'] = payload
          body_found = 1
      else:
        payload = part.get_payload(decode=True)
        # Keep the content type
        theMail['attachment_list'].append((file_name, content_type,
                                           payload))
    return theMail

  def _assertUserExists(self, login, password):
    """Checks that a user with login and password exists and can log in to the
    system.
    """
    from Products.PluggableAuthService.interfaces.plugins import\
                                                      IAuthenticationPlugin
    uf = self.getUserFolder()
    self.assertNotEquals(uf.getUserById(login, None), None)
    for plugin_name, plugin in uf._getOb('plugins').listPlugins(
                                IAuthenticationPlugin ):
      if plugin.authenticateCredentials(
                  {'login':login, 'password':password}) is not None:
        break
    else:
      self.fail("No plugin could authenticate '%s' with password '%s'" %
              (login, password))

  def checkWorklist(self, portal_type, count, validation_state, login):
    '''
      check that there is 'count' item in the worklist for 'portal_type' and
      'validation_state' logged with 'login'
    '''

    # save previous user
    previous_user = str(getSecurityManager().getUser())
    self.loginAsUser(login)

    worklist_dict = self.portal.getPortalTypeWorklistDictForWorkflow(\
        self.portal,
        workflow_list=['egov_universal_workflow', 'egov_anonymous_workflow'])
    self.assertNotEquals(worklist_dict, {})
    self.assertEquals(worklist_dict.has_key(portal_type), True)
    portal_type_dict = worklist_dict[portal_type]
    self.assertEquals(portal_type_dict.has_key(validation_state), True)
    self.assertEquals(portal_type_dict[validation_state]['count'], count)

    # relog with previous user
    if previous_user in ('Anonymous User', 'ERP5TypeTestCase'):
      self.logout()
    else:
      self.loginAsUser(previous_user)