--- /dev/null
+# ==================================================================================
+# Copyright (c) 2019 Nokia
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# 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
+#
+# http://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.
+# ==================================================================================
+"""
+Custom Exceptions
+"""
+
+
+class BadBufferAllocation(BaseException):
+ """a bad buffer was allocated, check the rmr context"""
# ==================================================================================
import time
import pytest
-from rmr import rmr, helpers
+from rmr import rmr, helpers, exceptions
SIZE = 256
while rmr.rmr_ready(MRC_RCV) == 0:
time.sleep(1)
+
def teardown_module():
"""
test rmr module teardown
errors = 0
econst = expected_constants
rconst = rmr._get_constants()
- for key in econst: # test all expected constants
- if key not in rconst: # expected value not listed by rmr
+ for key in econst: # test all expected constants
+ if key not in rconst: # expected value not listed by rmr
errors += 1
- print( "did not find required constant in list from RMR: %s" % key )
+ print("did not find required constant in list from RMR: %s" % key)
assert errors == 0
sbuf_send.contents.mtype = 0
sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
send_summary = rmr.message_summary(sbuf_send)
- assert send_summary["message state"] == 0 # if send fails don't attempt receive
+ assert send_summary["message state"] == 0 # if send fails don't attempt receive
assert send_summary["message status"] == "RMR_OK"
- time.sleep(.5)
+ time.sleep(0.5)
# receive it in other context
sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, SIZE)
Send a burst of messages optionally giving the type, payload
and number to send.
"""
- sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE) # seed message buffer
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE) # seed message buffer
for i in range(num):
payload = bytes(fmt % counter, "UTF-8")
while max_retries > 0:
sbuf_send = rmr.rmr_send_msg(mrc, sbuf_send)
ms = rmr.message_summary(sbuf_send)
- if ms["message state"] != 10: # 10 is retry
+ if ms["message state"] != 10: # 10 is retry
break
max_retries -= 1
- time.sleep(.75)
+ time.sleep(0.75)
assert ms["message state"] == 0
assert max_retries > 0
def test_rcv_all():
"""
- test the ability to receive a batch of queued messages.
+ test the ability to receive a batch of queued messages.
"""
- pay_fmt = "send to ring msg: %d" # dynamic message format with counter
+ pay_fmt = "send to ring msg: %d" # dynamic message format with counter
- send_burst(MRC_SEND, pay_fmt) # send a bunch of 13 messages that should queue
- time.sleep(1) # ensure underlying transport gets cycles to send/receive
+ send_burst(MRC_SEND, pay_fmt) # send a bunch of 13 messages that should queue
+ time.sleep(1) # ensure underlying transport gets cycles to send/receive
- bundle = helpers.rmr_rcvall_msgs(MRC_BUF_RCV) # use the buffered receiver to read all with a single call
+ bundle = helpers.rmr_rcvall_msgs(MRC_BUF_RCV) # use the buffered receiver to read all with a single call
assert len(bundle) == 13
for i in range(len(bundle)):
- ms = bundle[i] # validate each summary returned, and ordering preserved
+ ms = bundle[i] # validate each summary returned, and ordering preserved
assert ms["message state"] == 0
expected_pay = bytes(pay_fmt % i, "UTF-8")
assert ms["payload"] == expected_pay
- send_burst(MRC_SEND, pay_fmt, mtype=1, num=10) # send a second round with msg types 1 and 2 to test filter
+ send_burst(MRC_SEND, pay_fmt, mtype=1, num=10) # send a second round with msg types 1 and 2 to test filter
send_burst(MRC_SEND, pay_fmt, mtype=2, num=8)
send_burst(MRC_SEND, pay_fmt, mtype=1, num=5)
- send_burst(MRC_SEND, pay_fmt, mtype=2, num=4, counter=8) # total of 12 messages with type 2 should be queued
- time.sleep(1) # ensure underlying transport gets cycles to send/receive
+ send_burst(MRC_SEND, pay_fmt, mtype=2, num=4, counter=8) # total of 12 messages with type 2 should be queued
+ time.sleep(1) # ensure underlying transport gets cycles to send/receive
- bundle = helpers.rmr_rcvall_msgs(MRC_BUF_RCV, [2]) # receive only message type 2 messages
- assert len(bundle) == 12 # we should only get the second batch of 12 messages
+ bundle = helpers.rmr_rcvall_msgs(MRC_BUF_RCV, [2]) # receive only message type 2 messages
+ assert len(bundle) == 12 # we should only get the second batch of 12 messages
for i in range(len(bundle)):
- ms = bundle[i] # validate each summary
- assert ms["message state"] == 0 # all should be OK
- assert ms["message type"] == 2 # only mtype 2 should have been received
- expected_pay = bytes(pay_fmt % i, "UTF-8") # ordering should still jive with the counter
+ ms = bundle[i] # validate each summary
+ assert ms["message state"] == 0 # all should be OK
+ assert ms["message type"] == 2 # only mtype 2 should have been received
+ expected_pay = bytes(pay_fmt % i, "UTF-8") # ordering should still jive with the counter
assert ms["payload"] == expected_pay
+
+
+def test_bad_buffer():
+ """test that we get a proper exception when the buffer has a null pointer"""
+ with pytest.raises(exceptions.BadBufferAllocation):
+ rmr.rmr_alloc_msg(None, 4096)
Code Review / ric-plt / lib / rmr.git / commitdiff