[Zope-CVS] CVS: Packages3/workflow/stateful/tests - test_definition.py:1.3 test_instance.py:1.3

Ulrich Eck ueck@net-labs.de
Thu, 6 Feb 2003 17:03:39 -0500


Update of /cvs-repository/Packages3/workflow/stateful/tests
In directory cvs.zope.org:/tmp/cvs-serv23752/stateful/tests

Modified Files:
	test_definition.py test_instance.py 
Log Message:
Next round in Stateful workflow implementation:
- Transitions:
  + Conditions based on TALES Expressions work
  - Permissions are implemented but tests do not work yet (tests disabled)
- Instance:
  + Relevant-Data based on Schema basically works (schema written as persistent module)
  - Permissions for dataaccess not yet implemented
- BrowserViews:
 - basic setup for ManagementInterface made, no functionality yet
- extensive tests for the instance written.


=== Packages3/workflow/stateful/tests/test_definition.py 1.2 => 1.3 ===
--- Packages3/workflow/stateful/tests/test_definition.py:1.2	Wed Feb  5 20:09:33 2003
+++ Packages3/workflow/stateful/tests/test_definition.py	Thu Feb  6 17:03:07 2003
@@ -20,6 +20,7 @@
 
 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
@@ -39,6 +40,12 @@
     __implements__ = IDummyTransition
 
 
+class IDummyDataSchema(Interface):
+
+    text = TextLine(title=u'a text', default=u'no text')
+    
+
+
 class StatefulProcessDefinitionTests(unittest.TestCase):
 
     def setUp(self):
@@ -47,6 +54,7 @@
 
     def doMinimalSetup(self):
         pd = self.pd
+        pd.setSchema(IDummyDataSchema)
         self.st1 = st1 = DummyState()
         self.st2 = st2 = DummyState()
         pd.addState('st1', st1)
@@ -59,6 +67,9 @@
 
     def testInterface(self):
         verifyClass(IStatefulProcessDefinition, StatefulProcessDefinition)
+
+    def testGetSchema(self):
+        self.assertEqual(self.pd.getSchema(), IDummyDataSchema)
 
     def testGetStateNames(self):
         pd = self.pd


=== Packages3/workflow/stateful/tests/test_instance.py 1.2 => 1.3 ===
--- Packages3/workflow/stateful/tests/test_instance.py:1.2	Wed Feb  5 20:20:58 2003
+++ Packages3/workflow/stateful/tests/test_instance.py	Thu Feb  6 17:03:07 2003
@@ -21,25 +21,23 @@
 
 from zope.interface import Interface
 from zope.interface.verify import verifyClass
+from zope.schema import Text, Int
+
+from zope.security.management import newSecurityManager
+
+from zope.app.security.grants.principalpermission \
+     import principalPermissionManager
 
 from zope.proxy.context import ContextWrapper
 from zope.app.traversing import traverse
+
 from zope.app.container.zopecontainer import ZopeContainerAdapter
-from zope.app.services.service import ServiceManager
-from zope.app.services.service import ServiceConfiguration
-from zope.app.interfaces.dependable import IDependable
-from zope.app.services.tests.placefulsetup \
-     import PlacefulSetup
-from zope.app.interfaces.annotation import IAnnotatable
+
 from zope.app.interfaces.annotation import IAttributeAnnotatable
-from zope.app.attributeannotations import AttributeAnnotations
-from zope.app.interfaces.annotation import IAnnotations
-from zope.app.interfaces.dependable import IDependable
-from zope.app.dependable import Dependable
-from zope.component.adapter import provideAdapter
 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 \
@@ -50,61 +48,51 @@
      import StatefulProcessInstance
 
 
-class WorkflowServiceForTests(WorkflowService):
-
-    __implements__ = WorkflowService.__implements__, IAttributeAnnotatable
-
-
 # define and create ProcessDefinition (PD) for tests
 class TestProcessDefinition(StatefulProcessDefinition):
     __implements__ = IAttributeAnnotatable, 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(PlacefulSetup, unittest.TestCase):
 
-    def setUp(self):
-        PlacefulSetup.setUp(self)
 
