# this is an alias to rmr_bytes2meid using familiar set/get terminoloigy
rmr_set_meid = rmr_bytes2meid
-
# CAUTION: Some of the C functions expect a mutable buffer to copy the bytes into;
# if there is a get_* function below, use it to set up and return the
# buffer properly.
-
+# extern unsigned char* rmr_get_meid(rmr_mbuf_t* mbuf, unsigned char* dest);
+# we don't provide direct access to this function (unless it is asked for) because it is not really useful to provide your own buffer.
+# Rather, rmr_get_meid does this for you, and just returns the string.
_rmr_get_meid = rmr_c_lib.rmr_get_meid
_rmr_get_meid.argtypes = [POINTER(rmr_mbuf_t), c_char_p]
_rmr_get_meid.restype = c_char_p
-def rmr_get_meid(ptr_mbuf, dest):
+def rmr_get_meid(ptr_mbuf):
"""
- Refer to the rmr C documentation for rmr_get_meid
- extern unsigned char* rmr_get_meid(rmr_mbuf_t* mbuf, unsigned char* dest);
+ Get the managed equipment ID (meid) from the message header.
+
+ Parameters
+ ----------
+ ptr_mbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+
+ Returns
+ -------
+ string:
+ meid
"""
- return _rmr_get_meid(ptr_mbuf, dest)
+ sz = _get_constants().get("RMR_MAX_MEID", 64) # size for buffer to fill
+ buf = create_string_buffer(sz)
+ _rmr_get_meid(ptr_mbuf, buf)
+ return buf.value.decode() # decode turns into a string
_rmr_get_src = rmr_c_lib.rmr_get_src
if ptr_mbuf.contents.len > RMR_MAX_RCV_BYTES:
return "Malformed message: message length is greater than the maximum possible"
- meid = get_meid(ptr_mbuf)
- if meid == "\000" * _get_constants().get("RMR_MAX_MEID", 32): # special case all nils
- meid = None
-
return {
"payload": get_payload(ptr_mbuf),
"payload length": ptr_mbuf.contents.len,
"message state": ptr_mbuf.contents.state,
"message status": _state_to_status(ptr_mbuf.contents.state),
"payload max size": rmr_payload_size(ptr_mbuf),
- "meid": meid,
+ "meid": rmr_get_meid(ptr_mbuf),
"message source": get_src(ptr_mbuf),
"errno": ptr_mbuf.contents.tp_state,
}
memmove(ptr_mbuf.contents.xaction, uu_id, sz)
-def get_meid(ptr_mbuf):
- """
- | Get the managed equipment ID (meid) from the message header.
- | This is a 32 byte field and RMr returns all 32 bytes which if the sender did not set will be garbage.
-
- Parameters
- ----------
- ptr_mbuf: ctypes c_void_p
- Pointer to an rmr message buffer
-
- Returns
- -------
- string:
- meid
- """
- sz = _get_constants().get("RMR_MAX_MEID", 64) # size for buffer to fill
- buf = create_string_buffer(sz)
- rmr_get_meid(ptr_mbuf, buf)
- return buf.raw.decode()
-
-
def get_src(ptr_mbuf):
"""
Get the message source (likely host:port)
Returns
-------
string:
- meid
+ message source
"""
sz = _get_constants().get("RMR_MAX_SRC", 64) # size to fill
buf = create_string_buffer(sz)
SIZE = 256
-MRC = None
+MRC_SEND = None
+MRC_RCV = None
def setup_module():
- global MRC
- MRC = rmr.rmr_init(b"4562", rmr.RMR_MAX_RCV_BYTES, 0x00)
- while rmr.rmr_ready(MRC) == 0:
+ """
+ test_rmr module setup
+ """
+ global MRC_SEND
+ MRC_SEND = rmr.rmr_init(b"4562", rmr.RMR_MAX_RCV_BYTES, 0x00)
+ while rmr.rmr_ready(MRC_SEND) == 0:
+ time.sleep(1)
+
+ global MRC_RCV
+ MRC_RCV = rmr.rmr_init(b"4563", rmr.RMR_MAX_RCV_BYTES, 0x00)
+ while rmr.rmr_ready(MRC_RCV) == 0:
time.sleep(1)
def teardown_module():
- rmr.rmr_close(MRC)
+ """
+ test rmr module teardown
+ """
+ rmr.rmr_close(MRC_SEND)
+ rmr.rmr_close(MRC_RCV)
def _assert_new_sbuf(sbuf):
assert summary["transaction id"] == b""
assert summary["message state"] == 0
assert summary["message status"] == "RMR_OK"
- assert summary["meid"] is None
+ assert summary["meid"] == ""
assert summary["errno"] == 0
assert rmr._state_to_status(666) == "UNKNOWN STATE"
-def test_meid_prettify():
+def test_meid():
"""
- test the printing of meid based on it's value
+ test meid stringification
"""
- # TODO?? weirdness: setting it takes bytes, but getting it returns a string. This does NOT happen for payload; bytes in, bytes come out.
- sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
+ sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
+
+ rmr.rmr_set_meid(sbuf, b"\x01\x02", 2)
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == "\x01\x02"
+ assert len(rmr.rmr_get_meid(sbuf)) == 2
+
rmr.rmr_set_meid(sbuf, b"\x00" * 32, 32)
- summary = rmr.message_summary(sbuf)
- assert summary["meid"] is None # summary does a pretty print" of 32 null bytes
- assert rmr.get_meid(sbuf) == "\x00" * 32 # real underlying value
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == "" # NULL bytes get truncated
+
+ rmr.rmr_set_meid(sbuf, b"6" * 32, 32)
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == "6" * 32 # string in string out
+
+ rmr.rmr_set_meid(sbuf, b"\x01\x02", 2)
+ assert (
+ rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == "\x01\x02" + "6" * 30
+ ) # bytes in string out, 6s left over
+ assert len(rmr.rmr_get_meid(sbuf)) == 32
def test_rmr_set_get():
"""
test set functions
"""
- sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
+ sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
_assert_new_sbuf(sbuf)
# test payload
assert len(summary["transaction id"]) == 32
# test meid
- assert rmr.get_meid(sbuf) == "\x00" * 32
- assert summary["meid"] is None # the summary printing function shows the above horridness as None
- rmr.rmr_set_meid(sbuf, b"666", 3)
+ assert rmr.rmr_get_meid(sbuf) == summary["meid"] == ""
+ rmr.rmr_set_meid(sbuf, b"666\x01\x00\x01", 6)
summary = rmr.message_summary(sbuf)
- # TODO?? weirdness: setting it takes bytes, but getting it returns a string. This does NOT happen for payload; bytes in, bytes come out.
- assert rmr.get_meid(sbuf) == summary["meid"] == "666" + "\x00" * 29
+ assert rmr.rmr_get_meid(sbuf) == summary["meid"] == "666\x01"
+ assert (len(summary["meid"])) == 4
+
+
+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.
+
+ There is no unit test for rmr_rcv_msg; too dangerous, that is a blocking call that may never return.
+ """
+ sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, SIZE)
+ sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 50) # should time out after 50ms
+ summary = rmr.message_summary(sbuf_rcv)
+ assert summary["message state"] == 12
+ assert summary["message status"] == "RMR_ERR_TIMEOUT"
+
+
+def test_send_rcv():
+ """
+ test send and receive
+ """
+ pay = b"\x01\x00\x80"
+
+ # send a message
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
+ _assert_new_sbuf(sbuf_send)
+ rmr.set_payload_and_length(pay, sbuf_send)
+ sbuf_send.contents.mtype = 0
+ sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
+ send_summary = rmr.message_summary(sbuf_send)
+
+ # 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 type"] == 0
+ assert send_summary["message state"] == rcv_summary["message state"] == 0
+ assert send_summary["message status"] == rcv_summary["message status"] == "RMR_OK"
+
+ # send an ACK back
+ ack_pay = b"message recieved"
+ 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
+ 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"
Code Review / ric-plt / lib / rmr.git / commitdiff