Implemented publish queue class (#39) (#40)

Author: sunil-lakshman

.talismanrc

@@ -304,4 +304,12 @@ fileignoreconfig:
 - filename: tests/unit/management_token/test_management_token_unit.py
   checksum: ddb2b5dd54b6ac762fa78936196a4b2088d82254c0ee25fccae290375d1f71e7
 version: ""
+fileignoreconfig:
+- filename: tests/api/publish_queue/test_publish_queue_api.py
+  checksum: 8364fcd077ec8e99556738460d0659e2ec7bf8747969428f2d99761ef3ae4020
+- filename: tests/mock/publish_queue/test_publish_queue_mock.py
+  checksum: 6581d4def863f65e0321e0ff3e9ee074aa5ae1a53bf46035ea01b41e9de0bc36
+- filename: tests/unit/publish_queue/test_publish_queue_unit.py
+  checksum: add55174c8c3ee4e931433e9bf54a80304d6927681912a19e5ee78747dc608bf
+version: ""
 

contentstack_management/__init__.py

@@ -29,6 +29,7 @@
 from .release_items.release_item import ReleaseItems
 from .delivery_token.delivery_token import DeliveryToken
 from .management_token.management_token import ManagementToken
+from .publish_queue.publish_queue import PublishQueue
 
 
 __all__ = (
@@ -61,7 +62,8 @@
 "Releases",
 "ReleaseItems",
 "DeliveryToken",
-"ManagementToken"
+"ManagementToken",
+"PublishQueue"
 )
 
 __title__ = 'contentstack-management-python'

contentstack_management/publish_queue/__init__.py

@@ -0,0 +1 @@
+from contentstack_management import contentstack

contentstack_management/publish_queue/publish_queue.py

@@ -0,0 +1,88 @@
+"""This class takes a base URL as an argument when it's initialized, 
+which is the endpoint for the RESTFUL API that we'll be interacting with.
+The create(), read(), update(), and delete() methods each correspond to 
+the CRUD operations that can be performed on the API """
+
+import json
+from ..common import Parameter
+from urllib.parse import quote
+from .._errors import ArgumentException
+
+class PublishQueue(Parameter):
+    """
+    This class takes a base URL as an argument when it's initialized, 
+    which is the endpoint for the RESTFUL API that
+    we'll be interacting with. The create(), read(), update(), and delete() 
+    methods each correspond to the CRUD 
+    operations that can be performed on the API """
+
+    def __init__(self, client, publish_queue_uid: str):
+        self.client = client
+        self.publish_queue_uid = publish_queue_uid
+        super().__init__(self.client)
+
+        self.path = "publish-queue"
+
+    def find(self):
+        """
+        The "Get publish queue" request returns comprehensive information on activities such as publish, unpublish, and delete performed on entries and/or assets. 
+        This request also includes the details of the release deployments in the response body.
+        :return: Json, with publish_queue details.
+
+        -------------------------------
+        [Example:]
+
+            >>> from contentstack_management import contentstack
+            >>> client = contentstack.client(authtoken='your_authtoken')
+            >>> result = client.stack("api_key").publish_queue().find().json()
+
+        -------------------------------
+        """        
+        return self.client.get(self.path, headers = self.client.headers)
+    
+      
+    
+    def fetch(self):
+        """
+        The "Get publish queue activity" request returns comprehensive information on a specific publish, unpublish, or delete action performed on an entry and/or asset. You can also retrieve details of a specific release deployment.
+
+        :return: Json, with publish_queue details.
+        -------------------------------
+        [Example:]
+
+            >>> from contentstack_management import contentstack
+            >>> client = contentstack.client(authtoken='your_authtoken')
+            >>> result = client.stack('api_key').publish_queue('publish_queue_uid').fetch().json()
+
+        -------------------------------
+        """
+        self.validate_uid()
+        url = f"{self.path}/{self.publish_queue_uid}"
+        return self.client.get(url, headers = self.client.headers)
+        
+    
+    def cancel(self):
+        """
+        The "Cancel Scheduled Action" request will allow you to cancel any scheduled publishing or unpublishing activity of entries and/or assets and cancel the deployment of releases.
+
+        :param data: The `data` parameter is the payload that you want to send in the request body. It
+        should be a dictionary or a JSON serializable object that you want to send as the request body
+        :return: Json, with publish_queue details.
+
+        -------------------------------
+        [Example:]
+            >>> from contentstack_management import contentstack
+            >>> client = contentstack.client(authtoken='your_authtoken')
+            >>> result = client.stack('api_key').publish_queue().create(data).json()
+
+        -------------------------------
+        """
+        
+        self.validate_uid()
+        url = f"{self.path}/{self.publish_queue_uid}/unschedule"
+        return self.client.get(url, headers = self.client.headers)
+    
+    
+    def validate_uid(self):
+         if self.publish_queue_uid is None or '':
+            raise ArgumentException("Publish Queue Uid is required")

contentstack_management/stack/stack.py

@@ -19,6 +19,7 @@
 from ..releases.release import Releases
 from ..delivery_token.delivery_token import DeliveryToken
 from ..management_token.management_token import ManagementToken
+from ..publish_queue.publish_queue import PublishQueue
 
 
 class Stack(Parameter):
@@ -367,4 +368,7 @@ def delivery_token(self, delivery_token: str = None):
             return DeliveryToken(self.client, delivery_token)
 
     def management_token(self, management_token: str = None):
-            return ManagementToken(self.client, management_token)  
+            return ManagementToken(self.client, management_token)  
+    
+    def publish_queue(self, publish_queue_uid: str = None):
+            return PublishQueue(self.client, publish_queue_uid)  

tests/api/publish_queue/test_publish_queue_api.py

@@ -0,0 +1,40 @@
+import os
+import unittest
+from dotenv import load_dotenv
+from contentstack_management import contentstack
+from tests.cred import get_credentials
+
+credentials = get_credentials()
+username = credentials["username"]
+password = credentials["password"]
+host = credentials["host"]
+api_key = credentials["api_key"]
+publish_queue_uid = credentials["publish_queue_uid"]
+
+class publish_queueApiTests(unittest.TestCase):
+
+    def setUp(self):
+        self.client = contentstack.ContentstackClient(host=host)
+        self.client.login(username, password)
+
+    def test_get_all_publish_queue(self):
+        response = self.client.stack(api_key).publish_queue().find()
+        self.assertEqual(response.request.url, f"{self.client.endpoint}publish-queue")
+        self.assertEqual(response.status_code, 200)
+        
+
+    def test_get_a_publish_queue(self):
+        response = self.client.stack(api_key).publish_queue(publish_queue_uid).fetch()
+        self.assertEqual(response.request.url,
+                         f"{self.client.endpoint}publish-queue/{publish_queue_uid}")
+        self.assertEqual(response.status_code, 200)
+        
+
+    def test_cancel_a_publish_queue(self):
+        response = self.client.stack(api_key).publish_queue(publish_queue_uid).cancel()
+        self.assertEqual(response.request.url,
+                         f"{self.client.endpoint}publish-queue/{publish_queue_uid}/unschedule")
+        self.assertEqual(response.status_code, 200)
+        
+
+    

tests/cred.py

@@ -39,6 +39,7 @@
 default_release_uid = "release_uid" #default release uid
 default_delivery_token_uid = "delivery_token_uid" #default delivery token uid
 default_management_token_uid = "management_token_uid" #default management token uid
+default_publish_queue_uid = "publish_queue_uid" # default publish queue uid
 
 def get_credentials():
     load_dotenv()
@@ -83,6 +84,7 @@ def get_credentials():
         "release_uid": os.getenv("RELEASE_UID", default_release_uid),
         "delivery_token_uid": os.getenv("DELIVERY_TOKEN_UID", default_delivery_token_uid),
         "management_token_uid": os.getenv("MANAGEMENT_TOKEN_UID", default_management_token_uid),
+        "publish_queue_uid": os.getenv("PUBLISH_QUEUE_UID", default_publish_queue_uid),
 
     }
     return credentials

