test_7800_aq_raw.py

# -----------------------------------------------------------------------------
# Copyright (c) 2025, Oracle and/or its affiliates.
#
# This software is dual-licensed to you under the Universal Permissive License
# (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
# 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
# either license.
#
# If you elect to accept the software under the Apache License, Version 2.0,
# the following applies:
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------

"""
7800 - Module for testing AQ with raw queues
"""

importoracledb
importtest_env
importthreading


classTestCase(test_env.BaseTestCase):
raw_data= [
b"sample raw data 1",
b"sample raw data 2",
b"sample raw data 3",
b"sample raw data 4",
b"sample raw data 5",
b"sample raw data 6",
 ]

def__deq_in_thread(self, results):
withtest_env.get_connection() asconn:
queue=conn.queue("TEST_RAW_QUEUE")
queue.deqoptions.wait=10
props=queue.deqone()
ifpropsisnotNone:
results.append(props.payload)
conn.commit()

def__verify_attr(self, obj, attrName, value):
setattr(obj, attrName, value)
self.assertEqual(getattr(obj, attrName), value)

deftest_7800(self):
"7800 - test dequeuing an empty RAW queue"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
props=queue.deqone()
self.assertIsNone(props)

deftest_7801(self):
"7801 - test enqueuing and dequeuing multiple RAW messages"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
props=self.conn.msgproperties()
forvalueinself.raw_data:
props.payload=value
queue.enqone(props)
self.conn.commit()
queue.deqoptions.navigation=oracledb.DEQ_FIRST_MSG
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
results= []
whileTrue:
props=queue.deqone()
ifpropsisNone:
break
value=props.payload
results.append(value)
self.conn.commit()
self.assertEqual(results, self.raw_data)

deftest_7802(self):
"7802 - test dequeuing with DEQ_REMOVE_NODATA in RAW queue"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[1]
props=self.conn.msgproperties(payload=value)
queue.enqone(props)
queue.deqoptions.navigation=oracledb.DEQ_FIRST_MSG
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
queue.deqoptions.mode=oracledb.DEQ_REMOVE_NODATA
props=queue.deqone()
self.assertIsNotNone(props)
self.assertEqual(props.payload, b"")

deftest_7803(self):
"7803 - test getting/setting dequeue options attributes"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
options=queue.deqoptions
self.__verify_attr(options, "condition", "TEST_CONDITION")
self.__verify_attr(options, "consumername", "TEST_CONSUMERNAME")
self.__verify_attr(options, "correlation", "TEST_CORRELATION")
self.__verify_attr(options, "mode", oracledb.DEQ_LOCKED)
self.__verify_attr(
options, "navigation", oracledb.DEQ_NEXT_TRANSACTION
 )
self.__verify_attr(options, "transformation", "TEST_TRANSFORMATION")
self.__verify_attr(options, "visibility", oracledb.ENQ_IMMEDIATE)
self.__verify_attr(options, "wait", 1287)
self.__verify_attr(options, "msgid", b"mID")

deftest_7804(self):
"7804 - test enqueue options attributes RAW queue"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
options=queue.enqoptions
self.__verify_attr(options, "visibility", oracledb.ENQ_IMMEDIATE)

deftest_7805(self):
"7805 - test waiting for dequeue"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
results= []
thread=threading.Thread(target=self.__deq_in_thread, args=(results,))
thread.start()
value=self.raw_data[0]
props=self.conn.msgproperties(payload=value)
queue.enqone(props)
self.conn.commit()
thread.join()
self.assertEqual(results, [value])

deftest_7806(self):
"7806 - test getting/setting message properties attributes"
props=self.conn.msgproperties()
self.__verify_attr(props, "correlation", "TEST_CORRELATION")
self.__verify_attr(props, "delay", 60)
self.__verify_attr(props, "exceptionq", "TEST_EXCEPTIONQ")
self.__verify_attr(props, "expiration", 30)
self.assertEqual(props.attempts, 0)
self.__verify_attr(props, "priority", 1)
self.assertEqual(props.state, oracledb.MSG_READY)
self.assertEqual(props.deliverymode, 0)

deftest_7807(self):
"7807 - test enqueue visibility option - ENQ_ON_COMMIT"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
queue.enqoptions.visibility=oracledb.ENQ_ON_COMMIT
props=self.conn.msgproperties(payload=value)
queue.enqone(props)

