[Zope3-checkins] CVS: Zope3/src/zope/app/workflow/stateful/tests - __init__.py:1.1 test_contentworkflow.py:1.1 test_definition.py:1.1 test_instance.py:1.1 test_xmlimportexport.py:1.1

Ulrich Eck ueck@net-labs.de
Thu, 8 May 2003 13:27:21 -0400


Update of /cvs-repository/Zope3/src/zope/app/workflow/stateful/tests
In directory cvs.zope.org:/tmp/cvs-serv7538/src/zope/app/workflow/stateful/tests

Added Files:
	__init__.py test_contentworkflow.py test_definition.py 
	test_instance.py test_xmlimportexport.py 
Log Message:
Finally got it into Zope3:

Workflow has arrived!

this is a merge of the workflow package that was seperatly developed
at /Packages3/workflow.

please to a 

cvs update -dPA 

to ensure that old files/directories are deleted, otherwise you'll 
probably encounter errors when trying to run zope3

if you have problems .. send me an email ueck <at> net-labs.de

Ulrich


=== Added File Zope3/src/zope/app/workflow/stateful/tests/__init__.py ===


=== Added File Zope3/src/zope/app/workflow/stateful/tests/test_contentworkflow.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################

"""Stateful content workflow utility.

$Id: test_contentworkflow.py,v 1.1 2003/05/08 17:27:20 jack-e Exp $
"""
import unittest

from zope.interface import Interface
from zope.interface.verify import verifyClass

from zope.app.interfaces.workflow.stateful import IContentWorkflowsUtility
from zope.app.workflow.stateful.contentworkflow import ContentWorkflowsUtility



# XXX How to test this without fake a hole zope3-server ?????
class ContentWorkflowsUtilityTests(unittest.TestCase):

    def testInterface(self):
        verifyClass(IContentWorkflowsUtility,
                    ContentWorkflowsUtility)




def test_suite():
    suite = unittest.TestSuite()
    suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(
        ContentWorkflowsUtilityTests))
    return suite

if __name__ == '__main__':
    unittest.main()


=== Added File Zope3/src/zope/app/workflow/stateful/tests/test_definition.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################

"""Stateful workflow process definition.

$Id: test_definition.py,v 1.1 2003/05/08 17:27:20 jack-e Exp $
"""
import unittest

from zope.interface import Interface
from zope.interface.verify import verifyClass
from zope.schema import TextLine

from zope.app.interfaces.workflow.stateful import IStatefulProcessDefinition
from zope.app.workflow.stateful.definition import StatefulProcessDefinition


class IDummyState(Interface):
    """A really dummy state"""

class DummyState:
    __implements__ = IDummyState


class IDummyTransition(Interface):
    """A really dummy transition"""

class DummyTransition:
    __implements__ = IDummyTransition


class IDummyDataSchema(Interface):

    text = TextLine(title=u'a text', default=u'no text')
    

# XXX Tests missing for:
# State Class/Interface
# Transition Class/Interface


