Adding small updates and tests

ckanext/pages/actions.py

@@ -141,7 +141,7 @@ def _pages_update(context, data_dict):
     revisions = out.revisions
 
     new_revision = {
-        make_uuid() : {
+        make_uuid(): {
             "content": out.content,
             "user_id": user.id,
             "created": datetime.datetime.now(datetime.timezone.utc).isoformat(),
@@ -172,28 +172,10 @@ def _pages_update(context, data_dict):
     session.commit()
 
 
-def _pages_revision_restore(context, data_dict):
-    name = data_dict.get('page')
-    rev = data_dict.get('revision')
-    page = db.Page.get(name=name)
-
-    if page and page.revisions:
-        page.revisions = _remove_keys_revision_from_dict(page.revisions)
-        revision = page.revisions.get(rev)
-
-        try:
-            revision['current'] = True
-            page.content = revision['content']
-            page.save()
-            return revision
-        except TypeError:
-            raise TypeError("Unexpected value.")
-
-
 def _remove_keys_revision_from_dict(data_dict, keys=['current']):
     return {
-        id:{
-            key:data_dict[id][key] for key in data_dict[id] if key not in keys
+        id: {
+            key: data_dict[id][key] for key in data_dict[id] if key not in keys
             } for id in data_dict
     }
 
@@ -256,7 +238,21 @@ def pages_update(context, data_dict):
 
 def pages_revision_restore(context, data_dict):
     p.toolkit.check_access('ckanext_pages_update', context, data_dict)
-    return _pages_revision_restore(context, data_dict)
+    name = data_dict.get('page')
+    rev = data_dict.get('revision')
+    page = db.Page.get(name=name)
+
+    if page and page.revisions:
+        page.revisions = _remove_keys_revision_from_dict(page.revisions)
+        revision = page.revisions.get(rev)
+
+        try:
+            revision['current'] = True
+            page.content = revision['content']
+            page.save()
+            return revision
+        except TypeError:
+            raise TypeError("Unexpected value.")
 
 
 def pages_delete(context, data_dict):

ckanext/pages/tests/test_action.py

@@ -1,4 +1,6 @@
 import pytest
+import datetime
+from collections import OrderedDict
 
 from ckan.tests import factories, helpers
 
@@ -54,6 +56,54 @@ def test_pages_update_action(self, app):
         assert page["title"] == "New Page Updated"
         assert page["content"] == "This is a test content updated"
 
+    def test_pages_revision_restore_action(self, app):
+        user = factories.User()
+        helpers.call_action(
+            "ckanext_pages_update",
+            {"user": user["name"]},
+            name="page_name",
+            title="First Revision Title",
+            content="First Revision Content",
+        )
+
+        helpers.call_action(
+            "ckanext_pages_update",
+            {"user": user["name"]},
+            name="page_name",
+            title="Page Updated",
+            content="This is a test content updated",
+            page="page_name",
+        )
+
+        page = helpers.call_action("ckanext_pages_show", {}, page="page_name")
+
+        revisions = page.get('revisions')
+
+        assert len(revisions) == 2
+        assert page['content'] == "This is a test content updated"
+
+        sorted_revisions = OrderedDict(reversed(sorted(
+                revisions.items(),
+                key=lambda x: datetime.datetime.timestamp(
+                    datetime.datetime.fromisoformat(x[1]['created'])
+                    )
+        )))
+
+        last_revision = sorted_revisions.popitem()
+
+        helpers.call_action(
+            "ckanext_pages_revision_restore",
+            {"user": user["name"]},
+            page="page_name",
+            revision=last_revision[0]
+        )
+
+        page = helpers.call_action("ckanext_pages_show", {}, page="page_name")
+
+        assert page['title'] == "Page Updated"
+        assert page['content'] == "First Revision Content"
+        assert page['revisions'][last_revision[0]]['current']
+
     def test_pages_list(self, app):
         sysadmin = factories.Sysadmin()
         helpers.call_action(