cco/loops
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Resource constraints added to Task

Browse source 
git-svn-id: svn://svn.cy55.de/Zope3/src/loops/trunk@112 fd906abe-77d9-0310-91a1-e0d9ade77398
This commit is contained in:
helmutm 2004-09-23 07:16:31 +00:00
parent 3dfee96db3
commit f4cc0c132c
6 changed files with 230 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
constraint.py
View file

@ -37,12 +37,11 @@ class ResourceConstraint(object):
referenceKey = None
def __init__(self, task):
def __init__(self):
self.referenceValues = []
self._task = task
def isResourceAllowed(self, resource):
def isResourceAllowed(self, resource, task=None):
if self.referenceType == 'parent':
for r in self.referenceValues:
m = getattr(r, self.referenceKey)
@ -55,7 +54,7 @@ class ResourceConstraint(object):
return False
def getAllowedResources(self, candidates=None):
def getAllowedResources(self, candidates=None, task=None):
if self.referenceType == 'parent':
result = []
for r in self.referenceValues:

142
interfaces.py
View file

@ -28,6 +28,58 @@ from zope.app.container.interfaces import IOrderedContainer
from zope.schema import Text, TextLine, List, Object
class IResourceConstraint(Interface):
""" A ResourceConstraint governs which Resource objects may be
allocated to a Task object.
"""
explanation = Text(
title=u'Explanation',
description=u'Explanation of this constraint - '
'why or why not certain resources may be allowed',
required=True)
constraintType = TextLine(
title=u'Constraint Type',
description=u'Type of the constraint: select, require, disallow',
default=u'select',
required=True)
referenceType = TextLine(
title=u'Reference Type',
description=u'Type of reference to the resource attribute to check: '
'explicit, parent, type, attribute, property, method',
default=u'explicit',
required=True)
referenceKey = TextLine(
title=u'Reference Key',
description=u'Key for referencing the resource attribute')
referenceValues = List(
title=u'Reference Values',
description=u'Attribute values to check for; may be any kind of object',
value_type=Object(Interface, title=u'Value'),
unique=True)
def isResourceAllowed(resource, task=None):
""" Return True if this ResourceConstraint allows the resource given.
If a task parameter is given there may be special checks on it, e.g.
on concerning subtasks of the task's parent task (sibling constraints).
"""
def getAllowedResources(candidates=None, task=None):
""" Return a list of resource objects that are allowed by this
ResourceConstraint.
If given, use candidates as a list of possible resources
(candidates must implement the IResource interface).
If a task parameter is given there may be special checks on it, e.g.
on concerning subtasks of the task's parent task (sibling constraints).
"""
class ITask(IOrderedContainer):
""" A Task is a piece of work.
@ -41,6 +93,17 @@ class ITask(IOrderedContainer):
default=u'',
required=True)
resourceConstraints = List(
title=u'Resource Constraints',
description=u'Collection of Constraints controlling the resources '
'that may be allocated to a task',
default=[],
required=False,
value_type=Object(schema=IResourceConstraint, title=u'Resource Constraint'),
unique=True)
# subtask stuff:
def getSubtasks(taskTypes=None):
""" Return a tuple of subtasks of self,
possibly restricted to the task types given.
@ -67,6 +130,8 @@ class ITask(IOrderedContainer):
""" Remove the subtask relation to task from self.
"""
# resource allocations:
def getAllocatedResources(allocTypes=None, resTypes=None):
""" Return a tuple of resources allocated to self,
possibly restricted to the allocation types and
@ -109,6 +174,38 @@ class ITask(IOrderedContainer):
in the controller object that is responsible for self.
"""
# resource constraint stuff:
def isResourceAllowed(resource):
""" Return True if the resource given is allowed for this task.
"""
def getCandidateResources():
""" Return a tuple of resource objects that are allowed for this task.
Returns empty tuple if no usable resource constraints present.
"""
def getAllowedResources(candidates=None):
""" Return a list of resource objects that are allowed for this task.
If given, use candidates as a list of possible resources
(candidates must implement the IResource interface).
Returns None if no usable resource constraints are present.
Falls back to getCandidateResources if candidates is None
and usable resource constraints are present.
"""
# Task object as prototype:
def copyTask(targetContainer=None):
""" Copy self to the target container given and return the new task.
Also copy all subtasks. Keep the references to resources and
resource constraints without copying them.
targetContainer defaults to self.getParent().
"""
class IResource(IOrderedContainer):
""" A Resource is an object - a thing or a person - that may be
@ -122,48 +219,3 @@ class IResource(IOrderedContainer):
"""
class IResourceConstraint(Interface):
""" A ResourceConstraint governs which Resource objects may be
allocated to a Task object.
"""
explanation = Text(
title=u'Explanation',
description=u'Explanation of this constraint - '
'why or why not certain resources may be allowed',
required=True)
constraintType = TextLine(
title=u'Constraint Type',
description=u'Type of the constraint: select, require, disallow',
default=u'select',
required=True)
referenceType = TextLine(
title=u'Reference Type',
description=u'Type of reference to the resource attribute to check: '
'explicit, parent, type, attribute, property, method'
default=u'explicit',
required=True)
referenceKey = TextLine(
title=u'Reference Key',
description=u'Key for referencing the resource attribute')
referenceValues = List(
title=u'Reference Values',
description=u'Attribute values to check for; may be any kind of object',
value_type=Object(Interface, title=u'Value'))
def isResourceAllowed(resource):
""" Return True if this ResourceConstraint allows the resource given.
"""
def getAllowedResources(candidates=None):
""" Return a list of resource objects that are allowed by this
ResourceConstraint.
If given, use candidates as a list of possible resources
(candidates must implement the IResource interface).
"""

