Merge pull request #3 from IMIO/uid_update

Override 'update' and 'workflow transition' to use the uid
IMIO · Sep 17, 2021 · bc0f5b6 · bc0f5b6
2 parents 7b0de52 + f028bd4
commit bc0f5b6
Show file tree

Hide file tree

Showing 7 changed files with 346 additions and 1 deletion.
diff --git a/CHANGES.rst b/CHANGES.rst
@@ -5,7 +5,8 @@ Changelog
 1.0a15 (unreleased)
 -------------------
 
-- Nothing changed yet.
+- Override 'update' and 'workflow transition' to use the uid
+  [vpiret]
 
 
 1.0a14 (2021-07-16)

diff --git a/src/imio/restapi/services/configure.zcml b/src/imio/restapi/services/configure.zcml
@@ -73,6 +73,24 @@
     permission="cmf.ModifyPortalContent"
     />
 
+  <plone:service
+    method="PATCH"
+    name="@content"
+    for="Products.CMFCore.interfaces.ISiteRoot"
+    factory=".update.ContentPatch"
+    layer="imio.restapi.interfaces.IImioRestapiLayer"
+    permission="zope2.View"
+    />
+
+  <plone:service
+    method="POST"
+    name="@wf"
+    for="Products.CMFCore.interfaces.ISiteRoot"
+    factory=".transition.WorkflowTransition"
+    layer="imio.restapi.interfaces.IImioRestapiLayer"
+    permission="zope2.View"
+    />
+
   <plone:service
     method="DELETE"
     for="*"

diff --git a/src/imio/restapi/services/transition.py b/src/imio/restapi/services/transition.py
@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+from plone.app.uuid.utils import uuidToObject
+from plone.restapi.services.workflow import transition
+from zExceptions import BadRequest
+from zope.interface import implementer
+from zope.publisher.interfaces import IPublishTraverse
+from zope.publisher.interfaces import NotFound
+
+
+UID_REQUIRED_ERROR = "Missing UID"
+TRANSITION_REQUIRED_ERROR = "Missing workflow transition"
+UID_NOT_FOUND_ERROR = 'No element found with UID "%s"!'
+
+
+@implementer(IPublishTraverse)
+class WorkflowTransition(transition.WorkflowTransition):
+    """Updates an existing content object."""
+
+    def __init__(self, context, request):
+        super(WorkflowTransition, self).__init__(context, request)
+        self.uid = None
+
+    def publishTraverse(self, request, name):
+        if self.uid is None:
+            self.uid = name
+        else:
+            if self.transition is None:
+                self.transition = name
+            else:
+                raise NotFound(self, name, request)
+        return self
+
+    def reply(self):
+        if self.uid is None:
+            raise Exception(UID_REQUIRED_ERROR)
+        if self.transition is None:
+            raise Exception(TRANSITION_REQUIRED_ERROR)
+
+        obj = uuidToObject(uuid=self.uid)
+        if not obj:
+            raise BadRequest(UID_NOT_FOUND_ERROR % self.uid)
+
+        self.context = obj
+        super(WorkflowTransition, self).reply()
diff --git a/src/imio/restapi/services/update.py b/src/imio/restapi/services/update.py
@@ -0,0 +1,37 @@
+# -*- coding: utf-8 -*-
+from plone.app.uuid.utils import uuidToObject
+from plone.restapi.services.content import update
+from zExceptions import BadRequest
+from zope.interface import implementer
+from zope.publisher.interfaces import IPublishTraverse
+from zope.publisher.interfaces import NotFound
+
+
+UID_REQUIRED_ERROR = 'Missing UID'
+UID_NOT_FOUND_ERROR = 'No element found with UID "%s"!'
+
+
+@implementer(IPublishTraverse)
+class ContentPatch(update.ContentPatch):
+    """Updates an existing content object."""
+
+    def __init__(self, context, request):
+        super(ContentPatch, self).__init__(context, request)
+        self.uid = None
+
+    def publishTraverse(self, request, name):
+        if self.uid is None:
+            self.uid = name
+        else:
+            raise NotFound(self, name, request)
+        return self
+
+    def reply(self):
+        if self.uid is None:
+            raise Exception(UID_REQUIRED_ERROR)
+        obj = uuidToObject(uuid=self.uid)
+        if not obj:
+            raise BadRequest(UID_NOT_FOUND_ERROR % self.uid)
+
+        self.context = obj
+        super(ContentPatch, self).reply()
diff --git a/src/imio/restapi/testing.py b/src/imio/restapi/testing.py
@@ -7,6 +7,8 @@
 from plone.app.testing import PLONE_FIXTURE
 from plone.app.testing import PloneSandboxLayer
 from plone.restapi.testing import PLONE_RESTAPI_AT_FUNCTIONAL_TESTING
+from plone.restapi.testing import PLONE_RESTAPI_DX_FUNCTIONAL_TESTING
+from plone.restapi.testing import PLONE_RESTAPI_WORKFLOWS_INTEGRATION_TESTING
 from plone.testing import z2
 
 import collective.documentgenerator
