Adapted ITask to add method getting the highest parent

CHANGES.rst

@@ -1,11 +1,11 @@
 Changelog
 =========
 
-2.3 (unreleased)
+2.2.2 (unreleased)
 ----------------
 
-- Nothing changed yet.
-
+- Adapted ITask to add method getting the highest parent.
+  [sgeulette]
 
 2.2.1 (2016-01-13)
 ------------------

src/collective/task/adapters.py

@@ -1,11 +1,17 @@
 # -*- coding: utf-8 -*-
 
+from zope.component import adapts
+from zope.interface import implements
+
 from plone.indexer import indexer
 from Products.CMFCore.interfaces import IContentish
 from Products.CMFPlone.utils import base_hasattr
 from Products.PluginIndexes.common.UnIndex import _marker as common_marker
 from Products.PluginIndexes.DateIndex.DateIndex import _marker as date_marker
 
+from .behaviors import ITask
+from.interfaces import ITaskMethods
+
 
 @indexer(IContentish)
 def assigned_group_index(obj):
@@ -29,3 +35,29 @@ def due_date_index(obj):
     if base_hasattr(obj, 'due_date') and obj.due_date:
         return obj.due_date
     return date_marker
+
+
+class TaskAdapter(object):
+    implements(ITaskMethods)
+    adapts(ITask)
+
+    def __init__(self, context):
+        self.context = context
+
+    def get_highest_task_parent(self, task=False):
+        """
+            Get the object containing the highest ITask object
+            or the highest ITask object if task is True
+        """
+        obj = self.context
+        while obj is not None:
+#            if not hasattr(obj, "aq_parent"):
+#                raise RuntimeError("Parent traversing interrupted by object: " + str(obj))
+            parent = obj.aq_parent
+            if not ITask.providedBy(parent):
+                if task:  # we want the highest task, we return the current task obj
+                    return obj
+                else:  # we return the parent that's not a task
+                    return parent
+            obj = parent
+        return obj

src/collective/task/configure.zcml

@@ -38,4 +38,6 @@
     <adapter name="assigned_user" factory=".adapters.assigned_user_index" />
     <adapter name="due_date" factory=".adapters.due_date_index" />
 
+    <adapter factory=".adapters.TaskAdapter" />
+
 </configure>

src/collective/task/interfaces.py

@@ -2,6 +2,7 @@
 """Module where all interfaces, events and exceptions live."""
 
 from zope import schema
+from zope.interface import Interface
 from zope.publisher.interfaces.browser import IDefaultBrowserLayer
 from plone.app.dexterity import PloneMessageFactory as _PMF
 from plone.supermodel import model
@@ -18,3 +19,13 @@ class ITaskContent(model.Schema):
         title=_PMF(u'label_title', default=u'Title'),
         required=True
     )
+
+
+class ITaskMethods(Interface):
+
+    def get_highest_task_parent(task=False):
+        """
+            Get the object containing the highest ITask object
+            or the highest ITask object if task is True
+        """
+

src/collective/task/tests/test_adapters.py

@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+import unittest
+from plone import api
+from plone.app.testing import login, TEST_USER_NAME, setRoles, TEST_USER_ID
+
+from ..testing import COLLECTIVE_TASK_FUNCTIONAL_TESTING
+from ..interfaces import ITaskMethods
+
+
+class TestAdapters(unittest.TestCase):
+
+    layer = COLLECTIVE_TASK_FUNCTIONAL_TESTING
+
+    def setUp(self):
+        super(TestAdapters, self).setUp()
+        self.portal = self.layer['portal']
+        setRoles(self.portal, TEST_USER_ID, ['Manager'])
+        login(self.portal, TEST_USER_NAME)
+        self.task1 = api.content.create(container=self.portal, type='task', id='task1', title='Task1')
+        self.folder = api.content.create(container=self.portal, type='Folder', id='folder', title='Folder')
+        self.task2 = api.content.create(container=self.folder, type='task', id='task2', title='Task2')
+        self.task3 = api.content.create(container=self.task2, type='task', id='task3', title='Task3')
+
+    def test_get_highest_task_parent(self):
+        self.assertEqual(ITaskMethods(self.task1).get_highest_task_parent(), self.portal)
+        self.assertEqual(ITaskMethods(self.task1).get_highest_task_parent(task=True), self.task1)
+        self.assertEqual(ITaskMethods(self.task2).get_highest_task_parent(), self.folder)
+        self.assertEqual(ITaskMethods(self.task2).get_highest_task_parent(task=True), self.task2)
+        self.assertEqual(ITaskMethods(self.task3).get_highest_task_parent(), self.folder)
+        self.assertEqual(ITaskMethods(self.task3).get_highest_task_parent(task=True), self.task2)