Added more tests to the test suite.

Author: anthony-tuininga

tests/test_1300_cursor_var.py

@@ -231,6 +231,48 @@ def cursor(self):
         cursor.fetchall()
         self.assertEqual(Counter.num_cursors_created, 3)
 
+    def test_1312_fetch_nested_cursors_for_complex_sql(self):
+        "1312 - test that nested cursors are fetched correctly"
+        sql = """
+            select
+            'Level 1 String',
+            cursor(
+                select
+                'Level 2 String',
+                cursor(
+                    select
+                    'Level 3 String',
+                    cursor(
+                        select 1, 'Level 4 String A' from dual
+                        union all
+                        select 2, 'Level 4 String B' from dual
+                        union all
+                        select 3, 'Level 4 String C' from dual
+                    ) as nc3
+                    from dual
+                ) as nc2
+                from dual
+            ) as nc1
+            from dual"""
+        self.cursor.execute(sql)
+        transform_fn = lambda v: \
+                [transform_row(r) for r in v] \
+                if isinstance(v, oracledb.Cursor) \
+                else v
+        transform_row = lambda r: tuple(transform_fn(v) for v in r)
+        rows = [transform_row(r) for r in self.cursor]
+        expected_value = [
+            ('Level 1 String', [
+                ('Level 2 String', [
+                    ('Level 3 String', [
+                        (1, 'Level 4 String A'),
+                        (2, 'Level 4 String B'),
+                        (3, 'Level 4 String C')
+                    ]),
+                ]),
+            ])
+        ]
+        self.assertEqual(rows, expected_value)
 
 if __name__ == "__main__":
     test_env.run_test_cases()

tests/test_2400_pool.py

@@ -693,5 +693,18 @@ def test_2428_acquire_conn_from_empty_pool(self):
                 result, = cursor.fetchone()
                 self.assertEqual(result, test_env.get_main_user().upper())
 
+    @unittest.skipIf(test_env.get_is_thin(),
+                     "thin mode doesn't support soda_metadata_cache" \
+                     "parameter yet")
+    def test_2429_soda_metadata_cache(self):
+        "2429 - test soda_metadata_cache parameter"
+        pool = test_env.get_pool(min=0, max=2, increment=2)
+        self.assertEqual(pool.soda_metadata_cache, False)
+        pool = test_env.get_pool(min=0, max=2, increment=2,
+                                 soda_metadata_cache=True)
+        self.assertEqual(pool.soda_metadata_cache, True)
+        pool.soda_metadata_cache = False
+        self.assertEqual(pool.soda_metadata_cache, False)
+
 if __name__ == "__main__":
     test_env.run_test_cases()

tests/test_3400_soda_collection.py

@@ -27,6 +27,7 @@
 """
 
 import unittest
+import json
 
 import oracledb
 import test_env
@@ -368,5 +369,82 @@ def test_3415_soda_hint_with_invalid_type(self):
         self.assertRaises(TypeError, coll.saveAndGet,
                           dict(name="Sally", age=36), hint=5)
 
+    def test_3416_collection_name_and_metadata(self):
+        "3416 - test name and metadata attribute"
+        soda_db = self.connection.getSodaDatabase()
+        collection_name = "TestCollectionMetadata"
+        coll = soda_db.createCollection(collection_name)
+        self.assertEqual(coll.name, collection_name)
+        self.assertEqual(coll.metadata["tableName"], collection_name)
+        coll.drop()
+
+    def test_3417_insert_many(self):
+        "3417 - test insertMany"
+        soda_db = self.get_soda_database(minclient=(18, 5))
+        coll = soda_db.createCollection("TestInsertMany")
+        values_to_insert = [
+            dict(name="George", age=25),
+            soda_db.createDocument(dict(name="Lucas", age=47))
+        ]
+        coll.insertMany(values_to_insert)
+        self.connection.commit()
+        fetched_docs = list(coll.find().getCursor())
+        for fetched_doc, expected_doc in zip(fetched_docs, values_to_insert):
+            if isinstance(expected_doc, dict):
+                expected_doc = soda_db.createDocument(expected_doc)
+            self.assertEqual(fetched_doc.getContent(),
+                             expected_doc.getContent())
+        self.assertRaisesRegex(oracledb.DatabaseError, "^DPI-1031:",
+                               coll.insertMany, [])
+        coll.drop()
+
+    def test_3418_save(self):
+        "3418 - test save"
+        soda_db = self.get_soda_database(minclient=(19, 9))
+        coll = soda_db.createCollection("TestSodaSave")
+        values_to_save = [
+            12,
+            "23",
+            soda_db.createDocument(45)
+        ]
+        for value in values_to_save:
+            coll.save(value)
+        self.connection.commit()
+        fetched_docs = coll.find().getDocuments()
+        for fetched_doc, expected_doc in zip(fetched_docs, values_to_save):
+            if isinstance(expected_doc, (int, str)):
+                expected_doc = soda_db.createDocument(expected_doc)
+            self.assertEqual(fetched_doc.getContent(),
+                             expected_doc.getContent())
+        coll.drop()
+
+    def test_3419_save_and_get_with_hint(self):
+        "3419 - test saveAndGet with hint"
+        soda_db = self.get_soda_database(minclient=(19, 11))
+        cursor = self.connection.cursor()
+        statement = """
+                SELECT
+                    ( SELECT t2.sql_fulltext
+                      FROM v$sql t2
+                      WHERE t2.sql_id = t1.prev_sql_id
+                        AND t2.child_number = t1.prev_child_number
+                    )
+                FROM v$session t1
+                WHERE t1.audsid = sys_context('userenv', 'sessionid')"""
+        coll = soda_db.createCollection("TestSodaSaveWithHint")
+        coll.find().remove()
+        values_to_save = [
+            dict(name="Jordan", age=59),
+            dict(name="Curry", age=34)
+        ]
+        hints = ["MONITOR", "NO_MONITOR"]
+
+        for value, hint in zip(values_to_save, hints):
+            coll.saveAndGet(value, hint=hint)
+            coll.find().hint(hint).getOne().getContent()
+            cursor.execute(statement)
+            result, = cursor.fetchone()
+            self.assertTrue(hint in result.read())
+
 if __name__ == "__main__":
     test_env.run_test_cases()