From 79c13152da33f37ae2fd664db4bef10b720f3393 Mon Sep 17 00:00:00 2001
From: Ilias Katsakioris <elikatsis@arrikto.com>
Date: Tue, 10 Mar 2020 14:03:38 +0200
Subject: [PATCH 1/2] 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>
---
 sdk/python/kfp/dsl/_resource_op.py        | 18 +++++++++++++++
 sdk/python/tests/dsl/resource_op_tests.py | 27 +++++++++++++++++++++++
 sdk/python/tests/dsl/volume_op_tests.py   | 19 ++++++++++++++++
 3 files changed, 64 insertions(+)

diff --git a/sdk/python/kfp/dsl/_resource_op.py b/sdk/python/kfp/dsl/_resource_op.py
index f25195e388a..97bbee09f26 100644
--- a/sdk/python/kfp/dsl/_resource_op.py
+++ b/sdk/python/kfp/dsl/_resource_op.py
@@ -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)
diff --git a/sdk/python/tests/dsl/resource_op_tests.py b/sdk/python/tests/dsl/resource_op_tests.py
index 98b7b4cd96f..8f1f7297854 100644
--- a/sdk/python/tests/dsl/resource_op_tests.py
+++ b/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()
diff --git a/sdk/python/tests/dsl/volume_op_tests.py b/sdk/python/tests/dsl/volume_op_tests.py
index 08f85b99c36..592f0799689 100644
--- a/sdk/python/tests/dsl/volume_op_tests.py
+++ b/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()

From cc26aeddd1e8e1364523727e28227975455b6c7b Mon Sep 17 00:00:00 2001
From: Ilias Katsakioris <elikatsis@arrikto.com>
Date: Tue, 10 Mar 2020 14:03:38 +0200
Subject: [PATCH 2/2] Fix ValueError not being raised

---
 sdk/python/kfp/dsl/_resource_op.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/sdk/python/kfp/dsl/_resource_op.py b/sdk/python/kfp/dsl/_resource_op.py
index 97bbee09f26..70782c09371 100644
--- a/sdk/python/kfp/dsl/_resource_op.py
+++ b/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,