SDK/DSL: Enable the deletion of a resource via ResourceOp method (#3213)

* SDK/DSL: Enable the deletion of a resource via ResourceOp method

* Add the method delete() to ResourceOps
* Extend ResourceOp & VolumeOp tests

Signed-off-by: Ilias Katsakioris <elikatsis@arrikto.com>

* Fix ValueError not being raised

sdk/python/kfp/dsl/_resource_op.py

@@ -101,14 +101,14 @@ def __init__(self,
         ])
 
         if k8s_resource is None:
-            ValueError("You need to provide a k8s_resource.")
+            raise ValueError("You need to provide a k8s_resource.")
 
         if merge_strategy and action != "apply":
-            ValueError("You can't set merge_strategy when action != 'apply'")
+            raise ValueError("You can't set merge_strategy when action != 'apply'")
 
         # if action is delete, there should not be any outputs, success_condition, and failure_condition
         if action == "delete" and (success_condition or failure_condition or attribute_outputs):
-            ValueError("You can't set success_condition, failure_condition, or attribute_outputs when action == 'delete'")
+            raise ValueError("You can't set success_condition, failure_condition, or attribute_outputs when action == 'delete'")
 
         init_resource = {
             "action": action,
@@ -158,3 +158,21 @@ def resource(self):
         `io.argoproj.workflow.v1alpha1.Template`.
         """
         return self._resource
+
+    def delete(self):
+        """Returns a ResourceOp which deletes the resource."""
+        if self.resource.action == "delete":
+            raise ValueError("This operation is already a resource deletion.")
+
+        k8s_resource = dict()
+        if isinstance(self.k8s_resource, dict):
+            k8s_resource["apiVersion"] = self.k8s_resource["apiVersion"]
+            k8s_resource["kind"] = self.k8s_resource["kind"]
+        else:
+            k8s_resource["apiVersion"] = self.k8s_resource.api_version
+            k8s_resource["kind"] = self.k8s_resource.kind
+        k8s_resource["metadata"] = {"name": self.outputs["name"]}
+
+        return ResourceOp(name="del-%s" % self.name,
+                          action="delete",
+                          k8s_resource=k8s_resource)

sdk/python/tests/dsl/resource_op_tests.py

@@ -69,3 +69,30 @@ def my_pipeline(param):
             self.assertEqual(res.dependent_names, [])
 
         kfp.compiler.Compiler()._compile(my_pipeline)
+
+    def test_delete(self):
+        """Test delete method."""
+        param = PipelineParam("param")
+        k8s_resource = {"apiVersion": "version",
+                        "kind": "CustomResource",
+                        "metadata": {"name": "my-resource"}}
+        res = ResourceOp(name="resource",
+                         k8s_resource=k8s_resource,
+                         success_condition=param,
+                         attribute_outputs={"test": "attr"})
+
+        delete_res = res.delete()
+
+        self.assertEqual(delete_res.resource.action, "delete")
+        self.assertEqual(delete_res.attribute_outputs, {})
+        self.assertEqual(delete_res.outputs, {})
+        self.assertEqual(delete_res.output, None)
+
+        expected_res_name = PipelineParam(name="name", op_name=res.name)
+        expected_res_resource = {"apiVersion": "version",
+                                 "kind": "CustomResource",
+                                 "metadata": {"name": expected_res_name}}
+        self.assertEqual(delete_res.k8s_resource, expected_res_resource)
+
+        with self.assertRaises(ValueError):
+            delete_res.delete()

sdk/python/tests/dsl/volume_op_tests.py

@@ -66,3 +66,22 @@ def my_pipeline(param1, param2):
             )
 
         kfp.compiler.Compiler()._compile(my_pipeline)
+
+    def test_delete(self):
+        """Test delete method."""
+        vop = VolumeOp(name="vop", resource_name="vop", size="1Gi")
+
+        delete_vop = vop.delete()
+
+        self.assertEqual(delete_vop.resource.action, "delete")
+        self.assertEqual(delete_vop.attribute_outputs, {})
+        self.assertEqual(delete_vop.outputs, {})
+        self.assertEqual(delete_vop.output, None)
+        expected_name = PipelineParam(name="name", op_name=vop.name)
+        expected_resource = {"apiVersion": "v1",
+                             "kind": "PersistentVolumeClaim",
+                             "metadata": {"name": expected_name}}
+        self.assertEqual(delete_vop.k8s_resource, expected_resource)
+
+        with self.assertRaises(ValueError):
+            delete_vop.delete()