4
resource.py
View file

@ -69,7 +69,7 @@ class Resource(OrderedContainer):
def _createResourceName(self, container=None):
prefix = 'rsc'
container = container or zapi.getParent(self)
last = max([ int(n[len(prefix):]) for n in container.keys() ] or [1])
last = max([ int(n[len(prefix):]) for n in container.keys() ] or [0])
return prefix + str(last+1)

50
task.py
View file

@ -24,11 +24,14 @@ $Id$
from zope.interface import implements
from zope.app.container.ordered import OrderedContainer
from zope.app.copypastemove import ObjectCopier
from zope.app import zapi
from resource import Resource
from interfaces import ITask
from copy import copy
class Task(OrderedContainer):
implements(ITask)
@ -41,6 +44,7 @@ class Task(OrderedContainer):
self._subtasks = []
self._parentTasks = []
self._resourceAllocs = {}
self.resourceConstraints = []
# subtasks:
@ -112,11 +116,55 @@ class Task(OrderedContainer):
def getAllAllocTypes(self):
return ('standard',)
# resource constraint stuff:
def isResourceAllowed(self, resource):
rc = self.resourceConstraints
if not rc:
return True
for c in rc:
# that's too simple, we must check all constraints for constraintType:
if c.isResourceAllowed(resource):
return True
return False
def getCandidateResources(self):
rc = self.resourceConstraints
if not rc:
return ()
result = []
for c in rc:
result.extend(c.getAllowedResources())
return tuple(result)
def getAllowedResources(self, candidates=None):
rc = self.resourceConstraints
if not rc:
return None
if candidates is None:
result = self.getCandidateResources()
# Empty result means: can't tell
return result and result or None
return tuple([ c for c in candidates if self.isResourceAllowed(c) ])
# Task object as prototype:
def copyTask(self, targetContainer=None):
targetContainer = targetContainer or zapi.getParent(self)
newName = self._createTaskName(targetContainer)
newTask = copy(self)
targetContainer[newName] = newTask
newTask._subtasks = []
for st in self.getSubtasks():
newSt = st.copyTask(targetContainer)
newSt._parentTasks.remove(self)
newTask.assignSubtask(newSt)
return newTask
# Helper methods:
def _createTaskName(self, container=None):
prefix = 'tsk'
container = container or zapi.getParent(self)
last = max([ int(n[len(prefix):]) for n in container.keys() ] or [1])
last = max([ int(n[len(prefix):]) for n in container.keys() ] or [0])
return prefix + str(last+1)

