Additional test cases to improve code coverage.

anthony-tuininga · anthony-tuininga · commit ef6088457054 · 2018-01-11T16:24:08.000-07:00
diff --git a/test/Connection.py b/test/Connection.py
@@ -229,3 +229,11 @@ def testThreading(self):
         for thread in threads:
             thread.join()
 
+    def testStringFormat(self):
+        "test string format of connection"
+        connection = cx_Oracle.connect(self.username, self.password,
+                self.tnsentry)
+        expectedValue = "<cx_Oracle.Connection to %s@%s>" % \
+                (self.username, self.tnsentry)
+        self.assertEqual(str(connection), expectedValue)
+
diff --git a/test/Cursor.py b/test/Cursor.py
@@ -455,3 +455,25 @@ def testSetInputSizesByPosition(self):
                 end;""", [var, 'test_', 5, '_second_', 3, 7])
         self.assertEqual(var.getvalue(), u"test_5_second_37")
 
+    def testStringFormat(self):
+        """test string format of cursor"""
+        formatString = "<cx_Oracle.Cursor on <cx_Oracle.Connection to %s@%s>>"
+        expectedValue = formatString % (USERNAME, TNSENTRY)
+        self.assertEqual(str(self.cursor), expectedValue)
+
+    def testCursorFetchRaw(self):
+        """test cursor.fetchraw()"""
+        cursor = self.connection.cursor()
+        cursor.arraysize = 25
+        cursor.execute("select LongIntCol from TestNumbers order by IntCol")
+        self.assertEqual(cursor.fetchraw(), 10)
+        self.assertEqual(cursor.fetchvars[0].getvalue(), 38)
+
+    def testParse(self):
+        """test parsing statements"""
+        sql = "select LongIntCol from TestNumbers where IntCol = :val"
+        self.cursor.parse(sql)
+        self.assertEqual(self.cursor.statement, sql)
+        self.assertEqual(self.cursor.description,
+                [ ('LONGINTCOL', cx_Oracle.NUMBER, 17, None, 16, 0, 0) ])
+
diff --git a/test/NumberVar.py b/test/NumberVar.py
@@ -330,3 +330,10 @@ def testReturnFloatFromDivision(self):
         self.assertEqual(result, 1.0 / 7.0)
         self.assertTrue(isinstance(result, float), "float not returned")
 
+    def testStringFormat(self):
+        "test that string format is returned properly"
+        var = self.cursor.var(cx_Oracle.NUMBER)
+        self.assertEqual(str(var), "<cx_Oracle.NUMBER with value None>")
+        var.setvalue(0, 4)
+        self.assertEqual(str(var), "<cx_Oracle.NUMBER with value 4.0>")
+
diff --git a/test/ObjectVar.py b/test/ObjectVar.py
@@ -263,3 +263,9 @@ def testSettingVarWrongObjectType(self):
         var = self.cursor.var(cx_Oracle.OBJECT, typename = "UDT_OBJECT")
         self.assertRaises(cx_Oracle.DatabaseError, var.setvalue, 0, wrongObj)
 
+    def testStringFormat(self):
+        "test object string format"
+        objType = self.connection.gettype("UDT_OBJECT")
+        self.assertEqual(str(objType),
+                "<cx_Oracle.ObjectType CX_ORACLE.UDT_OBJECT>")
+
diff --git a/test/SessionPool.py b/test/SessionPool.py
@@ -53,6 +53,15 @@ def testPool(self):
         self.assertEqual(pool.busy, 2, "busy not 2 after release")
         del connection_2
         self.assertEqual(pool.busy, 1, "busy not 1 after del")
+        pool.getmode = cx_Oracle.SPOOL_ATTRVAL_NOWAIT
+        self.assertEqual(pool.getmode, cx_Oracle.SPOOL_ATTRVAL_NOWAIT)
+        pool.stmtcachesize = 50
+        self.assertEqual(pool.stmtcachesize, 50)
+        pool.timeout = 10
+        self.assertEqual(pool.timeout, 10)
+        if CLIENT_VERSION >= (12, 1):
+            pool.max_lifetime_session = 10
+            self.assertEqual(pool.max_lifetime_session, 10)
 
     def testProxyAuth(self):
         """test that proxy authentication is possible"""
@@ -135,3 +144,40 @@ def testThreadingWithErrors(self):
         for thread in threads:
             thread.join()
 
+    def testPurity(self):
+        """test session pool with various types of purity"""
+        action = "TEST_ACTION"
+        pool = cx_Oracle.SessionPool(USERNAME, PASSWORD, TNSENTRY, min = 1,
+                max = 8, increment = 1, encoding = ENCODING,
+                nencoding = NENCODING)
+
+        # get connection and set the action
+        connection = pool.acquire()
+        connection.action = action
+        cursor = connection.cursor()
+        cursor.execute("select 1 from dual")
+        cursor.close()
+        pool.release(connection)
+        self.assertEqual(pool.opened, 1, "opened (1)")
+
+        # verify that the connection still has the action set on it
+        connection = pool.acquire()
+        cursor = connection.cursor()
+        cursor.execute("select sys_context('userenv', 'action') from dual")
+        result, = cursor.fetchone()
+        self.assertEqual(result, action)
+        cursor.close()
+        pool.release(connection)
+        self.assertEqual(pool.opened, 1, "opened (2)")
+
+        # get a new connection with new purity (should not have state)
+        connection = pool.acquire(purity = cx_Oracle.ATTR_PURITY_NEW)
+        cursor = connection.cursor()
+        cursor.execute("select sys_context('userenv', 'action') from dual")
+        result, = cursor.fetchone()
+        self.assertEqual(result, None)
+        cursor.close()
+        self.assertEqual(pool.opened, 2, "opened (3)")
+        pool.drop(connection)
+        self.assertEqual(pool.opened, 1, "opened (4)")
+
diff --git a/test/StringVar.py b/test/StringVar.py
@@ -378,3 +378,9 @@ def testIssue50(self):
                 [2, u'd5ff845a', u'94275767', u'bf161ff6', u'', u'', idVar])
         cursor.execute("drop table issue_50 purge")
 
+    def testSetRowidToString(self):
+        "test assigning a string to rowid"
+        var = self.cursor.var(cx_Oracle.ROWID)
+        self.assertRaises(cx_Oracle.NotSupportedError, var.setvalue, 0,
+                "ABDHRYTHFJGKDKKDH")
+
diff --git a/test/Subscription.py b/test/Subscription.py
@@ -89,3 +89,8 @@ def testSubscription(self):
         self.assertEqual(data.rowOperations, rowOperations)
         self.assertEqual(data.rowids, rowids)
 
+        # test string format of subscription object is as expected
+        fmt = "<cx_Oracle.Subscription on <cx_Oracle.Connection to %s@%s>>"
+        expectedValue = fmt % (USERNAME, TNSENTRY)
+        self.assertEqual(str(sub), expectedValue)
+
diff --git a/test/test.py b/test/test.py
@@ -94,6 +94,7 @@ def tearDown(self):
     if inSetup:
         fileName = os.path.join("test", fileName)
     module = imp.new_module(name)
+    setattr(module, "CLIENT_VERSION", cx_Oracle.clientversion())
     setattr(module, "USERNAME", TestEnv.MAIN_USER)
     setattr(module, "PASSWORD", TestEnv.MAIN_PASSWORD)
     setattr(module, "PROXY_USERNAME", TestEnv.PROXY_USER)
