iframe-proxy | Sunbelt Computer Software

History

123 lines (101 loc) · 4.32 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

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

"""

3000 - Module for testing subscriptions

"""

import test_env

import cx_Oracle as oracledb

import threading

class SubscriptionData(object):

def __init__(self, num_messages_expected):

self.condition = threading.Condition()

self.num_messages_expected = num_messages_expected

self.num_messages_received = 0

self.table_operations = []

self.row_operations = []

self.rowids = []

def CallbackHandler(self, message):

if message.type != oracledb.EVENT_DEREG:

table, = message.tables

self.table_operations.append(table.operation)

for row in table.rows:

self.row_operations.append(row.operation)

self.rowids.append(row.rowid)

self.num_messages_received += 1

if message.type == oracledb.EVENT_DEREG or \

self.num_messages_received == self.num_messages_expected:

self.condition.acquire()

self.condition.notify()

self.condition.release()

class TestCase(test_env.BaseTestCase):

def test_3000_subscription(self):

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

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

# subscriptions currently

if self.is_on_oracle_cloud():

message = "Oracle Cloud does not support subscriptions currently"

self.skipTest(message)

# truncate table in order to run test in known state

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

# expected values

table_operations = [

oracledb.OPCODE_INSERT,

oracledb.OPCODE_UPDATE,

oracledb.OPCODE_INSERT,

oracledb.OPCODE_DELETE,

oracledb.OPCODE_ALTER | oracledb.OPCODE_ALLROWS

]

row_operations = [

oracledb.OPCODE_INSERT,

oracledb.OPCODE_UPDATE,

oracledb.OPCODE_INSERT,

oracledb.OPCODE_DELETE

]

rowids = []

# set up subscription

data = SubscriptionData(5)

connection = test_env.get_connection(threaded=True, events=True)

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

timeout=10, qos=oracledb.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.table_operations, table_operations)

self.assertEqual(data.row_operations, row_operations)

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>>"

expected = fmt % \

(test_env.get_main_user(), test_env.get_connect_string())

self.assertEqual(str(sub), expected)

if __name__ == "__main__":

test_env.run_test_cases()

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

test_3000_subscription.py

Latest commit

History

test_3000_subscription.py

File metadata and controls