tests/test_rmr.py

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