Add nested tuple tests

tests/integration/standard/test_types.py

@@ -503,9 +503,9 @@ class TypeTests(unittest.TestCase):
         s = c.connect()
         s.encoders[tuple] = cql_encode_tuple
 
-        s.execute("""CREATE KEYSPACE test_tuple_types
+        s.execute("""CREATE KEYSPACE test_tuple_subtypes
             WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor': '1'}""")
-        s.set_keyspace("test_tuple_types")
+        s.set_keyspace("test_tuple_subtypes")
 
         s.execute("CREATE TABLE mytable ("
                   "k int PRIMARY KEY, "
@@ -523,6 +523,69 @@ class TypeTests(unittest.TestCase):
             result = s.execute("SELECT v FROM mytable WHERE k=%s", (i,))[0]
             self.assertEqual(response_tuple, result.v)
 
+    def nested_tuples_schema_helper(self, depth):
+        """
+        Helper method for creating nested tuple schema
+        """
+
+        if depth == 0:
+            return 'int'
+        else:
+            return 'tuple<%s>' % self.nested_tuples_schema_helper(depth - 1)
+
+    def nested_tuples_creator_helper(self, depth):
+        """
+        Helper method for creating nested tuples
+        """
+
+        if depth == 0:
+            return 303
+        else:
+            return tuple((self.nested_tuples_creator_helper(depth - 1),))
+
+    def test_nested_tuples(self):
+        """
+        Ensure nested are appropriately handled.
+        """
+
+        if self._cass_version < (2, 1, 0):
+            raise unittest.SkipTest("The tuple type was introduced in Cassandra 2.1")
+
+        c = Cluster(protocol_version=PROTOCOL_VERSION)
+        s = c.connect()
+
+        # set the row_factory to dict_factory for programmatic access
+        # set the encoder for tuples for the ability to write tuples
+        s.row_factory = dict_factory
+        s.encoders[tuple] = cql_encode_tuple
+
+        s.execute("""CREATE KEYSPACE test_nested_tuples
+            WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor': '1'}""")
+        s.set_keyspace("test_nested_tuples")
+
+        # create a table with multiple sizes of nested tuples
+        s.execute("CREATE TABLE mytable ("
+                  "k int PRIMARY KEY, "
+                  "v_1 %s,"
+                  "v_2 %s,"
+                  "v_3 %s,"
+                  "v_128 %s"
+                  ")" % (self.nested_tuples_schema_helper(1),
+                        self.nested_tuples_schema_helper(2),
+                        self.nested_tuples_schema_helper(3),
+                        self.nested_tuples_schema_helper(128)))
+
+        for i in (1, 2, 3, 128):
+            # create tuple
+            created_tuple = self.nested_tuples_creator_helper(i)
+
+            # write tuple
+            s.execute("INSERT INTO mytable (k, v_%s) VALUES (%s, %s)", (i, i, created_tuple))
+
+            # verify tuple was written and read correctly
+            result = s.execute("SELECT v_%s FROM mytable WHERE k=%s", (i, i))[0]
+            self.assertEqual(created_tuple, result['v_%s' % i])
+
     def test_unicode_query_string(self):
         c = Cluster(protocol_version=PROTOCOL_VERSION)
         s = c.connect()