1 # =================================================================================2
2 # Copyright (c) 2019-2020 Nokia
3 # Copyright (c) 2018-2020 AT&T Intellectual Property.
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
9 # http://www.apache.org/licenses/LICENSE-2.0
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 # ==================================================================================
19 from ricxappframe.rmr import rmr, helpers, exceptions
33 MRC_SEND = rmr.rmr_init(b"3562", rmr.RMR_MAX_RCV_BYTES, 0x00)
34 while rmr.rmr_ready(MRC_SEND) == 0:
38 MRC_RCV = rmr.rmr_init(b"3563", rmr.RMR_MAX_RCV_BYTES, 0x00)
39 while rmr.rmr_ready(MRC_RCV) == 0:
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:
48 def teardown_module():
50 test rmr module teardown
52 rmr.rmr_close(MRC_SEND)
53 rmr.rmr_close(MRC_RCV)
54 rmr.rmr_close(MRC_BUF_RCV)
57 def _assert_new_sbuf(sbuf):
59 verify the initial state of an alloced message is what we expect
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
74 test meid stringification
76 sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
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
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
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
88 rmr.rmr_set_meid(sbuf, b"\x01\x02")
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
93 assert len(rmr.rmr_get_meid(sbuf)) == 2
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)
100 def test_rmr_set_get():
104 sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
105 _assert_new_sbuf(sbuf)
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
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
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
129 # exercise the logging setter; cannot test the result
130 rmr.rmr_set_vlevel(0)
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
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"
155 def test_rcv_timeout():
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.
160 There is no unit test for rmr_rcv_msg; too dangerous, that is a blocking call that may never return.
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
173 test send and receive
175 pay = b"\x01\x00\x80"
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"
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
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)
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)
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
212 def test_send_rcv_subid_good():
214 test send and receive where subid is used for routing
216 pay = b"\x01\x00\x80"
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)
226 # receive it in other context
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)
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
240 def test_send_rcv_subid_bad_subid():
242 test send and receive where subid is used for routing but nobody receives this subid
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"
250 def test_send_rcv_subid_bad_mtype():
252 test send and receive where subid is used for routing but nobody receives this mtype
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"
260 def send_burst(mrc, fmt, mtype=1, num=13, counter=0):
262 Internal function to support test_rcv_all.
263 Send a burst of messages optionally giving the type, payload
266 sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE) # seed message buffer
269 payload = bytes(fmt % counter, "UTF-8")
272 rmr.set_payload_and_length(payload, sbuf_send)
273 sbuf_send.contents.mtype = mtype
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:
284 assert ms[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
285 assert max_retries > 0
290 test the ability to receive a batch of queued messages.
292 pay_fmt = "send to ring msg: %d" # dynamic message format with counter
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
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
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
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
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
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)
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
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
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)
342 def test_resize_payload():
343 """test the ability to insert a larger payload into an existing message"""
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
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
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")
366 state = rmr.rmr_wh_state(MRC_SEND, whid)
367 assert state == rmr.RMR_OK
369 sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
370 _assert_new_sbuf(sbuf_send)
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)
377 # send via call, but don't wait long for a response
378 rmr.rmr_wh_call(MRC_SEND, whid, sbuf_send, 1, 100)
380 # receive message in other context
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)
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
391 sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
392 _assert_new_sbuf(sbuf_send)
394 sbuf_send.contents.mtype = mtype
395 payload = b"Birds like worms"
396 rmr.set_payload_and_length(payload, sbuf_send)
398 # send without waiting for a response
399 rmr.rmr_wh_send_msg(MRC_SEND, whid, sbuf_send)
401 # receive message in other context
403 sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
404 rcv_summary = rmr.message_summary(sbuf_rcv)
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
411 rmr.rmr_wh_close(MRC_SEND, whid)
412 # don't check state after close, connection is always reported as open