2
tests/test_constraint.py
View file

@ -22,7 +22,7 @@ class Test(unittest.TestCase):
self.f1['rsc1'] = self.r1
self.t1 = Task()
self.f1['tsk1'] = self.t1
self.rc1 = ResourceConstraint(self.t1)
self.rc1 = ResourceConstraint()
def tearDown(self):
pass

79
tests/test_task.py
View file

@ -86,7 +86,7 @@ class TestTaskResource(unittest.TestCase):
self.t1 = Task()
self.r1 = Resource()
self.f1['tsk1'] = self.t1
self.r1['rsc1'] = self.r1
self.f1['rsc1'] = self.r1
def tearDown(self):
pass
@ -115,11 +115,88 @@ class TestTaskResource(unittest.TestCase):
self.assertEqual((r1,), t1.getAllocatedResources())
from loops.constraint import ResourceConstraint
class TestTaskResourceConstraints(unittest.TestCase):
"Test methods of the Task class related to checking for allowed resources."
def setUp(self):
self.f1 = Folder()
self.f1.__name__ = u'f1'
self.t1 = Task()
self.r1 = Resource()
self.f1['tsk1'] = self.t1
self.r1['rsc1'] = self.r1
self.rc1 = ResourceConstraint()
def tearDown(self):
pass
# the tests...
def testSelectExplicit(self):
t1 = self.t1
r1 = self.r1
rc1 = self.rc1
self.assertEqual(True, t1.isResourceAllowed(r1)) # no check
self.assertEqual((), t1.getCandidateResources()) # no candidates
self.assertEqual(None, t1.getAllowedResources([r1]))# can't say without constraint
self.t1.resourceConstraints.append(self.rc1) # empty constraint
self.assertEqual(False, t1.isResourceAllowed(r1)) # does not allow
self.assertEqual((), t1.getCandidateResources()) # anything
self.assertEqual((), t1.getAllowedResources([r1]))
rc1.referenceValues = ([r1])
self.assertEqual(True, rc1.isResourceAllowed(r1))
self.assertEqual((r1,), t1.getCandidateResources())
self.assertEqual((r1,), rc1.getAllowedResources([r1]))
class TestTaskCopy(unittest.TestCase):
"Tests related to copying tasks e.g. when using task prototypes."
def setUp(self):
self.f1 = Folder()
self.f1.__name__ = u'f1'
self.t1 = Task()
self.r1 = Resource()
self.f1['tsk1'] = self.t1
self.f1['rsc1'] = self.r1
self.rc1 = ResourceConstraint()
def tearDown(self):
pass
# the tests...
def testCopyTask(self):
t1 = self.t1
r1 = self.r1
rc1 = self.rc1
ts1 = t1.createSubtask()
ts1.resourceConstraints.append(rc1)
ts1.allocateResource(r1)
t2 = t1.copyTask()
self.failIf(t1 is t2, 't1 and t2 are still the same')
st2 = t2.getSubtasks()
self.assertEquals(1, len(st2))
ts2 = st2[0]
self.failIf(ts1 is ts2, 'ts1 and ts2 are still the same')
self.assertEquals((r1,), ts2.getAllocatedResources())
self.assertEquals([rc1,], ts2.resourceConstraints)
def test_suite():
return unittest.TestSuite((
# DocTestSuite('loops.tests.doctests'),
unittest.makeSuite(TestTask),
unittest.makeSuite(TestTaskResource),
unittest.makeSuite(TestTaskResourceConstraints),
unittest.makeSuite(TestTaskCopy),
))
if __name__ == '__main__':