Implementing Bigtable Cluster.list_tables().

This uses the table stub instead of the cluster stub. This
method is actually talking to a different service than
every method until now.

Also implementing Table.__eq__ so table comparison succeeds
in unit tests.

gcloud/bigtable/cluster.py

@@ -20,6 +20,8 @@
 from gcloud.bigtable._generated import bigtable_cluster_data_pb2 as data_pb2
 from gcloud.bigtable._generated import (
     bigtable_cluster_service_messages_pb2 as messages_pb2)
+from gcloud.bigtable._generated import (
+    bigtable_table_service_messages_pb2 as table_messages_pb2)
 from gcloud.bigtable.table import Table
 
 
@@ -248,3 +250,27 @@ def delete(self):
         # We expect a `._generated.empty_pb2.Empty`
         self._client._cluster_stub.DeleteCluster(
             request_pb, self._client.timeout_seconds)
+
+    def list_tables(self):
+        """List the tables in this cluster.
+
+        :rtype: list of :class:`Table <gcloud.bigtable.table.Table>`
+        :returns: The list of tables owned by the cluster.
+        :raises: :class:`ValueError <exceptions.ValueError>` if one of the
+                 returned tables has a name that is not of the expected format.
+        """
+        request_pb = table_messages_pb2.ListTablesRequest(name=self.name)
+        # We expect a `table_messages_pb2.ListTablesResponse`
+        table_list_pb = self._client._table_stub.ListTables(
+            request_pb, self._client.timeout_seconds)
+
+        result = []
+        for table_pb in table_list_pb.tables:
+            before, table_id = table_pb.name.split(
+                self.name + '/tables/', 1)
+            if before != '':
+                raise ValueError('Table name %s not of expected format' % (
+                    table_pb.name,))
+            result.append(self.table(table_id))
+
+        return result

gcloud/bigtable/table.py

@@ -55,3 +55,12 @@ def row(self, row_key):
         :returns: A row owned by this table.
         """
         return Row(row_key, self)
+
+    def __eq__(self, other):
+        if not isinstance(other, self.__class__):
+            return False
+        return (other.table_id == self.table_id and
+                other._cluster == self._cluster)
+
+    def __ne__(self, other):
+        return not self.__eq__(other)

gcloud/bigtable/test_cluster.py

@@ -301,6 +301,71 @@ def test_delete(self):
             {},
         )])
 
+    def _list_tables_helper(self, table_id, table_name=None):
+        from gcloud.bigtable._generated import (
+            bigtable_table_data_pb2 as table_data_pb2)
+        from gcloud.bigtable._generated import (
+            bigtable_table_service_messages_pb2 as table_messages_pb2)
+        from gcloud.bigtable._testing import _FakeStub
+
+        project = 'PROJECT'
+        zone = 'zone'
+        cluster_id = 'cluster-id'
+        timeout_seconds = 45
+
+        client = _Client(project, timeout_seconds=timeout_seconds)
+        cluster = self._makeOne(zone, cluster_id, client)
+
+        # Create request_
+        cluster_name = ('projects/' + project + '/zones/' + zone +
+                        '/clusters/' + cluster_id)
+        request_pb = table_messages_pb2.ListTablesRequest(name=cluster_name)
+
+        # Create response_pb
+        table_name = table_name or (cluster_name + '/tables/' + table_id)
+        response_pb = table_messages_pb2.ListTablesResponse(
+            tables=[
+                table_data_pb2.Table(name=table_name),
+            ],
+        )
+
+        # Patch the stub used by the API method.
+        client._table_stub = stub = _FakeStub(response_pb)
+
+        # Create expected_result.
+        expected_table = cluster.table(table_id)
+        expected_result = [expected_table]
+
+        # Perform the method and check the result.
+        result = cluster.list_tables()
+
+        self.assertEqual(result, expected_result)
+        self.assertEqual(stub.method_calls, [(
+            'ListTables',
+            (request_pb, timeout_seconds),
+            {},
+        )])
+
+    def test_list_tables(self):
+        table_id = 'table_id'
+        self._list_tables_helper(table_id)
+
+    def test_list_tables_failure_bad_split(self):
+        with self.assertRaises(ValueError):
+            self._list_tables_helper(None, table_name='wrong-format')
+
+    def test_list_tables_failure_name_bad_before(self):
+        project = 'PROJECT'
+        zone = 'zone'
+        cluster_id = 'cluster-id'
+
+        table_id = 'table_id'
+        bad_table_name = ('nonempty-section-before' +
+                          'projects/' + project + '/zones/' + zone +
+                          '/clusters/' + cluster_id + '/tables/' + table_id)
+        with self.assertRaises(ValueError):
+            self._list_tables_helper(table_id, table_name=bad_table_name)
+
 
 class Test__get_pb_property_value(unittest2.TestCase):
 

gcloud/bigtable/test_table.py

@@ -56,3 +56,28 @@ def test_row_factory(self):
         self.assertTrue(isinstance(row, Row))
         self.assertEqual(row._row_key, row_key)
         self.assertEqual(row._table, table)
+
+    def test___eq__(self):
+        table_id = 'table_id'
+        cluster = object()
+        table1 = self._makeOne(table_id, cluster)
+        table2 = self._makeOne(table_id, cluster)
+        self.assertEqual(table1, table2)
+
+    def test___eq__type_differ(self):
+        table1 = self._makeOne('table_id', None)
+        table2 = object()
+        self.assertNotEqual(table1, table2)
+
+    def test___ne__same_value(self):
+        table_id = 'table_id'
+        cluster = object()
+        table1 = self._makeOne(table_id, cluster)
+        table2 = self._makeOne(table_id, cluster)
+        comparison_val = (table1 != table2)
+        self.assertFalse(comparison_val)
+
+    def test___ne__(self):
+        table1 = self._makeOne('table_id1', 'cluster1')
+        table2 = self._makeOne('table_id2', 'cluster2')
+        self.assertNotEqual(table1, table2)