class StatefulProcessDefinitionTests(unittest.TestCase):

    def setUp(self):
        self.pd = pd = StatefulProcessDefinition()
        self.doMinimalSetup()

    def doMinimalSetup(self):
        pd = self.pd
        pd.setRelevantDataSchema(IDummyDataSchema)
        self.st1 = st1 = DummyState()
        self.st2 = st2 = DummyState()
        pd.addState('st1', st1)
        pd.addState('st2', st2)

        self.tr1 = tr1 = DummyTransition()
        self.tr2 = tr2 = DummyTransition()
        pd.addTransition('tr1', tr1)
        pd.addTransition('tr2', tr2)

    def testInterface(self):
        verifyClass(IStatefulProcessDefinition, StatefulProcessDefinition)

    def testGetSchema(self):
        self.assertEqual(self.pd.getRelevantDataSchema(), IDummyDataSchema)

    def testGetStateNames(self):
        pd = self.pd

        names = pd.getStateNames()

        self.failUnless('st1' in names)
        self.failUnless('st2' in names)
        self.failUnless(pd.getInitialStateName() in names)

    def testGetState(self):
        pd = self.pd

        st1 = pd.getState('st1')

        self.assertEqual(st1, self.st1)

    def testRemoveState(self):
        pd = self.pd

        pd.removeState('st1')
        names = pd.getStateNames()

        self.assertRaises(KeyError, pd.getState, 'st1')
        self.failIf('st1' in names)
        self.failUnless('st2' in names)
        self.failUnless(pd.getInitialStateName() in names)
        
    def testGetTransistionNames(self):
        pd = self.pd

        names = pd.getTransitionNames()

        self.failUnless('tr1' in names)
        self.failUnless('tr2' in names)

    def testGetTransation(self):
        pd = self.pd

        tr1 = pd.getTransition('tr1')

        self.assertEqual(tr1, self.tr1)

    def testRemoveTransistion(self):
        pd = self.pd

        pd.removeTransition('tr1')
        names = pd.getTransitionNames()

        self.assertRaises(KeyError, pd.getTransition, 'tr1')
        self.failIf('tr1' in names)
        self.failUnless('tr2' in names)

    # this needs a rather complicated setup
    # that is done in the test_instance.py tests
    # can we leave this test out in here ??
    def tobedone_testCreateProcessInstance(self):
        pass

class StatefulProcessDefinitionAttributesTests(unittest.TestCase):

    def setUp(self):
        self.pd = pd = StatefulProcessDefinition()
        self.doMinimalSetup()

    def doMinimalSetup(self):
        pd = self.pd
        self.st1 = st1 = DummyState()
        self.st2 = st2 = DummyState()
        pd.states.setObject('st1', st1)
        pd.states.setObject('st2', st2)

        self.tr1 = tr1 = DummyTransition()
        self.tr2 = tr2 = DummyTransition()
        pd.transitions.setObject('tr1', tr1)
        pd.transitions.setObject('tr2', tr2)

    def testInterface(self):
        verifyClass(IStatefulProcessDefinition, StatefulProcessDefinition)

    def testGetStateNames(self):
        pd = self.pd

        names = pd.states.keys()

        self.failUnless('st1' in names)
        self.failUnless('st2' in names)
        self.failUnless(pd.getInitialStateName() in names)

    def testGetState(self):
        pd = self.pd

        st1 = pd.states['st1']

        self.assertEqual(st1, self.st1)

    def testRemoveState(self):
        pd = self.pd

        del pd.states['st1']
        names = pd.getStateNames()

        self.assertRaises(KeyError, pd.states.__getitem__, 'st1')
        self.failIf('st1' in names)
        self.failUnless('st2' in names)
        self.failUnless(pd.getInitialStateName() in names)
        
    def testGetTransistionNames(self):
        pd = self.pd

        names = pd.transitions.keys()

        self.failUnless('tr1' in names)
        self.failUnless('tr2' in names)

    def testGetTransation(self):
        pd = self.pd

        tr1 = pd.transitions['tr1']

        self.assertEqual(tr1, self.tr1)

    def testRemoveTransistion(self):
        pd = self.pd

        del pd.transitions['tr1']
        names = pd.transitions.keys()

        self.assertRaises(KeyError, pd.transitions.__getitem__, 'tr1')
        self.failIf('tr1' in names)
        self.failUnless('tr2' in names)



def test_suite():
    return unittest.TestSuite((
        unittest.makeSuite(StatefulProcessDefinitionTests),
        unittest.makeSuite(StatefulProcessDefinitionAttributesTests),
        ))

if __name__ == '__main__':
    unittest.main()


=== Added File Zope3/src/zope/app/workflow/stateful/tests/test_instance.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""