tests/mock/publish_queue/test_publish_queue_mock.py

@@ -0,0 +1,51 @@
+import json
+import os
+import unittest
+
+from dotenv import load_dotenv
+from contentstack_management import contentstack
+from tests.cred import get_credentials
+
+credentials = get_credentials()
+username = credentials["username"]
+password = credentials["password"]
+api_key = credentials["api_key"]
+host = credentials["host"]
+publish_queue_uid = credentials["publish_queue_uid"]
+
+
+class publish_queueMockTests(unittest.TestCase):
+
+    def setUp(self):
+        
+        self.client = contentstack.ContentstackClient(host = host)
+        self.client.login(username, password)
+
+    
+    def read_file(self, file_name):
+        file_path= f"tests/resources/mock_publish_queue/{file_name}"
+        infile = open(file_path, 'r')
+        data = infile.read()
+        infile.close()
+        return data
+
+
+    def test_get_all_publish_queue(self):
+        response = self.client.stack(api_key).publish_queue().find().json()
+        read_mock_publish_queue_data  = self.read_file("find.json")
+        mock_publish_queue_data = json.loads(read_mock_publish_queue_data)
+        self.assertEqual(mock_publish_queue_data.keys(), response.keys())
+
+    def test_get_a_publish_queue(self):
+        response = self.client.stack(api_key).publish_queue(publish_queue_uid).fetch().json()
+        read_mock_publish_queue_data  = self.read_file("fetch.json")
+        mock_publish_queue_data = json.loads(read_mock_publish_queue_data)
+        self.assertEqual(mock_publish_queue_data.keys(), response.keys())
+
+    def test_cancel_a_publish_queue(self):
+        response = self.client.stack(api_key).publish_queue(publish_queue_uid).cancel().json()
+        read_mock_publish_queue_data  = self.read_file("cancel.json")
+        mock_publish_queue_data = json.loads(read_mock_publish_queue_data)
+        self.assertEqual(mock_publish_queue_data.keys(), response.keys())
+
+    

