1 # vim: ts=4 sw=4 expandtab:
2 # =================================================================================2
3 # Copyright (c) 2019-2020 Nokia
4 # Copyright (c) 2018-2020 AT&T Intellectual Property.
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
10 # http://www.apache.org/licenses/LICENSE-2.0
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 # ==================================================================================
20 from ricxappframe.rmr import rmr, helpers, exceptions
34 MRC_SEND = rmr.rmr_init(b"3562", rmr.RMR_MAX_RCV_BYTES, 0x00)
35 while rmr.rmr_ready(MRC_SEND) == 0:
39 MRC_RCV = rmr.rmr_init(b"3563", rmr.RMR_MAX_RCV_BYTES, 0x00)
40 while rmr.rmr_ready(MRC_RCV) == 0:
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:
49 def teardown_module():
51 test rmr module teardown
53 rmr.rmr_close(MRC_SEND)
54 rmr.rmr_close(MRC_RCV)
55 rmr.rmr_close(MRC_BUF_RCV)
58 def _assert_new_sbuf(sbuf):
60 verify the initial state of an alloced message is what we expect
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
75 test meid stringification
77 sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
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
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
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
89 rmr.rmr_set_meid(sbuf, b"\x01\x02")
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
94 assert len(rmr.rmr_get_meid(sbuf)) == 2
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)
101 def test_rmr_set_get():
105 sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
106 _assert_new_sbuf(sbuf)
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
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
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
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
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"
153 def test_rcv_timeout():
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.
158 There is no unit test for rmr_rcv_msg; too dangerous, that is a blocking call that may never return.
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
171 test send and receive
173 pay = b"\x01\x00\x80"
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"
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
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)
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)
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
210 def test_send_rcv_subid_good():
212 test send and receive where subid is used for routing
214 pay = b"\x01\x00\x80"
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)
224 # receive it in other context
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)
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
238 def test_send_rcv_subid_bad_subid():
240 test send and receive where subid is used for routing but nobody recieves this subid
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"
248 def test_send_rcv_subid_bad_mtype():
250 test send and receive where subid is used for routing but nobody recieves this mtype
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"
258 def send_burst(mrc, fmt, mtype=1, num=13, counter=0):
260 Internal function to support test_rcv_all.
261 Send a burst of messages optionally giving the type, payload
264 sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE) # seed message buffer
267 payload = bytes(fmt % counter, "UTF-8")
270 rmr.set_payload_and_length(payload, sbuf_send)
271 sbuf_send.contents.mtype = mtype
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
282 assert ms["message state"] == 0
283 assert max_retries > 0
288 test the ability to receive a batch of queued messages.
290 pay_fmt = "send to ring msg: %d" # dynamic message format with counter
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
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
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
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
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
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)
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
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
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)
340 def test_resize_payload():
341 """test the ability to insert a larger payload into an existing message"""
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
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