Revision information:
$Id: test_instance.py,v 1.1 2003/05/08 17:27:20 jack-e Exp $
"""

import unittest

from zope.interface import Interface
from zope.interface.verify import verifyClass
from zope.schema import Text, Int

from zope.component.service import serviceManager
from zope.app.interfaces.security import IPermissionService
from zope.app.security.registries.permissionregistry \
     import permissionRegistry
from zope.app.services.servicenames import Permissions
from zope.security.checker import CheckerPublic
from zope.security.management import newSecurityManager
from zope.security.management import system_user

from zope.proxy.context import ContextWrapper
from zope.app.traversing import traverse

from zope.app.container.zopecontainer import ZopeContainerAdapter

from zope.app.interfaces.services.configuration \
     import IUseConfigurable
from zope.app.interfaces.annotation import IAttributeAnnotatable
from zope.app.interfaces.services.configuration \
     import Active, Unregistered, Registered

from zope.app.workflow.tests.workflowsetup import WorkflowSetup
from zope.app.workflow.service import WorkflowService
from zope.app.workflow.service import ProcessDefinitionConfiguration
from zope.app.interfaces.workflow.stateful \
     import IStatefulProcessInstance
from zope.app.workflow.stateful.definition \
     import StatefulProcessDefinition, State, Transition
from zope.app.workflow.stateful.instance \
     import StatefulProcessInstance, StateChangeInfo




# define and create ProcessDefinition (PD) for tests
class TestProcessDefinition(StatefulProcessDefinition):
    __implements__ = IAttributeAnnotatable, IUseConfigurable, \
                     StatefulProcessDefinition.__implements__



class ITestDataSchema(Interface):

    text = Text(title=u'a text', default=u'no text')

    value = Int(title=u'an int', default=1)


def sort(l):
    l.sort()
    return l



class SimpleProcessInstanceTests(WorkflowSetup, unittest.TestCase):

    def setUp(self):
        WorkflowSetup.setUp(self)

        pd = TestProcessDefinition()

        pd.setRelevantDataSchema(ITestDataSchema)

        pd.states.setObject('private', State())
        pd.states.setObject('published', State())
        pd.states.setObject('pending', State())
        
        pd.transitions.setObject('show',
                                 Transition('INITIAL', 'private'))
        pd.transitions.setObject('publish_direct',
                                 Transition('private', 'published'))
        pd.transitions.setObject('publish_pending',
                                 Transition('pending', 'published'))
        pd.transitions.setObject('submit_pending',
                                 Transition('private', 'pending'))
        pd.transitions.setObject('retract_published',
                                 Transition('published', 'private'))
        pd.transitions.setObject('retract_pending',
                                 Transition('pending', 'private'))

        self.default.setObject('pd1', pd )

        self.cm.setObject('', ProcessDefinitionConfiguration('definition1',
                                '/++etc++site/default/pd1'))
        traverse(self.default.getConfigurationManager(), '2').status = Active

        self.pd = self.service.getProcessDefinition('definition1')
        # give the pi some context to find a service
        self.pi = ContextWrapper(self.service.createProcessInstance('definition1'),
                                 self.rootFolder)


    def testInterface(self):
        verifyClass(IStatefulProcessInstance, StatefulProcessInstance)


    def testRelevantData(self):
        pi = self.pi
        data = pi.data

        self.assert_(ITestDataSchema.isImplementedBy(data))

        self.assertEqual(data.text, 'no text')
        self.assertEqual(data.value, 1)

        data.text = 'another text'
        data.value = 10

        self.assertEqual(data.text, 'another text')
        self.assertEqual(data.value, 10)
        

    def testSimpleTranstitions(self):
        pi = self.pi
        pd = self.pd
        
        self.assertEqual(pi.status, pd.getInitialStateName())
        self.assertEqual(pi.getOutgoingTransitions(), ['show'])
        
        pi.fireTransition('show')
        self.assertEqual(pi.status, 'private')
        self.assertEqual(sort(pi.getOutgoingTransitions()),
                         ['publish_direct', 'submit_pending'])

        pi.fireTransition('submit_pending')
        self.assertEqual(pi.status, 'pending')
        self.assertEqual(sort(pi.getOutgoingTransitions()),
                         ['publish_pending', 'retract_pending'])

        pi.fireTransition('publish_pending')
        self.assertEqual(pi.status, 'published')
        self.assertEqual(sort(pi.getOutgoingTransitions()),
                         ['retract_published'])

        pi.fireTransition('retract_published')
        self.assertEqual(pi.status, 'private')

        pi.fireTransition('submit_pending')
        self.assertEqual(pi.status, 'pending')

        pi.fireTransition('retract_pending')
        self.assertEqual(pi.status, 'private')





class ConditionProcessInstanceTests(WorkflowSetup, unittest.TestCase):

    def setUp(self):
        WorkflowSetup.setUp(self)

        pd = TestProcessDefinition()

        pd.setRelevantDataSchema(ITestDataSchema)

        pd.states.setObject('state1', State())
        pd.states.setObject('state2', State())
        
        pd.transitions.setObject('initial_state1',
                                 Transition('INITIAL', 'state1',
                                            condition='data/value'))
        pd.transitions.setObject('initial_state2',
                                 Transition('INITIAL', 'state2',
                                            condition='not: data/value'))
        pd.transitions.setObject('state1_state2',
                                 Transition('state1', 'state2',
                                            condition='python: data.text == "some text"'))
        pd.transitions.setObject('state2_state1',
                                 Transition('state2', 'state1',
                                            condition='python: data.text == "no text"'))
        pd.transitions.setObject('state1_initial',
                                 Transition('state1', 'INITIAL'))
        pd.transitions.setObject('state2_initial',
                                 Transition('state2', 'INITIAL'))

        self.default.setObject('pd1', pd )

        self.cm.setObject('', ProcessDefinitionConfiguration('definition1',
                                '/++etc++site/default/pd1'))
        traverse(self.default.getConfigurationManager(), '2').status = Active

        self.pd = self.service.getProcessDefinition('definition1')
        # give the pi some context to find a service
        self.pi = ContextWrapper(self.service.createProcessInstance('definition1'),
                                 self.rootFolder)



    def testConditionalTranstitions(self):
        pi = self.pi
        pd = self.pd

        data = pi.data
        
        self.assertEqual(pi.status, pd.getInitialStateName())
        self.assertEqual(data.text, 'no text')
        self.assertEqual(data.value, 1)

        self.assertEqual(pi.getOutgoingTransitions(), ['initial_state1'])
        self.assertRaises(KeyError, pi.fireTransition, 'initial_state2')

        pi.fireTransition('initial_state1')
        self.assertEqual(pi.status, 'state1')
        self.assertEqual(pi.getOutgoingTransitions(), ['state1_initial'])

        data.text = 'some text'
        
        self.assertEqual(sort(pi.getOutgoingTransitions()),
                         ['state1_initial', 'state1_state2'])

        pi.fireTransition('state1_state2')
        self.assertEqual(pi.status, 'state2')
        self.assertEqual(pi.getOutgoingTransitions(), ['state2_initial'])
        self.assertRaises(KeyError, pi.fireTransition, 'state2_state1')

        data.text = 'no text'

        pi.fireTransition('state2_initial')
        self.assertEqual(pi.status, 'INITIAL')
        self.assertEqual(pi.getOutgoingTransitions(), ['initial_state1'])

        data.value = 0

        self.assertEqual(pi.getOutgoingTransitions(), ['initial_state2'])
        
        pi.fireTransition('initial_state2')
        self.assertEqual(pi.status, 'state2')
        self.assertEqual(pi.getOutgoingTransitions(),
                         ['state2_initial', 'state2_state1'])






def transition_script1(contexts):
    return contexts['data'].text == "some text"

def transition_script2(contexts):
    return contexts['data'].text == "no text"

class ScriptProcessInstanceTests(WorkflowSetup, unittest.TestCase):

    def setUp(self):
        WorkflowSetup.setUp(self)

        pd = TestProcessDefinition()

        pd.setRelevantDataSchema(ITestDataSchema)

        pd.states.setObject('state1', State())
        pd.states.setObject('state2', State())
        
        pd.transitions.setObject('initial_state1',
                                 Transition('INITIAL', 'state1',
                                            script=lambda c: c['data'].value))
        pd.transitions.setObject('initial_state2',
                                 Transition('INITIAL', 'state2',
                                            script=lambda c: not c['data'].value))
        pd.transitions.setObject('state1_state2',
                                 Transition('state1', 'state2',
                                            script=transition_script1))
        pd.transitions.setObject('state2_state1',
                                 Transition('state2', 'state1',
                                            script=transition_script2))
        pd.transitions.setObject('state1_initial',
                                 Transition('state1', 'INITIAL'))
        pd.transitions.setObject('state2_initial',
                                 Transition('state2', 'INITIAL'))

        self.default.setObject('pd1', pd )

        self.cm.setObject('', ProcessDefinitionConfiguration('definition1',
                                '/++etc++site/default/pd1'))
        traverse(self.default.getConfigurationManager(), '2').status = Active

        self.pd = self.service.getProcessDefinition('definition1')
        # give the pi some context to find a service
        self.pi = ContextWrapper(self.service.createProcessInstance('definition1'),
                                 self.rootFolder)



    def testConditionalTranstitions(self):
        pi = self.pi
        pd = self.pd

        data = pi.data
        
        self.assertEqual(pi.status, pd.getInitialStateName())
        self.assertEqual(data.text, 'no text')
        self.assertEqual(data.value, 1)

        self.assertEqual(pi.getOutgoingTransitions(), ['initial_state1'])
        self.assertRaises(KeyError, pi.fireTransition, 'initial_state2')

        pi.fireTransition('initial_state1')
        self.assertEqual(pi.status, 'state1')
        self.assertEqual(pi.getOutgoingTransitions(), ['state1_initial'])

        data.text = 'some text'
        
        self.assertEqual(sort(pi.getOutgoingTransitions()),
                         ['state1_initial', 'state1_state2'])

        pi.fireTransition('state1_state2')
        self.assertEqual(pi.status, 'state2')
        self.assertEqual(pi.getOutgoingTransitions(), ['state2_initial'])
        self.assertRaises(KeyError, pi.fireTransition, 'state2_state1')

        data.text = 'no text'

        pi.fireTransition('state2_initial')
        self.assertEqual(pi.status, 'INITIAL')
        self.assertEqual(pi.getOutgoingTransitions(), ['initial_state1'])

        data.value = 0

        self.assertEqual(pi.getOutgoingTransitions(), ['initial_state2'])
        
        pi.fireTransition('initial_state2')
        self.assertEqual(pi.status, 'state2')
        self.assertEqual(pi.getOutgoingTransitions(),
                         ['state2_initial', 'state2_state1'])




class PermissionProcessInstanceTests(WorkflowSetup, unittest.TestCase):

    def setUp(self):
        WorkflowSetup.setUp(self)

        serviceManager.defineService(Permissions, IPermissionService)
        serviceManager.provideService(Permissions, permissionRegistry)
        permissionRegistry.definePermission('deny', 'Deny')

        #newSecurityManager(system_user)
        newSecurityManager('test')

        pd = TestProcessDefinition()

        pd.setRelevantDataSchema(ITestDataSchema)

        pd.states.setObject('state1', State())
        pd.states.setObject('state2', State())
        
        pd.transitions.setObject('initial_state1',
                                 Transition('INITIAL', 'state1', permission=CheckerPublic))
        pd.transitions.setObject('initial_state2',
                                 Transition('INITIAL', 'state2', permission='deny'))
        pd.transitions.setObject('state1_state2',
                                 Transition('state1', 'state2', permission=CheckerPublic))
        pd.transitions.setObject('state2_state1',
                                 Transition('state2', 'state1'))
        pd.transitions.setObject('state1_initial',
                                 Transition('state1', 'INITIAL', permission='deny'))
        pd.transitions.setObject('state2_initial',
                                 Transition('state2', 'INITIAL', permission=CheckerPublic))

        self.default.setObject('pd1', pd )

        self.cm.setObject('', ProcessDefinitionConfiguration('definition1',
                                '/++etc++site/default/pd1'))
        traverse(self.default.getConfigurationManager(), '2').status = Active

        self.pd = self.service.getProcessDefinition('definition1')
        # give the pi some context to find a service
        self.pi = ContextWrapper(self.service.createProcessInstance('definition1'),
                                 self.rootFolder)



    def testPermissionedTranstitions(self):
        pi = self.pi
        pd = self.pd

        self.assertEqual(pi.status, pd.getInitialStateName())

        self.assertEqual(pi.getOutgoingTransitions(), ['initial_state1'])
        self.assertRaises(KeyError, pi.fireTransition, 'initial_state2')

        pi.fireTransition('initial_state1')
        self.assertEqual(pi.status, 'state1')
        self.assertEqual(pi.getOutgoingTransitions(), ['state1_state2'])



class DummyTransition:
    def __init__(self, source, destination):
        self.sourceState = source
        self.destinationState = destination


class TestStateChangeInfo(unittest.TestCase):

    def testStateChangeInfo(self):
        t = DummyTransition(1,2)
        sci = StateChangeInfo(t)
        self.assertEqual(sci.old_state, 1)
        self.assertEqual(sci.new_state, 2)
        




def test_suite():
    return unittest.TestSuite((
        unittest.makeSuite(SimpleProcessInstanceTests),
        unittest.makeSuite(ConditionProcessInstanceTests),
        unittest.makeSuite(ScriptProcessInstanceTests),
        unittest.makeSuite(PermissionProcessInstanceTests),
        unittest.makeSuite(TestStateChangeInfo),
        ))

if __name__ == '__main__':
    unittest.main()


=== Added File Zope3/src/zope/app/workflow/stateful/tests/test_xmlimportexport.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
import unittest
from StringIO import StringIO

from zope.interface.verify import verifyClass
from zope.interface.implements import implements

from zope.app.services.tests.placefulsetup import PlacefulSetup
from zope.component.adapter import provideAdapter
from zope.component import getAdapter

from zope.security.checker import CheckerPublic

from zope.app.dublincore.annotatableadapter \
     import ZDCAnnotatableAdapter
from zope.app.attributeannotations import AttributeAnnotations
from zope.app.interfaces.dublincore import IZopeDublinCore
from zope.app.interfaces.annotation \
     import IAnnotatable, IAnnotations, IAttributeAnnotatable

from zope.app.interfaces.services.configuration \
     import IUseConfigurable

from zope.app.workflow.stateful.definition \
     import StatefulProcessDefinition, State, Transition

from zope.app.interfaces.workflow import IProcessDefinitionImportHandler
from zope.app.interfaces.workflow import IProcessDefinitionExportHandler
from zope.app.workflow.stateful.xmlimportexport \
     import XMLImportHandler, XMLExportHandler




xml_text = """<?xml version="1.0"?>
<workflow type="StatefulWorkflow" title="TestPD">

  <schema name="Some.path.to.an.ISchemaClass">
  </schema>

  <states>
    <state title="State2" name="state2"/>
    <state title="State1" name="state1"/>
    <state title="initial" name="INITIAL"/>
  </states>

  <transitions>
     
      <transition sourceState="state2"
                  destinationState="INITIAL"
                  script="some.path.to.some.script"
                  permission="zope.View"
                  triggerMode="Manual"
                  title="State2toINITIAL"
                  name="state2_initial"/>
    
     
      <transition sourceState="INITIAL"
                  destinationState="state1"
                  permission="zope.Public"
                  triggerMode="Automatic"
                  title="INITIALtoState1"
                  name="initial_state1"/>
    
     
      <transition sourceState="state1"
                  destinationState="state2"
                  condition="python: 1==1"
                  permission="zope.Public"
                  triggerMode="Manual"
                  title="State1toState2"
                  name="state1_state2"/>
    
  </transitions>
  
