Codebase list python-cx-oracle / master test / Rowid.py
master

Tree @master (Download .tar.gz)

Rowid.py @master — raw · history · blame 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
# -*- coding: utf-8 -*-
#------------------------------------------------------------------------------
# Copyright (c) 2017, 2019, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------

"""Module for testing Rowids"""

import TestEnv

import cx_Oracle

class TestCase(TestEnv.BaseTestCase):

    def __TestSelectRowids(self, tableName):
        self.cursor.execute("select rowid, IntCol from %s""" % tableName)
        rowidDict = dict(self.cursor)
        sql = "select IntCol from %s where rowid = :val" % tableName
        for rowid, intVal in rowidDict.items():
            self.cursor.execute(sql, val = rowid)
            rows = self.cursor.fetchall()
            self.assertEqual(len(rows), 1)
            self.assertEqual(rows[0][0], intVal)

    def testSelectRowidsRegular(self):
        "test selecting all rowids from a regular table"
        self.__TestSelectRowids("TestNumbers")

    def testSelectRowidsIndexOrganised(self):
        "test selecting all rowids from an index organised table"
        self.__TestSelectRowids("TestUniversalRowids")

    def testInsertInvalidRowid(self):
        "test inserting an invalid rowid"
        self.assertRaises(cx_Oracle.DatabaseError, self.cursor.execute,
                "insert into TestRowids (IntCol, RowidCol) values (1, :rid)",
                rid = 12345)
        self.assertRaises(cx_Oracle.DatabaseError, self.cursor.execute,
                "insert into TestRowids (IntCol, RowidCol) values (1, :rid)",
                rid = "523lkhlf")

    def testInsertRowids(self):
        "test inserting rowids and verify they are inserted correctly"
        self.cursor.execute("select IntCol, rowid from TestNumbers")
        rows = self.cursor.fetchall()
        self.cursor.execute("truncate table TestRowids")
        self.cursor.executemany("""
                insert into TestRowids
                (IntCol, RowidCol)
                values (:1, :2)""", rows)
        self.connection.commit()
        self.cursor.execute("select IntCol, RowidCol from TestRowids")
        rows = self.cursor.fetchall()
        sql = "select IntCol from TestNumbers where rowid = :val"
        for intVal, rowid in rows:
            self.cursor.execute(sql, val = rowid)
            rows = self.cursor.fetchall()
            self.assertEqual(len(rows), 1)
            self.assertEqual(rows[0][0], intVal)

if __name__ == "__main__":
    TestEnv.RunTestCases()