dnrops
/
python-oracledb
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Test improvements.

This commit is contained in:
Anthony Tuininga 2023-08-14 18:53:26 -06:00
parent ae37b6e0ee
commit 575ca43a13
2 changed files with 25 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
tests/test_2700_aq.py
View File

@ -426,6 +426,7 @@ class TestCase(test_env.BaseTestCase):
queue = self.connection.queue(self.book_queue_name,
payloadType=books_type)
self.assertEqual(queue.payload_type, books_type)
self.assertEqual(queue.payloadType, books_type)
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2014:",
self.connection.queue, self.book_queue_name,
books_type, payloadType=books_type)
@ -475,8 +476,8 @@ class TestCase(test_env.BaseTestCase):
props1 = queue.deqone()
self.assertTrue(props1 is None)
def test_2720_aq_notification(self):
"2720 - verify msgid of aq message which spawned notification "
def test_2720_aq_message_attributes(self):
"2720 - verify attributes of AQ message which spawned notification"
if self.is_on_oracle_cloud(self.connection):
self.skipTest("AQ notification not supported on the cloud")
queue = self.get_and_clear_queue(self.book_queue_name,
@ -486,7 +487,12 @@ class TestCase(test_env.BaseTestCase):
def notification_callback(message):
self.cursor.execute("select msgid from book_queue_tab")
actual_msgid, = self.cursor.fetchone()
self.assertEqual(actual_msgid, message.msgid)
self.assertEqual(message.msgid, actual_msgid)
self.assertEqual(message.consumer_name, None)
main_user = test_env.get_main_user().upper()
self.assertEqual(message.queue_name,
f'"{main_user}"."{queue.name}"')
self.assertEqual(message.type, oracledb.EVENT_AQ)
with condition:
condition.notify()
sub = connection.subscribe(namespace=oracledb.SUBSCR_NAMESPACE_AQ,

17
tests/test_2800_bulk_aq.py
View File

@ -1,5 +1,5 @@
#------------------------------------------------------------------------------
# Copyright (c) 2020, 2022, Oracle and/or its affiliates.
# Copyright (c) 2020, 2023, 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
@ -191,5 +191,20 @@ class TestCase(test_env.BaseTestCase):
actual_data = [m.payload for m in messages]
self.assertEqual(actual_data, JSON_DATA_PAYLOAD)
def test_2808_no_json_payload(self):
"2808 - test enqueuing to a JSON queue without a JSON payload"
queue = self.get_and_clear_queue(JSON_QUEUE_NAME, "JSON")
props = self.connection.msgproperties(payload="string message")
self.assertRaisesRegex(oracledb.DatabaseError, "^DPI-1071:",
queue.enqmany, [props, props])
def test_2809_errors_for_invalid_values(self):
"2809 - test errors for invalid values for enqmany and deqmany"
queue = self.get_and_clear_queue(JSON_QUEUE_NAME, "JSON")
props = self.connection.msgproperties(payload="string message")
self.assertRaises(TypeError, queue.enqmany, props)
self.assertRaises(TypeError, queue.enqmany, ["Not", "msgproperties"])
self.assertRaises(TypeError, queue.deqmany, "5")
if __name__ == "__main__":
test_env.run_test_cases()