</workflow>
"""


class TestProcessDefinition(StatefulProcessDefinition):
    __implements__ = IAttributeAnnotatable, IUseConfigurable, \
                     StatefulProcessDefinition.__implements__

# need to patch this cause these classes are used directly
# in the import/export classes
implements(State, IAttributeAnnotatable)
implements(Transition, IAttributeAnnotatable)



class Test(PlacefulSetup, unittest.TestCase):

    def setUp(self):
        PlacefulSetup.setUp(self)
        provideAdapter(IAttributeAnnotatable, IAnnotations, AttributeAnnotations)
        provideAdapter(IAnnotatable, IZopeDublinCore, ZDCAnnotatableAdapter)

    def testInterface(self):
        verifyClass(IProcessDefinitionImportHandler, XMLImportHandler)
        verifyClass(IProcessDefinitionExportHandler, XMLExportHandler)

    def testImport(self):
        testpd = TestProcessDefinition()
        handler = XMLImportHandler()
        
        self.assertEqual(handler.canImport(testpd, StringIO(xml_text)), True)
        self.assertEqual(handler.canImport(None, StringIO(xml_text)), False)
        self.assertEqual(handler.canImport(None, StringIO('<some><nonworking/><xml/></some>')), False)

        handler.doImport(testpd, StringIO(xml_text))

        self.assertEqual(testpd.getRelevantDataSchema(), 'Some.path.to.an.ISchemaClass')
        self.assertEqual(getAdapter(testpd, IZopeDublinCore).title, 'TestPD')
        
        self.assertEqual(len(testpd.states), 3)
        self.assertEqual(len(testpd.transitions), 3)

        st = testpd.states['INITIAL']
        self.assert_(isinstance(st, State))
        self.assertEqual(getAdapter(st, IZopeDublinCore).title, 'initial')
                           
        st = testpd.states['state1']
        self.assert_(isinstance(st, State))
        self.assertEqual(getAdapter(st, IZopeDublinCore).title, 'State1')

        st = testpd.states['state2']
        self.assert_(isinstance(st, State))
        self.assertEqual(getAdapter(st, IZopeDublinCore).title, 'State2')


        tr = testpd.transitions['initial_state1']
        self.assert_(isinstance(tr, Transition))
        self.assertEqual(getAdapter(tr, IZopeDublinCore).title, 'INITIALtoState1')
        self.assertEqual(tr.sourceState, 'INITIAL')
        self.assertEqual(tr.destinationState, 'state1')
        self.assertEqual(tr.condition, None)
        self.assertEqual(tr.script, None)
        self.assertEqual(tr.permission, CheckerPublic)
        self.assertEqual(tr.triggerMode, 'Automatic')
        
        tr = testpd.transitions['state1_state2']
        self.assert_(isinstance(tr, Transition))
        self.assertEqual(getAdapter(tr, IZopeDublinCore).title, 'State1toState2')
        self.assertEqual(tr.sourceState, 'state1')
        self.assertEqual(tr.destinationState, 'state2')
        self.assertEqual(tr.condition, 'python: 1==1')
        self.assertEqual(tr.script, None)
        self.assertEqual(tr.permission, CheckerPublic)
        self.assertEqual(tr.triggerMode, 'Manual')
        
        tr = testpd.transitions['state2_initial']
        self.assert_(isinstance(tr, Transition))
        self.assertEqual(getAdapter(tr, IZopeDublinCore).title, 'State2toINITIAL')
        self.assertEqual(tr.sourceState, 'state2')
        self.assertEqual(tr.destinationState, 'INITIAL')
        self.assertEqual(tr.condition, None)
        self.assertEqual(tr.script, 'some.path.to.some.script')
        self.assertEqual(tr.permission, 'zope.View')
        self.assertEqual(tr.triggerMode, 'Manual')

    def testExport(self):
        # XXX TBD before Merge into HEAD !!!!
        pass
        

        

def test_suite():
    loader=unittest.TestLoader()
    return loader.loadTestsFromTestCase(Test)

if __name__=='__main__':
    unittest.TextTestRunner().run(test_suite())