-# ==================================================================================
+# vim: ts=4 sw=4 expandtab:
+# =================================================================================2
# Copyright (c) 2019 Nokia
# Copyright (c) 2018-2019 AT&T Intellectual Property.
#
# ==================================================================================
import time
import pytest
-from rmr import rmr
+from rmr import rmr, helpers
SIZE = 256
while rmr.rmr_ready(MRC_RCV) == 0:
time.sleep(1)
+ 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:
+ time.sleep(1)
def teardown_module():
"""
def test_get_constants(expected_constants):
"""
- test getting constants
+ test getting constants. We don't care what values are returned as those
+ should be meaningful only to RMR. We do care that all of the constants
+ which are defined in expected_contents are returned. Further, we don't
+ consider it to be an error if the returned list has more constants than
+ what are in our list.
+
+ To avoid frustration, this should list all missing keys, not fail on the
+ first missing key.
"""
- assert rmr._get_constants() == expected_constants
+ 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
+ errors += 1
+ print( "did not find required constant in list from RMR: %s" % key )
+
+ assert errors == 0
def test_get_mapping_dict(expected_states):
def test_rcv_timeout():
"""
test torcv; this is a scary test because if it fails... it doesn't fail, it will run forever!
- We recieve a message (though nothing has been sent) and make sure the function doesn't block forever.
+ We receive a message (though nothing has been sent) and make sure the function doesn't block forever.
There is no unit test for rmr_rcv_msg; too dangerous, that is a blocking call that may never return.
"""
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 status"] == "RMR_OK"
+ time.sleep(.5)
# receive it in other context
sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, SIZE)
sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
rcv_summary = rmr.message_summary(sbuf_rcv)
- assert rcv_summary["payload"] == pay
+ assert rcv_summary["message state"] == 0
+ assert rcv_summary["message status"] == "RMR_OK"
assert rcv_summary["message type"] == 0
- assert send_summary["message state"] == rcv_summary["message state"] == 0
- assert send_summary["message status"] == rcv_summary["message status"] == "RMR_OK"
+ assert rcv_summary["payload"] == pay
# send an ACK back
- ack_pay = b"message recieved"
+ ack_pay = b"message received"
rmr.set_payload_and_length(ack_pay, sbuf_rcv)
sbuf_rcv = rmr.rmr_rts_msg(MRC_RCV, sbuf_rcv)
rcv_ack_summary = rmr.message_summary(sbuf_rcv)
- # have the sender recieve it
+ # have the sender receive it
sbuf_send = rmr.rmr_torcv_msg(MRC_SEND, sbuf_send, 2000)
send_ack_summary = rmr.message_summary(sbuf_send)
- assert send_ack_summary["payload"] == ack_pay
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
+
+
+def send_burst(mrc, fmt, mtype=1, num=13, counter=0):
+ """
+ Internal function to support test_rcv_all.
+ 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
+
+ for i in range(num):
+ payload = bytes(fmt % counter, "UTF-8")
+ counter += 1
+
+ rmr.set_payload_and_length(payload, sbuf_send)
+ sbuf_send.contents.mtype = mtype
+
+ max_retries = 5
+ 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
+ break
+ max_retries -= 1
+ time.sleep(.75)
+
+ assert ms["message state"] == 0
+ assert max_retries > 0
+
+
+def test_rcv_all():
+ """
+ test the ability to receive a batch of queued messages.
+ """
+ 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
+
+ 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
+ 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=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
+
+ 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
Code Review / ric-plt / lib / rmr.git / blobdiff