tests/test_rmr.py

   1 # vim: ts=4 sw=4 expandtab:
   2 # =================================================================================2
   3 #       Copyright (c) 2019-2020 Nokia
   4 #       Copyright (c) 2018-2020 AT&T Intellectual Property.
   5 #
   6 #   Licensed under the Apache License, Version 2.0 (the "License");
   7 #   you may not use this file except in compliance with the License.
   8 #   You may obtain a copy of the License at
   9 #
  10 #          http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 #   Unless required by applicable law or agreed to in writing, software
  13 #   distributed under the License is distributed on an "AS IS" BASIS,
  14 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 #   See the License for the specific language governing permissions and
  16 #   limitations under the License.
  17 # ==================================================================================
  18 import time
  19 import pytest
  20 from ricxappframe.rmr import rmr, helpers, exceptions
  21
  22
  23 SIZE = 256
  24 MRC_SEND = None
  25 MRC_RCV = None
  26 MRC_BUF_RCV = None
  27
  28
  29 def setup_module():
  30     """
  31     test_rmr module setup
  32     """
  33     global MRC_SEND
  34     MRC_SEND = rmr.rmr_init(b"3562", rmr.RMR_MAX_RCV_BYTES, 0x00)
  35     while rmr.rmr_ready(MRC_SEND) == 0:
  36         time.sleep(1)
  37
  38     global MRC_RCV
  39     MRC_RCV = rmr.rmr_init(b"3563", rmr.RMR_MAX_RCV_BYTES, 0x00)
  40     while rmr.rmr_ready(MRC_RCV) == 0:
  41         time.sleep(1)
  42
  43     global MRC_BUF_RCV
  44     MRC_BUF_RCV = rmr.rmr_init(b"3564", rmr.RMR_MAX_RCV_BYTES, rmr.RMRFL_MTCALL)
  45     while rmr.rmr_ready(MRC_BUF_RCV) == 0:
  46         time.sleep(1)
  47
  48
  49 def teardown_module():
  50     """
  51     test rmr module teardown
  52     """
  53     rmr.rmr_close(MRC_SEND)
  54     rmr.rmr_close(MRC_RCV)
  55     rmr.rmr_close(MRC_BUF_RCV)
  56
  57
  58 def _assert_new_sbuf(sbuf):
  59     """
  60     verify the initial state of an alloced message is what we expect
  61     """
  62     summary = rmr.message_summary(sbuf)
  63     assert summary["payload"] == b""
  64     assert summary["payload length"] == 0
  65     assert summary["subscription id"] == -1
  66     assert summary["transaction id"] == b""
  67     assert summary["message state"] == 0
  68     assert summary["message status"] == "RMR_OK"
  69     assert summary["meid"] == b""
  70     assert summary["errno"] == 0
  71
  72
  73 def test_meid():
  74     """
  75     test meid stringification
  76     """
  77     sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
  78
  79     rmr.rmr_set_meid(sbuf, b"\x01\x02")
  80     assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"\x01\x02"
  81     assert len(rmr.rmr_get_meid(sbuf)) == 2
  82
  83     rmr.rmr_set_meid(sbuf, b"\x00" * 31)
  84     assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b""  # NULL bytes get truncated
  85
  86     rmr.rmr_set_meid(sbuf, b"6" * 31)
  87     assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"6" * 31  # string in string out
  88
  89     rmr.rmr_set_meid(sbuf, b"\x01\x02")
  90     assert (
  91         rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"\x01\x02"
  92     )  # Ctypes will chop at first nil, so expect only 2 bytes back
  93
  94     assert len(rmr.rmr_get_meid(sbuf)) == 2
  95
  96     # test that an exception is raised for buffers which are too long
  97     with pytest.raises(exceptions.MeidSizeOutOfRange):
  98         rmr.rmr_set_meid(sbuf, b"8" * 32)
  99
 100
 101 def test_rmr_set_get():
 102     """
 103     test set functions
 104     """
 105     sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
 106     _assert_new_sbuf(sbuf)
 107
 108     # test payload
 109     pay = b"\x01\x00\x80"
 110     rmr.set_payload_and_length(pay, sbuf)
 111     summary = rmr.message_summary(sbuf)
 112     assert summary["payload"] == pay
 113     assert summary["payload length"] == 3
 114
 115     # test transid (note we cant test payload because it's randomly gen)
 116     assert summary["transaction id"] == b""
 117     assert len(summary["transaction id"]) == 0
 118     rmr.generate_and_set_transaction_id(sbuf)
 119     summary = rmr.message_summary(sbuf)
 120     assert summary["transaction id"] != b""
 121     assert len(summary["transaction id"]) == 32
 122
 123     # test meid
 124     assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b""
 125     rmr.rmr_set_meid(sbuf, b"666\x01\x00\x01")
 126     summary = rmr.message_summary(sbuf)
 127     assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b"666\x01"
 128     assert (len(summary["meid"])) == 4
 129
 130
 131 def test_alloc_fancy():
 132     """test allocation with setting payload, trans, mtype, subid"""
 133     pay = b"yoo\x01\x00\x80"
 134     sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE, payload=pay, gen_transaction_id=True, mtype=14, meid=b"asdf", sub_id=654321)
 135     summary = rmr.message_summary(sbuf)
 136     assert summary["payload"] == pay
 137     assert summary["payload length"] == 6
 138     assert summary["transaction id"] != b""  # hard to test what it will be, but make sure not empty
 139     assert len(summary["transaction id"]) == 32
 140     assert summary["message state"] == 0
 141     assert summary["message type"] == sbuf.contents.mtype == 14
 142     assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b"asdf"
 143     assert sbuf.contents.sub_id == summary["subscription id"] == 654321
 144
 145
 146 def test_alloc_overlapping_flags():
 147     """test allocation with setting the transaction id"""
 148     sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE, gen_transaction_id=True, fixed_transaction_id=b"6" * 32)
 149     summary = rmr.message_summary(sbuf)
 150     assert summary["transaction id"] == b"66666666666666666666666666666666"
 151
 152
 153 def test_rcv_timeout():
 154     """
 155     test torcv; this is a scary test because if it fails... it doesn't fail, it will run forever!
 156     We receive a message (though nothing has been sent) and make sure the function doesn't block forever.
 157
 158     There is no unit test for rmr_rcv_msg; too dangerous, that is a blocking call that may never return.
 159     """
 160     sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, SIZE)
 161     start_rcv_sec = time.time()
 162     sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 500)  # should wait a bit before returning
 163     summary = rmr.message_summary(sbuf_rcv)
 164     assert summary["message state"] == 12
 165     assert summary["message status"] == "RMR_ERR_TIMEOUT"
 166     assert(time.time() - start_rcv_sec > 0.5)  # test duration should be longer than the timeout
 167
 168
 169 def test_send_rcv():
 170     """
 171     test send and receive
 172     """
 173     pay = b"\x01\x00\x80"
 174
 175     # send a message
 176     sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
 177     _assert_new_sbuf(sbuf_send)
 178     rmr.set_payload_and_length(pay, sbuf_send)
 179     sbuf_send.contents.mtype = 0
 180     sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
 181     send_summary = rmr.message_summary(sbuf_send)
 182     assert send_summary["message state"] == 0  # if send fails don't attempt receive
 183     assert send_summary["message status"] == "RMR_OK"
 184     time.sleep(0.5)
 185
 186     # receive it in other context
 187     sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, SIZE)
 188     sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
 189     rcv_summary = rmr.message_summary(sbuf_rcv)
 190     assert rcv_summary["message state"] == 0
 191     assert rcv_summary["message status"] == "RMR_OK"
 192     assert rcv_summary["message type"] == 0
 193     assert rcv_summary["payload"] == pay
 194
 195     # send an ACK back
 196     ack_pay = b"message received"
 197     sbuf_rcv = rmr.rmr_rts_msg(MRC_RCV, sbuf_rcv, payload=ack_pay, mtype=6666)
 198     rcv_ack_summary = rmr.message_summary(sbuf_rcv)
 199
 200     # have the sender receive it
 201     sbuf_send = rmr.rmr_torcv_msg(MRC_SEND, sbuf_send, 2000)
 202     send_ack_summary = rmr.message_summary(sbuf_send)
 203
 204     assert send_ack_summary["message state"] == rcv_ack_summary["message state"] == 0
 205     assert send_ack_summary["message status"] == rcv_ack_summary["message status"] == "RMR_OK"
 206     assert send_ack_summary["payload"] == ack_pay
 207     assert send_ack_summary["message type"] == 6666
 208
 209
 210 def test_send_rcv_subid_good():
 211     """
 212     test send and receive where subid is used for routing
 213     """
 214     pay = b"\x01\x00\x80"
 215     test_mtype = 46656
 216     test_subid = 777
 217
 218     # send a message
 219     sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, pay, mtype=test_mtype, sub_id=test_subid)
 220     pre_send_summary = rmr.message_summary(sbuf_send)
 221     sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
 222     send_summary = rmr.message_summary(sbuf_send)
 223
 224     # receive it in other context
 225     time.sleep(0.5)
 226     sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, 3)
 227     sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
 228     rcv_summary = rmr.message_summary(sbuf_rcv)
 229
 230     # asserts
 231     assert send_summary["message state"] == rcv_summary["message state"] == 0
 232     assert send_summary["message status"] == rcv_summary["message status"] == "RMR_OK"
 233     assert pre_send_summary["payload"] == rcv_summary["payload"] == pay
 234     assert pre_send_summary["message type"] == rcv_summary["message type"] == test_mtype
 235     assert pre_send_summary["subscription id"] == rcv_summary["subscription id"] == test_subid
 236
 237
 238 def test_send_rcv_subid_bad_subid():
 239     """
 240     test send and receive where subid is used for routing but nobody recieves this subid
 241     """
 242     sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, b"\x01\x00\x80", mtype=46656, sub_id=778)
 243     sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
 244     assert rmr.message_summary(sbuf_send)["message state"] == 2
 245     assert rmr.message_summary(sbuf_send)["message status"] == "RMR_ERR_NOENDPT"
 246
 247
 248 def test_send_rcv_subid_bad_mtype():
 249     """
 250     test send and receive where subid is used for routing but nobody recieves this mtype
 251     """
 252     sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, b"\x01\x00\x80", mtype=46657, sub_id=777)
 253     sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
 254     assert rmr.message_summary(sbuf_send)["message state"] == 2
 255     assert rmr.message_summary(sbuf_send)["message status"] == "RMR_ERR_NOENDPT"
 256
 257
 258 def send_burst(mrc, fmt, mtype=1, num=13, counter=0):
 259     """
 260         Internal function to support test_rcv_all.
 261         Send a burst of messages optionally giving the type, payload
 262         and number to send.
 263     """
 264     sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE)  # seed message buffer
 265
 266     for i in range(num):
 267         payload = bytes(fmt % counter, "UTF-8")
 268         counter += 1
 269
 270         rmr.set_payload_and_length(payload, sbuf_send)
 271         sbuf_send.contents.mtype = mtype
 272
 273         max_retries = 5
 274         while max_retries > 0:
 275             sbuf_send = rmr.rmr_send_msg(mrc, sbuf_send)
 276             ms = rmr.message_summary(sbuf_send)
 277             if ms["message state"] != 10:  # 10 is retry
 278                 break
 279             max_retries -= 1
 280             time.sleep(0.75)
 281
 282         assert ms["message state"] == 0
 283         assert max_retries > 0
 284
 285
 286 def test_rcv_all():
 287     """
 288     test the ability to receive a batch of queued messages.
 289     """
 290     pay_fmt = "send to ring msg: %d"  # dynamic message format with counter
 291
 292     send_burst(MRC_SEND, pay_fmt)  # send a bunch of 13 messages that should queue
 293     time.sleep(1)  # ensure underlying transport gets cycles to send/receive
 294
 295     bundle = helpers.rmr_rcvall_msgs(MRC_BUF_RCV)  # use the buffered receiver to read all with a single call
 296     assert len(bundle) == 13
 297
 298     for i, ms in enumerate(bundle):
 299         ms = bundle[i]  # validate each summary returned, and ordering preserved
 300         assert ms["message state"] == 0
 301         expected_pay = bytes(pay_fmt % i, "UTF-8")
 302         assert ms["payload"] == expected_pay
 303
 304     send_burst(MRC_SEND, pay_fmt, mtype=1, num=10)  # send a second round with msg types 1 and 2 to test filter
 305     send_burst(MRC_SEND, pay_fmt, mtype=2, num=8)
 306     send_burst(MRC_SEND, pay_fmt, mtype=1, num=5)
 307     send_burst(MRC_SEND, pay_fmt, mtype=2, num=4, counter=8)  # total of 12 messages with type 2 should be queued
 308     time.sleep(1)  # ensure underlying transport gets cycles to send/receive
 309
 310     bundle = helpers.rmr_rcvall_msgs_raw(MRC_BUF_RCV, [2])  # receive only message type 2 messages
 311     assert len(bundle) == 12  # we should only get the type 2 batch of 12 messages
 312
 313     for i, (ms, sbuf) in enumerate(bundle):  # test the raw version
 314         test_summary = rmr.message_summary(sbuf)
 315         assert test_summary == ms
 316         assert ms["message state"] == 0  # all should be OK
 317         assert ms["message type"] == 2  # only mtype 2 should have been received
 318         expected_pay = bytes(pay_fmt % i, "UTF-8")  # ordering should still jive with the counter
 319         assert ms["payload"] == expected_pay
 320         rmr.rmr_free_msg(sbuf)
 321
 322     # check the timeout scenarios
 323     start_rcv_sec = time.time()
 324     bundle = helpers.rmr_rcvall_msgs(MRC_RCV, timeout=1001)  # non-zero timeout means wait
 325     assert len(bundle) == 0  # we should get none
 326     assert(time.time() - start_rcv_sec > 1)  # test duration should be longer than 1 second
 327
 328     start_rcv_sec = time.time()
 329     bundle = helpers.rmr_rcvall_msgs_raw(MRC_RCV, timeout=1002)  # non-zero timeout means wait
 330     assert len(bundle) == 0  # we should get none
 331     assert(time.time() - start_rcv_sec > 1)  # test duration should be longer than 1 second
 332
 333
 334 def test_bad_buffer():
 335     """test that we get a proper exception when the buffer has a null pointer"""
 336     with pytest.raises(exceptions.BadBufferAllocation):
 337         rmr.rmr_alloc_msg(None, 4096)
 338
 339
 340 def test_resize_payload():
 341     """test the ability to insert a larger payload into an existing message"""
 342     mtype = 99
 343     subid = 100
 344
 345     mbuf = rmr.rmr_alloc_msg(MRC_SEND, 25)  # allocate buffer with small payload
 346     mbuf.contents.mtype = mtype  # type and sub-id should not change
 347     mbuf.contents.sub_id = subid
 348
 349     long_payload = b"This is a long payload that should force the message buffer to be reallocated"
 350     rmr.set_payload_and_length(long_payload, mbuf)
 351     summary = rmr.message_summary(mbuf)
 352     assert summary["payload max size"] >= len(long_payload)  # RMR may allocate a larger payload space
 353     assert summary["payload length"] == len(long_payload)  # however, the length must be exactly the same
 354     assert summary["message type"] == mtype  # both mtype and sub-id should be preserved in new
 355     assert summary["subscription id"] == subid