iframe-proxy | Sunbelt Computer Software

History

114 lines (91 loc) · 4.18 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

#------------------------------------------------------------------------------

"""Module for testing subscriptions."""

import TestEnv

import cx_Oracle

import threading

class SubscriptionData(object):

def __init__(self, numMessagesExpected):

self.condition = threading.Condition()

self.numMessagesExpected = numMessagesExpected

self.numMessagesReceived = 0

self.tableOperations = []

self.rowOperations = []

self.rowids = []

def CallbackHandler(self, message):

if message.type != cx_Oracle.EVENT_DEREG:

table, = message.tables

self.tableOperations.append(table.operation)

for row in table.rows:

self.rowOperations.append(row.operation)

self.rowids.append(row.rowid)

self.numMessagesReceived += 1

if message.type == cx_Oracle.EVENT_DEREG or \

self.numMessagesReceived == self.numMessagesExpected:

self.condition.acquire()

self.condition.notify()

self.condition.release()

class TestCase(TestEnv.BaseTestCase):

def testSubscription(self):

"test Subscription for insert, update, delete and truncate"

# skip if running on the Oracle Cloud, which does not support

# subscriptions currently

if self.isOnOracleCloud():

self.skipTest("Oracle Cloud does not support subscriptions " \

"currently")

# truncate table in order to run test in known state

self.cursor.execute("truncate table TestTempTable")

# expected values

tableOperations = [ cx_Oracle.OPCODE_INSERT, cx_Oracle.OPCODE_UPDATE,

cx_Oracle.OPCODE_INSERT, cx_Oracle.OPCODE_DELETE,

cx_Oracle.OPCODE_ALTER | cx_Oracle.OPCODE_ALLROWS ]

rowOperations = [ cx_Oracle.OPCODE_INSERT, cx_Oracle.OPCODE_UPDATE,

cx_Oracle.OPCODE_INSERT, cx_Oracle.OPCODE_DELETE ]

rowids = []

# set up subscription

data = SubscriptionData(5)

connection = TestEnv.GetConnection(threaded=True, events=True)

sub = connection.subscribe(callback = data.CallbackHandler,

timeout = 10, qos = cx_Oracle.SUBSCR_QOS_ROWIDS)

sub.registerquery("select * from TestTempTable")

connection.autocommit = True

cursor = connection.cursor()

# insert statement

cursor.execute("""

insert into TestTempTable (IntCol, StringCol)

values (1, 'test')""")

cursor.execute("select rowid from TestTempTable where IntCol = 1")

rowids.extend(r for r, in cursor)

# update statement

cursor.execute("""

update TestTempTable set

StringCol = 'update'

where IntCol = 1""")

cursor.execute("select rowid from TestTempTable where IntCol = 1")

rowids.extend(r for r, in cursor)

# second insert statement

cursor.execute("""

insert into TestTempTable (IntCol, StringCol)

values (2, 'test2')""")

cursor.execute("select rowid from TestTempTable where IntCol = 2")

rowids.extend(r for r, in cursor)

# delete statement

cursor.execute("delete TestTempTable where IntCol = 2")

rowids.append(rowids[-1])

# truncate table

cursor.execute("truncate table TestTempTable")

# wait for all messages to be sent

data.condition.acquire()

data.condition.wait(10)

# verify the correct messages were sent

self.assertEqual(data.tableOperations, tableOperations)

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 % \

(TestEnv.GetMainUser(), TestEnv.GetConnectString())

self.assertEqual(str(sub), expectedValue)

if __name__ == "__main__":

TestEnv.RunTestCases()

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

Subscription.py

Latest commit

History

Subscription.py

File metadata and controls