other_conn=test_env.get_connection()
queue=other_conn.queue("TEST_RAW_QUEUE")
queue.deqoptions.navigation=oracledb.DEQ_FIRST_MSG
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
props=queue.deqone()
self.assertIsNone(props)
self.conn.commit()
props=queue.deqone()
self.assertIsNotNone(props)

deftest_7808(self):
"7808 - test enqueue visibility option - ENQ_IMMEDIATE"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
queue.enqoptions.visibility=oracledb.ENQ_IMMEDIATE
props=self.conn.msgproperties(payload=value)
queue.enqone(props)

other_conn=test_env.get_connection()
queue=other_conn.queue("TEST_RAW_QUEUE")
queue.deqoptions.navigation=oracledb.DEQ_FIRST_MSG
queue.deqoptions.visibility=oracledb.DEQ_ON_COMMIT
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
props=queue.deqone()
value=props.payload
results=value
other_conn.commit()
self.assertEqual(results, self.raw_data[0])

deftest_7809(self):
"7809 - test enqueue/dequeue delivery modes identical - buffered"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
queue.enqoptions.deliverymode=oracledb.MSG_BUFFERED
queue.enqoptions.visibility=oracledb.ENQ_IMMEDIATE
props=self.conn.msgproperties(payload=value)
queue.enqone(props)

other_conn=test_env.get_connection()
queue=other_conn.queue("TEST_RAW_QUEUE")
queue.deqoptions.deliverymode=oracledb.MSG_BUFFERED
queue.deqoptions.navigation=oracledb.DEQ_FIRST_MSG
queue.deqoptions.visibility=oracledb.DEQ_IMMEDIATE
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
props=queue.deqone()
value=props.payload
results=value
other_conn.commit()
self.assertEqual(results, self.raw_data[0])

deftest_7810(self):
"7810 - test enqueue/dequeue delivery modes identical - persistent"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
queue.enqoptions.deliverymode=oracledb.MSG_PERSISTENT
queue.enqoptions.visibility=oracledb.ENQ_IMMEDIATE
props=self.conn.msgproperties(payload=value)
queue.enqone(props)

other_conn=test_env.get_connection()
queue=other_conn.queue("TEST_RAW_QUEUE")
queue.deqoptions.deliverymode=oracledb.MSG_PERSISTENT
queue.deqoptions.navigation=oracledb.DEQ_FIRST_MSG
queue.deqoptions.visibility=oracledb.DEQ_IMMEDIATE
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
props=queue.deqone()
value=props.payload
results=value
other_conn.commit()
self.assertEqual(results, self.raw_data[0])

deftest_7811(self):
"7811 - test enqueue/dequeue delivery modes the same"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
queue.enqoptions.deliverymode=oracledb.MSG_PERSISTENT_OR_BUFFERED
queue.enqoptions.visibility=oracledb.ENQ_IMMEDIATE
props=self.conn.msgproperties(payload=value)
queue.enqone(props)

other_conn=test_env.get_connection()
queue=other_conn.queue("TEST_RAW_QUEUE")
queue.deqoptions.deliverymode=oracledb.MSG_PERSISTENT_OR_BUFFERED
queue.deqoptions.navigation=oracledb.DEQ_FIRST_MSG
queue.deqoptions.visibility=oracledb.DEQ_IMMEDIATE
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
props=queue.deqone()
value=props.payload
results=value
other_conn.commit()
self.assertEqual(results, self.raw_data[0])

deftest_7812(self):
"7812 - test enqueue/dequeue delivery modes different"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
queue.enqoptions.deliverymode=oracledb.MSG_BUFFERED
queue.enqoptions.visibility=oracledb.ENQ_IMMEDIATE
props=self.conn.msgproperties(payload=value)
queue.enqone(props)

other_conn=test_env.get_connection()
queue=other_conn.queue("TEST_RAW_QUEUE")
queue.deqoptions.deliverymode=oracledb.MSG_PERSISTENT
queue.deqoptions.navigation=oracledb.DEQ_FIRST_MSG
queue.deqoptions.visibility=oracledb.DEQ_IMMEDIATE
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
props=queue.deqone()
self.assertIsNone(props)

deftest_7813(self):
"7813 - test error for message with no payload"
queue=self.conn.queue("TEST_RAW_QUEUE")
props=self.conn.msgproperties()
withself.assertRaisesFullCode("DPY-2000"):
queue.enqone(props)