tests/resources/mock_publish_queue/cancel.json

@@ -0,0 +1,3 @@
+{
+    "notice": "Release unscheduled successfully."
+ }

tests/resources/mock_publish_queue/fetch.json

@@ -0,0 +1,32 @@
+{
+    "entry": {
+       "uid": "publish_queue_uid",
+       "stack": "api_key",
+       "created_at": "2023-09-27T17:50:55.624Z",
+       "updated_at": "2023-09-27T17:50:55.624Z",
+       "created_by": "created_by_uid",
+       "updated_by": "updated_by_uid",
+       "type": "release",
+       "publish_details": {
+          "status": "success"
+       },
+       "entry": {
+          "title": "New_Name",
+          "uid": "entry_uid",
+          "skip_items_webhook_trigger": false,
+          "release": {
+             "uid": "release_uid",
+             "title": "New_Name"
+          }
+       },
+       "environment": [
+          "environment_uid"
+       ],
+       "action": "deploy",
+       "published_at": "2023-09-27T17:50:55.536Z",
+       "user": "user_uid",
+       "approval": false,
+       "approved": true,
+       "rejected": false
+    }
+ }

tests/resources/mock_publish_queue/find.json

@@ -0,0 +1,64 @@
+{
+    "queue": [
+       {
+          "uid": "queue_uid",
+          "stack": "api_key",
+          "created_at": "2023-09-28T10:48:20.416Z",
+          "updated_at": "2023-09-28T10:48:20.416Z",
+          "created_by": "created_by_uid",
+          "updated_by": "updated_by_uid",
+          "type": "release",
+          "publish_details": {
+             "status": "success"
+          },
+          "entry": {
+             "title": "tesrting2",
+             "uid": "entry_uid",
+             "skip_items_webhook_trigger": false,
+             "release": {
+                "uid": "release_uid",
+                "title": "tesrting2"
+             }
+          },
+          "environment": [
+             "environment_uid"
+          ],
+          "action": "deploy",
+          "published_at": "2023-09-28T10:48:20.334Z",
+          "user": "user_uid",
+          "approval": false,
+          "approved": true,
+          "rejected": false
+       },
+       {
+          "uid": "queue_uid",
+          "stack": "api_key",
+          "created_at": "2023-09-27T18:32:01.934Z",
+          "updated_at": "2023-09-27T18:32:01.934Z",
+          "created_by": "created_by_uid",
+          "updated_by": "updated_by_uid",
+          "type": "release",
+          "publish_details": {
+             "status": "success"
+          },
+          "entry": {
+             "title": "New_Name",
+             "uid": "entry_uid",
+             "skip_items_webhook_trigger": false,
+             "release": {
+                "uid": "release_uid",
+                "title": "New_Name"
+             }
+          },
+          "environment": [
+             "environment_uid"
+          ],
+          "action": "deploy",
+          "published_at": "2023-09-27T18:32:01.780Z",
+          "user": "user_uid",
+          "approval": false,
+          "approved": true,
+          "rejected": false
+       }
+    ]
+ }