# ==================================================================================
import time
import pytest
-from rmr import rmr, helpers
+from rmr import rmr, helpers, exceptions
SIZE = 256
global MRC_BUF_RCV
MRC_BUF_RCV = rmr.rmr_init(b"4564", rmr.RMR_MAX_RCV_BYTES, 0x02)
- while rmr.rmr_ready(MRC_RCV) == 0:
+ while rmr.rmr_ready(MRC_BUF_RCV) == 0:
time.sleep(1)
+
def teardown_module():
"""
test rmr module teardown
summary = rmr.message_summary(sbuf)
assert summary["payload"] == b""
assert summary["payload length"] == 0
+ assert summary["subscription id"] == -1
assert summary["transaction id"] == b""
assert summary["message state"] == 0
assert summary["message status"] == "RMR_OK"
- assert summary["meid"] == ""
+ assert summary["meid"] == b""
assert summary["errno"] == 0
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 = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
- rmr.rmr_set_meid(sbuf, b"\x01\x02", 2)
- assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == "\x01\x02"
+ rmr.rmr_set_meid(sbuf, b"\x01\x02")
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"\x01\x02"
assert len(rmr.rmr_get_meid(sbuf)) == 2
- rmr.rmr_set_meid(sbuf, b"\x00" * 32, 32)
- assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == "" # NULL bytes get truncated
+ rmr.rmr_set_meid(sbuf, b"\x00" * 32)
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"" # NULL bytes get truncated
- rmr.rmr_set_meid(sbuf, b"6" * 32, 32)
- assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == "6" * 32 # string in string out
+ rmr.rmr_set_meid(sbuf, b"6" * 32)
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"6" * 32 # string in string out
- rmr.rmr_set_meid(sbuf, b"\x01\x02", 2)
+ rmr.rmr_set_meid(sbuf, b"\x01\x02")
assert (
- rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == "\x01\x02" + "6" * 30
+ rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"\x01\x02" + b"6" * 30
) # bytes in string out, 6s left over
assert len(rmr.rmr_get_meid(sbuf)) == 32
assert len(summary["transaction id"]) == 32
# test meid
- assert rmr.rmr_get_meid(sbuf) == summary["meid"] == ""
- rmr.rmr_set_meid(sbuf, b"666\x01\x00\x01", 6)
+ assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b""
+ rmr.rmr_set_meid(sbuf, b"666\x01\x00\x01")
summary = rmr.message_summary(sbuf)
- assert rmr.rmr_get_meid(sbuf) == summary["meid"] == "666\x01"
+ assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b"666\x01"
assert (len(summary["meid"])) == 4
+def test_alloc_fancy():
+ """test allocation with setting payload, trans, mtype, subid"""
+ pay = b"yoo\x01\x00\x80"
+ sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE, payload=pay, gen_transaction_id=True, mtype=14, meid=b"asdf", sub_id=654321)
+ summary = rmr.message_summary(sbuf)
+ assert summary["payload"] == pay
+ assert summary["payload length"] == 6
+ assert summary["transaction id"] != b"" # hard to test what it will be, but make sure not empty
+ assert len(summary["transaction id"]) == 32
+ assert summary["message state"] == 0
+ assert summary["message type"] == sbuf.contents.mtype == 14
+ assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b"asdf"
+ assert sbuf.contents.sub_id == summary["subscription id"] == 654321
+
+
+def test_alloc_overlapping_flags():
+ """test allocation with setting the transaction id"""
+ sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE, gen_transaction_id=True, fixed_transaction_id=b"6" * 32)
+ summary = rmr.message_summary(sbuf)
+ assert summary["transaction id"] == b"66666666666666666666666666666666"
+
+
def test_rcv_timeout():
"""
test torcv; this is a scary test because if it fails... it doesn't fail, it will run forever!
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 an ACK back
ack_pay = b"message received"
- rmr.set_payload_and_length(ack_pay, sbuf_rcv)
- sbuf_rcv = rmr.rmr_rts_msg(MRC_RCV, sbuf_rcv)
+ sbuf_rcv = rmr.rmr_rts_msg(MRC_RCV, sbuf_rcv, payload=ack_pay, mtype=6666)
rcv_ack_summary = rmr.message_summary(sbuf_rcv)
# have the sender receive it
assert send_ack_summary["message state"] == rcv_ack_summary["message state"] == 0
assert send_ack_summary["message status"] == rcv_ack_summary["message status"] == "RMR_OK"
assert send_ack_summary["payload"] == ack_pay
+ assert send_ack_summary["message type"] == 6666
+
+
+def test_send_rcv_subid_good():
+ """
+ test send and receive where subid is used for routing
+ """
+ pay = b"\x01\x00\x80"
+ test_mtype = 46656
+ test_subid = 777
+
+ # send a message
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, pay, mtype=test_mtype, sub_id=test_subid)
+ pre_send_summary = rmr.message_summary(sbuf_send)
+ sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
+ send_summary = rmr.message_summary(sbuf_send)
+
+ # receive it in other context
+ time.sleep(0.5)
+ sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, 3)
+ sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
+ rcv_summary = rmr.message_summary(sbuf_rcv)
+
+ # asserts
+ assert send_summary["message state"] == rcv_summary["message state"] == 0
+ assert send_summary["message status"] == rcv_summary["message status"] == "RMR_OK"
+ assert pre_send_summary["payload"] == rcv_summary["payload"] == pay
+ assert pre_send_summary["message type"] == rcv_summary["message type"] == test_mtype
+ assert pre_send_summary["subscription id"] == rcv_summary["subscription id"] == test_subid
+
+
+def test_send_rcv_subid_bad_subid():
+ """
+ test send and receive where subid is used for routing but nobody recieves this subid
+ """
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, b"\x01\x00\x80", mtype=46656, sub_id=778)
+ sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
+ assert rmr.message_summary(sbuf_send)["message state"] == 2
+ assert rmr.message_summary(sbuf_send)["message status"] == "RMR_ERR_NOENDPT"
+
+
+def test_send_rcv_subid_bad_mtype():
+ """
+ test send and receive where subid is used for routing but nobody recieves this mtype
+ """
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, b"\x01\x00\x80", mtype=46657, sub_id=777)
+ sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
+ assert rmr.message_summary(sbuf_send)["message state"] == 2
+ assert rmr.message_summary(sbuf_send)["message status"] == "RMR_ERR_NOENDPT"
def send_burst(mrc, fmt, mtype=1, num=13, counter=0):
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
+ for i, ms in enumerate(bundle):
+ 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_raw(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, (ms, sbuf) in enumerate(bundle): # test the raw version
+ test_summary = rmr.message_summary(sbuf)
+ assert test_summary == ms
+ 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
+ rmr.rmr_free_msg(sbuf)
- 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
- 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)
+
+
+def test_resize_payload():
+ """test the ability to insert a larger payload into an existing message"""
+ mtype = 99
+ subid = 100
+
+ mbuf = rmr.rmr_alloc_msg(MRC_SEND, 25) # allocate buffer with small payload
+ mbuf.contents.mtype = mtype # type and sub-id should not change
+ mbuf.contents.sub_id = subid
+
+ long_payload = b"This is a long payload that should force the message buffer to be reallocated"
+ rmr.set_payload_and_length(long_payload, mbuf)
+ summary = rmr.message_summary(mbuf)
+ assert summary["payload max size"] >= len(long_payload) # RMR may allocate a larger payload space
+ assert summary["payload length"] == len(long_payload) # however, the length must be exactly the same
+ assert summary["message type"] == mtype # both mtype and sub-id should be preserved in new
+ assert summary["subscription id"] == subid
Code Review / ric-plt / lib / rmr.git / blobdiff