-        provideAdapter(IAttributeAnnotatable,
-                       IAnnotations, AttributeAnnotations)
-        provideAdapter(IAnnotatable, IDependable, Dependable)
-
-        # Set up a local workflow service
-        self.buildFolders()
-        self.rootFolder.setServiceManager(ServiceManager())
-
-        self.default = traverse(self.rootFolder,
-                                '++etc++Services/Packages/default')
-        self.default.setObject('workflow_srv', WorkflowServiceForTests())
-        self.service = traverse(self.default, 'workflow_srv')
-
-        self.cm = ZopeContainerAdapter(traverse(self.default, "configure"))
-        self.cm.setObject('', ServiceConfiguration('Workflows',
-                                '/++etc++Services/Packages/default/workflow_srv'))
-        traverse(self.default, 'configure/1').status = Active
+class SimpleProcessInstanceTests(WorkflowSetup, unittest.TestCase):
+
+    def setUp(self):
+        WorkflowSetup.setUp(self)
 
         pd = TestProcessDefinition()
 
+        pd.setSchema(ITestDataSchema)
+
         pd.states.setObject('private', State())
         pd.states.setObject('published', State())
         pd.states.setObject('pending', State())
         
         pd.transitions.setObject('show',
-                                 Transition('INITIAL', 'private', None))
+                                 Transition('INITIAL', 'private'))
         pd.transitions.setObject('publish_direct',
-                                 Transition('private', 'published', None))
+                                 Transition('private', 'published'))
         pd.transitions.setObject('publish_pending',
-                                 Transition('pending', 'published', None))
+                                 Transition('pending', 'published'))
         pd.transitions.setObject('submit_pending',
-                                 Transition('private', 'pending', None))
+                                 Transition('private', 'pending'))
         pd.transitions.setObject('retract_published',
-                                 Transition('published', 'private', None))
+                                 Transition('published', 'private'))
         pd.transitions.setObject('retract_pending',
-                                 Transition('pending', 'private', None))
+                                 Transition('pending', 'private'))
 
         self.default.setObject('pd1', pd )
 
@@ -121,9 +109,27 @@
     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'])
         
@@ -152,10 +158,174 @@
         self.assertEqual(pi.status, 'private')
 
 
+class ConditionProcessInstanceTests(WorkflowSetup, unittest.TestCase):
+
+    def setUp(self):
+        WorkflowSetup.setUp(self)
+
+        pd = TestProcessDefinition()
+
+        pd.setSchema(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++Services/Packages/default/pd1'))
+        traverse(self.default, 'configure/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)
+
+        # Setup a test principal and role, assing role to principal
+        self.auth_svc = self.setupAuthService()
+        self.auth_svc.definePrincipal("TestPrincipal", "Test Principal")
+        self.principal = self.auth_svc.getPrincipal("TestPrincipal")
+        self.principalId = self.principal.getId()
+
+        # Setup a permission service
+        self.perm_svc = self.setupPermissionService()
+        # set up some permissions
+        read = self.perm_svc.definePermission('read', 'Read',
+                                              'Read something')
+        self.read = read.getId()
+        write = self.perm_svc.definePermission('write', 'Write',
+                                               'Write something')
+        self.write = write.getId()
+
+        # grant and deny a permission to principal
+        principalPermissionManager.grantPermissionToPrincipal(self.read, self.principalId)
+        principalPermissionManager.denyPermissionToPrincipal(self.write, self.principalId)
+        
+        # switch to security-context of test principal
+        newSecurityManager(self.principal)
+
+        # create TestProcessDefinition
+        pd = TestProcessDefinition()
+
+        pd.setSchema(ITestDataSchema)
+
+        pd.states.setObject('state1', State())
+        pd.states.setObject('state2', State())
+        
+        pd.transitions.setObject('initial_state1',
+                                 Transition('INITIAL', 'state1', permission=self.read))
+        pd.transitions.setObject('initial_state2',
+                                 Transition('INITIAL', 'state2', permission=self.write))
+        pd.transitions.setObject('state1_state2',
+                                 Transition('state1', 'state2', permission=self.read))
+        pd.transitions.setObject('state2_state1',
+                                 Transition('state2', 'state1'))
+        pd.transitions.setObject('state1_initial',
+                                 Transition('state1', 'INITIAL', permission=self.write))
+        pd.transitions.setObject('state2_initial',
+                                 Transition('state2', 'INITIAL', permission=self.read))
+
+        self.default.setObject('pd1', pd )
+
+        self.cm.setObject('', ProcessDefinitionConfiguration('definition1',
+                                '/++etc++Services/Packages/default/pd1'))
+        traverse(self.default, 'configure/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 DISABLED_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'])
+
+
 def test_suite():
     suite = unittest.TestSuite()
     suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(
-        SimpleProcessInstanceTests))
+        SimpleProcessInstanceTests, ConditionProcessInstanceTests,
+        PermissionProcessInstanceTests))
     return suite
 
 if __name__ == '__main__':