@@ -65,3 +67,13 @@ def setUpPloneSite(self, portal):
     bases=(IMIO_RESTAPI_FIXTURE, REMOTE_LIBRARY_BUNDLE_FIXTURE, z2.ZSERVER_FIXTURE),
     name="ImioRestapiLayer:AcceptanceTesting",
 )
+
+IMIO_RESTAPI_WORKFLOWS_INTEGRATION_TESTING = IntegrationTesting(
+    bases=(IMIO_RESTAPI_FIXTURE, PLONE_RESTAPI_WORKFLOWS_INTEGRATION_TESTING, z2.ZSERVER_FIXTURE),
+    name="ImioRestapiLayer:IntegrationWorkflowTesting"
+)
+
+IMIO_RESTAPI_DX_FUNCTIONAL_TESTING = FunctionalTesting(
+    bases=(IMIO_RESTAPI_FIXTURE, PLONE_RESTAPI_DX_FUNCTIONAL_TESTING),
+    name="ImioRestapiLayer:FunctionalDXTesting",
+)
diff --git a/src/imio/restapi/tests/test_service_transition.py b/src/imio/restapi/tests/test_service_transition.py
@@ -0,0 +1,155 @@
+# -*- coding: utf-8 -*-
+from DateTime import DateTime
+from imio.restapi.testing import IMIO_RESTAPI_WORKFLOWS_INTEGRATION_TESTING
+from plone.app.testing import login
+from plone.app.testing import setRoles
+from plone.app.testing import SITE_OWNER_NAME
+from plone.app.testing import SITE_OWNER_PASSWORD
+from plone.app.testing import TEST_USER_ID
+from plone.app.testing import TEST_USER_NAME
+from plone.app.testing import TEST_USER_PASSWORD
+from Products.CMFCore.utils import getToolByName
+from unittest import TestCase
+
+import requests
+import transaction
+
+
+class TestWorkflowTransition(TestCase):
+
+    layer = IMIO_RESTAPI_WORKFLOWS_INTEGRATION_TESTING
+
+    def setUp(self):
+        self.portal = self.layer["portal"]
+        self.request = self.layer["request"]
+        self.portal_url = self.portal.absolute_url()
+        self.wftool = getToolByName(self.portal, "portal_workflow")
+        login(self.portal, SITE_OWNER_NAME)
+        self.portal.invokeFactory("Document", id="doc1")
+        self.folder = self.portal[
+            self.portal.invokeFactory("Folder", id="folder", title="Test")
+        ]
+        self.subfolder = self.folder[
+            self.folder.invokeFactory("Folder", id="subfolder")
+        ]
+        transaction.commit()
+
+    def tearDown(self):
+        login(self.portal, SITE_OWNER_NAME)
+        self.portal.manage_delObjects(["doc1"])
+        self.portal.manage_delObjects(["folder"])
+        transaction.commit()
+
+    def test_transition_action_succeeds(self):
+        uid = self.portal.doc1.UID()
+        endpoint_url = "{0}/@wf/{1}/publish".format(self.portal_url, uid)
+        requests.post(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD),
+            json={},
+        )
+        transaction.commit()
+        self.assertEqual(
+            u"published", self.wftool.getInfoFor(self.portal.doc1, u"review_state")
+        )
+
+    def test_transition_action_succeeds_changes_effective(self):
+        uid = self.portal.doc1.UID()
+        endpoint_url = "{0}/@wf/{1}/publish".format(self.portal_url, uid)
+        self.assertEqual(self.portal.doc1.effective_date, None)
+        now = DateTime()
+        requests.post(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD),
+            json={},
+        )
+        transaction.commit()
+        self.assertTrue(isinstance(self.portal.doc1.effective_date, DateTime))
+        self.assertTrue(self.portal.doc1.effective_date >= now)
+
+    def test_calling_workflow_with_additional_path_segments_results_in_404(self):
+        uid = self.portal.doc1.UID()
+        endpoint_url = "{0}/@wf/{1}/publish/test".format(self.portal_url, uid)
+        response = requests.post(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD),
+            json={},
+        )
+        transaction.commit()
+        self.assertEqual(404, response.status_code)
+
+    def test_transition_including_children(self):
+        transaction.commit()
+        uid = self.folder.UID()
+        endpoint_url = "{0}/@wf/{1}/publish".format(self.portal_url, uid)
+        response = requests.post(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD),
+            json={"include_children": "true"},
+        )
+        transaction.commit()
+        self.assertEqual(200, response.status_code)
+        self.assertEqual(
+            u"published", self.wftool.getInfoFor(self.folder, u"review_state")
+        )
+        self.assertEqual(
+            u"published", self.wftool.getInfoFor(self.subfolder, u"review_state")
+        )
+
+    def test_transition_with_effective_date(self):
+        uid = self.portal.doc1.UID()
+        endpoint_url = "{0}/@wf/{1}/publish".format(self.portal_url, uid)
+        requests.post(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD),
+            json={"effective": "2018-06-24T09:17:02"},
+        )
+        transaction.commit()
+        self.assertEqual(
+            "2018-06-24T09:17:00+00:00", self.portal.doc1.effective().ISO8601()
+        )
+
+    def test_transition_with_expiration_date(self):
+        uid = self.portal.doc1.UID()
+        endpoint_url = "{0}/@wf/{1}/publish".format(self.portal_url, uid)
+        requests.post(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD),
+            json={"expires": "2019-06-20T18:00:00",
+                  "comment": "A comment"},
+        )
+        transaction.commit()
+        self.assertEqual(
+            "A comment", self.wftool.getInfoFor(self.portal.doc1, u"comments")
+        )
+        self.assertEqual(
+            "2019-06-20T18:00:00+00:00", self.portal.doc1.expires().ISO8601()
+        )
+
+    def test_transition_with_no_access_to_review_history_in_target_state(self):
+        self.wftool.setChainForPortalTypes(["Folder"], "restriction_workflow")
+        folder = self.portal[
+            self.portal.invokeFactory("Folder", id="folder_test", title="Test")
+        ]
+        transaction.commit()
+        uid = folder.UID()
+        setRoles(
+            self.portal, TEST_USER_ID, ["Contributor", "Editor", "Member", "Reviewer"]
+        )
+        login(self.portal, TEST_USER_NAME)
+        endpoint_url = "{0}/@wf/{1}/restrict".format(self.portal_url, uid)
+        response = requests.post(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(TEST_USER_NAME, TEST_USER_PASSWORD),
+            json={},
+        )
+        transaction.commit()
+        self.assertEqual(200, response.status_code)
+        self.assertEqual(u"restricted", self.wftool.getInfoFor(folder, u"review_state"))
diff --git a/src/imio/restapi/tests/test_service_update.py b/src/imio/restapi/tests/test_service_update.py
@@ -0,0 +1,78 @@
+# -*- coding: utf-8 -*-
+from imio.restapi.testing import IMIO_RESTAPI_DX_FUNCTIONAL_TESTING
+from plone.app.testing import login
+from plone.app.testing import setRoles
+from plone.app.testing import SITE_OWNER_NAME
+from plone.app.testing import SITE_OWNER_PASSWORD
+from plone.app.testing import TEST_USER_ID
+from plone.app.testing import TEST_USER_NAME
+from plone.app.testing import TEST_USER_PASSWORD
+from Products.CMFCore.utils import getToolByName
+
+import requests
+import transaction
+import unittest
+
+
+class TestContentPatch(unittest.TestCase):
+    layer = IMIO_RESTAPI_DX_FUNCTIONAL_TESTING
+
+    def setUp(self):
+        self.app = self.layer["app"]
+        self.portal = self.layer["portal"]
+        self.request = self.layer["request"]
+        self.portal_url = self.portal.absolute_url()
+        setRoles(self.portal, TEST_USER_ID, ["Member"])
+        login(self.portal, SITE_OWNER_NAME)
+        self.portal.invokeFactory(
+            "Document", id="doc1", title="My Document", description="Some Description"
+        )
+        wftool = getToolByName(self.portal, "portal_workflow")
+        wftool.doActionFor(self.portal.doc1, "publish")
+        transaction.commit()
+
+    def tearDown(self):
+        login(self.portal, SITE_OWNER_NAME)
+        self.portal.manage_delObjects(["doc1"])
+        transaction.commit()
+
+    def test_patch_document(self):
+        self.request["BODY"] = '{"title": "Patched Document"}'
+
+        uid = self.portal.doc1.UID()
+        endpoint_url = "{0}/@content/{1}".format(self.portal_url, uid)
+        response = requests.patch(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD),
+            json={"title": "Patched Document"},
+        )
+        transaction.commit()
+        self.assertEqual(204, response.status_code)
+        self.assertEqual("Patched Document", self.portal.doc1.Title())
+
+    def test_patch_document_will_delete_value_with_null(self):
+        self.assertEqual(self.portal.doc1.description, "Some Description")
+        uid = self.portal.doc1.UID()
+        endpoint_url = "{0}/@content/{1}".format(self.portal_url, uid)
+        response = requests.patch(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD),
+            json={"description": ""},
+        )
+        transaction.commit()
+        self.assertEqual(204, response.status_code)
+        self.assertEqual(u"", self.portal.doc1.description)
+
+    def test_patch_document_unauthorized(self):
+        uid = self.portal.doc1.UID()
+        endpoint_url = "{0}/@content/{1}".format(self.portal_url, uid)
+        response = requests.patch(
+            endpoint_url,
+            headers={"Accept": "application/json"},
+            auth=(TEST_USER_NAME, TEST_USER_PASSWORD),
+            json={"description": ""},
+        )
+        transaction.commit()
+        self.assertEqual(401, response.status_code)