Implementing Bigtable Cluster.delete().

gcloud/bigtable/cluster.py

@@ -218,3 +218,33 @@ def create(self):
             request_pb, self._client.timeout_seconds)
 
         self._operation = cluster_pb.current_operation
+
+    def delete(self):
+        """Delete this cluster.
+
+        Marks a cluster and all of its tables for permanent deletion in 7 days.
+
+        Immediately upon completion of the request:
+
+        * Billing will cease for all of the cluster's reserved resources.
+        * The cluster's ``delete_time`` field will be set 7 days in the future.
+
+        Soon afterward:
+
+        * All tables within the cluster will become unavailable.
+
+        Prior to the cluster's ``delete_time``:
+
+        * The cluster can be recovered with a call to ``UndeleteCluster``.
+        * All other attempts to modify or delete the cluster will be rejected.
+
+        At the cluster's ``delete_time``:
+
+        * The cluster and **all of its tables** will immediately and
+          irrevocably disappear from the API, and their data will be
+          permanently deleted.
+        """
+        request_pb = messages_pb2.DeleteClusterRequest(name=self.name)
+        # We expect a `._generated.empty_pb2.Empty`
+        self._client._cluster_stub.DeleteCluster(
+            request_pb, self._client.timeout_seconds)

gcloud/bigtable/test_cluster.py

@@ -263,6 +263,44 @@ def mock_prep_create_req(cluster):
         self.assertEqual(cluster._operation, current_op)
         self.assertEqual(prep_create_called, [cluster])
 
+    def test_delete(self):
+        from gcloud.bigtable._generated import (
+            bigtable_cluster_service_messages_pb2 as messages_pb2)
+        from gcloud.bigtable._generated import empty_pb2
+        from gcloud.bigtable._testing import _FakeStub
+
+        project = 'PROJECT'
+        zone = 'zone'
+        cluster_id = 'cluster-id'
+        timeout_seconds = 57
+
+        client = _Client(project, timeout_seconds=timeout_seconds)
+        cluster = self._makeOne(zone, cluster_id, client)
+
+        # Create request_pb
+        cluster_name = ('projects/' + project + '/zones/' + zone +
+                        '/clusters/' + cluster_id)
+        request_pb = messages_pb2.DeleteClusterRequest(name=cluster_name)
+
+        # Create response_pb
+        response_pb = empty_pb2.Empty()
+
+        # Patch the stub used by the API method.
+        client._cluster_stub = stub = _FakeStub(response_pb)
+
+        # Create expected_result.
+        expected_result = None  # delete() has no return value.
+
+        # Perform the method and check the result.
+        result = cluster.delete()
+
+        self.assertEqual(result, expected_result)
+        self.assertEqual(stub.method_calls, [(
+            'DeleteCluster',
+            (request_pb, timeout_seconds),
+            {},
+        )])
+
 
 class Test__get_pb_property_value(unittest2.TestCase):