deftest_7814(self):
"7814 - verify that the msgid property is returned correctly"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
props=self.conn.msgproperties(payload=value)
self.assertIsNone(props.msgid)
queue.enqone(props)
self.cursor.execute("select msgid from RAW_QUEUE_TAB")
 (actual_msgid,) =self.cursor.fetchone()
self.assertEqual(props.msgid, actual_msgid)
props=queue.deqone()
self.assertEqual(props.msgid, actual_msgid)

deftest_7815(self):
"7815 - test message props enqtime"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
self.cursor.execute("select current_timestamp from dual")
 (start_date,) =self.cursor.fetchone()
start_date=start_date.replace(microsecond=0)
props=self.conn.msgproperties(payload=value)
queue.enqone(props)
props=queue.deqone()
self.cursor.execute("select current_timestamp from dual")
 (end_date,) =self.cursor.fetchone()
end_date=end_date.replace(microsecond=0)
self.assertTrue(start_date<=props.enqtime<=end_date)

deftest_7816(self):
"7816 - test message props declared attributes"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
values=dict(
payload=value,
correlation="TEST_CORRELATION",
delay=0,
exceptionq="PYTHONTEST.TEST_EXCEPTIONQ",
expiration=15,
priority=1,
 )
props=self.conn.msgproperties(**values)
forattr_nameinvalues:
self.assertEqual(getattr(props, attr_name), values[attr_name])
queue.enqone(props)
self.conn.commit()
prop=queue.deqone()
forattr_nameinvalues:
self.assertEqual(getattr(prop, attr_name), values[attr_name])

deftest_7817(self):
"7817 - test getting queue attributes"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
self.assertEqual(queue.name, "TEST_RAW_QUEUE")
self.assertEqual(queue.connection, self.conn)

deftest_7818(self):
"7818 - test getting write-only attributes"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
foroptionsin (queue.enqoptions, queue.deqoptions):
withself.assertRaises(AttributeError):
options.deliverymode

deftest_7819(self):
"7819 - test deqoption condition with priority"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
priorities= [5, 5, 5, 5, 10, 9, 9, 10, 9]
forpriorityinpriorities:
value=self.raw_data[0]
props=self.conn.msgproperties(payload=value, priority=priority)
queue.enqone(props)

queue.deqoptions.condition="priority = 9"
results= []
whileTrue:
props=queue.deqone()
ifpropsisNone:
break
results.append(props.payload)
self.conn.commit()
self.assertEqual(len(results), 3)

deftest_7820(self):
"7820 - test deqoption correlation"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
correlations= [
"sample",
"sample correlation",
"sample",
"sample",
"sample correlation",
 ]
forcorrelationincorrelations:
value=self.raw_data[0]
props=self.conn.msgproperties(
payload=value, correlation=correlation
 )
queue.enqone(props)
self.conn.commit()
queue.deqoptions.correlation="sample correlation"
results= []
whileTrue:
props=queue.deqone()
ifpropsisNone:
break
results.append(props.payload)
self.conn.commit()
self.assertEqual(len(results), 2)

deftest_7821(self):
"7821 - test deqoption msgid"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
value=self.raw_data[0]
props=self.conn.msgproperties(payload=value)
queue.enqone(props)
queue.enqone(props)
self.conn.commit()
msgid=props.msgid
queue.enqone(props)
self.conn.commit()
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
queue.deqoptions.msgid=msgid
prop=queue.deqone()
self.conn.commit()
self.assertEqual(prop.msgid, msgid)

deftest_7822(self):
"7822 - test payload_type returns the correct value"
queue=self.conn.queue("TEST_RAW_QUEUE")
self.assertIsNone(queue.payload_type)

deftest_7823(self):
"7823 - test deprecated attributes (enqOptions, deqOptions)"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
self.assertEqual(queue.enqOptions, queue.enqoptions)
self.assertEqual(queue.deqOptions, queue.deqoptions)

deftest_7824(self):
"7824 - test deprecated AQ methods (enqOne, deqOne)"
value=b"Test 7823"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
queue.enqOne(self.conn.msgproperties(value))
props=queue.deqOne()
self.assertEqual(props.payload, value)

deftest_7825(self):
"7825 - test wrong payload type"
queue=self.get_and_clear_queue("TEST_RAW_QUEUE")
typ=self.conn.gettype("UDT_BOOK")
obj=typ.newobject()
props=self.conn.msgproperties(payload=obj)
withself.assertRaisesFullCode("DPY-2062"):
queue.enqone(props)


if__name__=="__main__":
test_env.run_test_cases()