iframe-proxy | Sunbelt Computer Software

History

149 lines (133 loc) · 5.63 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

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

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

"""

2800 - Module for testing AQ Bulk enqueue/dequeue

"""

import decimal

import threading

import cx_Oracle as oracledb

import test_env

RAW_QUEUE_NAME = "TEST_RAW_QUEUE"

RAW_PAYLOAD_DATA = [

"The first message",

"The second message",

"The third message",

"The fourth message",

"The fifth message",

"The sixth message",

"The seventh message",

"The eighth message",

"The ninth message",

"The tenth message",

"The eleventh message",

"The twelfth and final message"

]

class TestCase(test_env.BaseTestCase):

def __deq_in_thread(self, results):

connection = test_env.get_connection(threaded=True)

queue = connection.queue(RAW_QUEUE_NAME)

queue.deqoptions.wait = 10

queue.deqoptions.navigation = oracledb.DEQ_FIRST_MSG

while len(results) < len(RAW_PAYLOAD_DATA):

messages = queue.deqmany(5)

if not messages:

break

for m in messages:

results.append(m.payload.decode(connection.encoding))

connection.commit()

def __get_and_clear_raw_queue(self):

queue = self.connection.queue(RAW_QUEUE_NAME)

queue.deqoptions.wait = oracledb.DEQ_NO_WAIT

queue.deqoptions.navigation = oracledb.DEQ_FIRST_MSG

while queue.deqone():

pass

self.connection.commit()

return queue

def test_2800_enq_and_deq(self):

"2800 - test bulk enqueue and dequeue"

queue = self.__get_and_clear_raw_queue()

messages = [self.connection.msgproperties(payload=d) \

for d in RAW_PAYLOAD_DATA]

queue.enqmany(messages)

messages = queue.deqmany(len(RAW_PAYLOAD_DATA))

data = [m.payload.decode(self.connection.encoding) for m in messages]

self.connection.commit()

self.assertEqual(data, RAW_PAYLOAD_DATA)

def test_2801_dequeue_empty(self):

"2801 - test empty bulk dequeue"

queue = self.__get_and_clear_raw_queue()

messages = queue.deqmany(5)

self.connection.commit()

self.assertEqual(messages, [])

def test_2802_deq_with_wait(self):

"2802 - test bulk dequeue with wait"

queue = self.__get_and_clear_raw_queue()

results = []

thread = threading.Thread(target=self.__deq_in_thread, args=(results,))

thread.start()

messages = [self.connection.msgproperties(payload=d) \

for d in RAW_PAYLOAD_DATA]

queue.enqoptions.visibility = oracledb.ENQ_IMMEDIATE

queue.enqmany(messages)

thread.join()

self.assertEqual(results, RAW_PAYLOAD_DATA)

def test_2803_enq_and_deq_multiple_times(self):

"2803 - test enqueue and dequeue multiple times"

queue = self.__get_and_clear_raw_queue()

data_to_enqueue = RAW_PAYLOAD_DATA

for num in (2, 6, 4):

messages = [self.connection.msgproperties(payload=d) \

for d in data_to_enqueue[:num]]

data_to_enqueue = data_to_enqueue[num:]

queue.enqmany(messages)

self.connection.commit()

all_data = []

for num in (3, 5, 10):

messages = queue.deqmany(num)

all_data.extend(m.payload.decode(self.connection.encoding) \

for m in messages)

self.connection.commit()

self.assertEqual(all_data, RAW_PAYLOAD_DATA)

def test_2804_enq_and_deq_visibility(self):

"2804 - test visibility option for enqueue and dequeue"

queue = self.__get_and_clear_raw_queue()

# first test with ENQ_ON_COMMIT (commit required)

queue.enqoptions.visibility = oracledb.ENQ_ON_COMMIT

props1 = self.connection.msgproperties(payload="A first message")

props2 = self.connection.msgproperties(payload="A second message")

queue.enqmany([props1, props2])

other_connection = test_env.get_connection()

other_queue = other_connection.queue(RAW_QUEUE_NAME)

other_queue.deqoptions.wait = oracledb.DEQ_NO_WAIT

other_queue.deqoptions.visibility = oracledb.DEQ_ON_COMMIT

messages = other_queue.deqmany(5)

self.assertEqual(len(messages), 0)

self.connection.commit()

messages = other_queue.deqmany(5)

self.assertEqual(len(messages), 2)

other_connection.rollback()

# second test with ENQ_IMMEDIATE (no commit required)

queue.enqoptions.visibility = oracledb.ENQ_IMMEDIATE

other_queue.deqoptions.visibility = oracledb.DEQ_IMMEDIATE

queue.enqmany([props1, props2])

messages = other_queue.deqmany(5)

self.assertEqual(len(messages), 4)

other_connection.rollback()

messages = other_queue.deqmany(5)

self.assertEqual(len(messages), 0)

def test_2806_verify_msgid(self):

"2806 - verify that the msgid property is returned correctly"

queue = self.__get_and_clear_raw_queue()

messages = [self.connection.msgproperties(payload=d) \

for d in RAW_PAYLOAD_DATA]

queue.enqmany(messages)

self.cursor.execute("select msgid from raw_queue_tab")

actual_msgids = set(m for m, in self.cursor)

msgids = set(m.msgid for m in messages)

self.assertEqual(msgids, actual_msgids)

messages = queue.deqmany(len(RAW_PAYLOAD_DATA))

msgids = set(m.msgid for m in messages)

self.assertEqual(msgids, actual_msgids)

if __name__ == "__main__":

test_env.run_test_cases()

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

test_2800_bulk_aq.py

Latest commit

History

test_2800_bulk_aq.py

File metadata and controls