uncordon the node

piqe_ocp_lib/api/resources/ocp_nodes.py

@@ -343,6 +343,53 @@ def mark_node_unschedulable(self, node_name: str) -> Optional[ResourceInstance]:
                 logger.error("Exception encountered while marking node unschedulable: %s\n", e)
         return api_response
 
+    def make_node_cordon(self, node_name: str) -> bool:
+        """
+        Make node cordon and return whether it's cordon or not
+        :param node_name:
+        :return:
+        """
+        api_response = None
+        body = {
+            "spec": {
+                "taints": [{"effect": "NoSchedule", "key": "node.kubernetes.io/unschedulable"}],
+                "unschedulable": True,
+            }
+        }
+        try:
+            api_response = self.ocp_nodes.patch(name=node_name, body=body)
+            logger.info("Node %s maked cordon" % node_name)
+        except ApiException as e:
+            logger.error("Exception encountered while making node cordon: %s\n", e)
+        if "'unschedulable': True" in str(api_response.spec):
+            made_cordon = True
+        else:
+            made_cordon = False
+        return made_cordon
+
+    def make_node_uncordon(self, node_name: str) -> bool:
+        """
+        Make node uncordon and return whether it's uncordon or not
+        :param node_name:
+        :return:
+        """
+        body = {
+            "spec": {
+                "taints": [{"effect": "NoSchedule", "key": "node.kubernetes.io/unschedulable"}],
+                "unschedulable": False,
+            }
+        }
+        try:
+            api_response = self.ocp_nodes.patch(name=node_name, body=body)
+            logger.info("Node %s maked uncordon" % node_name)
+        except ApiException as e:
+            logger.error("Exception encountered while making node uncordon: %s\n", e)
+        if "'unschedulable': True" not in str(api_response.spec):
+            made_uncordon = True
+        else:
+            made_uncordon = False
+        return made_uncordon
+
     def are_all_nodes_ready(self) -> bool:
         """
         Return the status of all node based on the condition type Ready.

piqe_ocp_lib/tests/resources/test_ocp_nodes.py

@@ -340,6 +340,22 @@ def test_mark_node_schedulable(self, setup_params):
             api_response = node_api_obj.mark_node_schedulable(node_name=worker_node_name.metadata.name)
             assert api_response.kind == "Node"
 
+    def test_make_node_cordon(self, setup_params):
+        """
+        Check if we can make node cordon
+        and make the node uncordon again
+        :param setup_params:
+        :return:
+        """
+        node_api_obj = setup_params["node_api_obj"]
+        worker_node_list = node_api_obj.get_worker_nodes()
+        for worker_node_name in worker_node_list.items:
+            made_cordon = node_api_obj.make_node_cordon(node_name=worker_node_name.metadata.name)
+            assert made_cordon is True
+        for worker_node_name in worker_node_list.items:
+            made_uncordon = node_api_obj.make_node_uncordon(node_name=worker_node_name.metadata.name)
+            assert made_uncordon is True
+
     def test_are_all_nodes_ready(self, setup_params):
         """
         Verify that all nodes status is Ready.