The decision was made to move rmr python into the xapp frame.
The module name has not been changed otherwise, so apps transitioning
to this should only need to prefix "rmr" with "ricxappframe." to
transition (and, replace rmr with ricxappframe if they are not already
using the framework).
Some small changes have been made in the unit tests, such as port
numbers and monkeypatching statements, but otherwise all "new" code is
identical to rmr python.
Issue-ID: RIC-228
Change-Id: I5e80cfd39c1511db93b95ac5e442a2acfc0733a2
Signed-off-by: Tommy Carpenter <tc677g@att.com>
:depth: 3
:local:
+[1.0.0] - 4/6/2020
+-------------------
+::
+
+ * Python rmr has been moved into this repo. The module name has NOT changed in order to make the transition for repos very easy. The only transition needed should be prefixing rmr with ricxappframe in import statements, and to include this rather than rmr in setup.
+
+
[0.7.0] - 4/2/2020
-------------------
::
--- /dev/null
+# ==================================================================================
+# Copyright (c) 2019 Nokia
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
--- /dev/null
+# ==================================================================================
+# Copyright (c) 2019-2020 Nokia
+# Copyright (c) 2018-2020 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+"""
+Custom Exceptions
+"""
+
+
+class BadBufferAllocation(BaseException):
+ """a bad buffer was allocated, check the rmr context"""
+
+
+class MeidSizeOutOfRange(BaseException):
+ """an attempt to set the MEID with a buffer that was too large"""
+
+
+class InitFailed(BaseException):
+ """rmr init failure, the context is unusable"""
--- /dev/null
+# vim: ts=4 sw=4 expandtab:
+# ==================================================================================
+# Copyright (c) 2019 Nokia
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+
+# Mnemonic: helpers.py
+# Abstract: This is a colleciton of extensions to the RMR base package
+# which are likely to be convenient for python programmes.
+# Date: 26 September 2019
+# ---------------------------------------------------------------------------
+
+from ricxappframe.rmr import rmr
+
+
+def rmr_rcvall_msgs(mrc, pass_filter=[]):
+ """
+ Assemble an array of all messages which can be received without
+ blocking. Effectively draining the message queue if RMR is started
+ in mt-call mode, or draining any waiting TCP buffers. If the
+ pass_filter parameter is supplied it is treated as one or more message
+ types to accept (pass through). Using the default, an empty list, results
+ in messages with any type being captured.
+
+ Parameters
+ ----------
+ mrc: ctypes c_void_p
+ Pointer to the RMR context
+
+ pass_filter: list (optional)
+ The message type(s) to capture.
+
+ Returns
+ -------
+ list of dict
+ List of message summaries, one for each message captured.
+ """
+
+ new_messages = []
+ mbuf = rmr.rmr_alloc_msg(mrc, 4096) # allocate buffer to have something for a return status
+
+ while True:
+ mbuf = rmr.rmr_torcv_msg(mrc, mbuf, 0) # set the timeout to 0 so this doesn't block!!
+
+ summary = rmr.message_summary(mbuf)
+ if summary["message status"] != "RMR_OK": # ok indicates msg received, stop on all other states
+ break
+
+ if len(pass_filter) == 0 or summary["message type"] in pass_filter: # no filter, or passes; capture it
+ new_messages.append(summary)
+
+ rmr.rmr_free_msg(mbuf) # must free message to avoid leak
+ return new_messages
+
+
+def rmr_rcvall_msgs_raw(mrc, pass_filter=[]):
+ """
+ Same as rmr_rcvall_msgs, but the raw sbuf is also returned.
+ Useful, for example, if rts is to be used.
+
+ Parameters
+ ----------
+ mrc: ctypes c_void_p
+ Pointer to the RMR context
+
+ pass_filter: list (optional)
+ The message type(s) to capture.
+
+ Returns
+ -------
+ list of tuple:$
+ List of tuples [(S, sbuf),...] where S is a message summary and sbuf is the raw message$
+ the caller is responsible for calling rmr.rmr_free_msg(sbuf) for each sbuf afterwards to prevent memory leaks.
+ """
+
+ new_messages = []
+
+ while True:
+ mbuf = rmr.rmr_alloc_msg(mrc, 4096) # allocate buffer to have something for a return status
+ mbuf = rmr.rmr_torcv_msg(mrc, mbuf, 0) # set the timeout to 0 so this doesn't block!!
+ summary = rmr.message_summary(mbuf)
+ if summary["message status"] != "RMR_OK":
+ break
+
+ if len(pass_filter) == 0 or mbuf.contents.mtype in pass_filter: # no filter, or passes; capture it
+ new_messages.append((summary, mbuf))
+ else:
+ rmr.rmr_free_msg(mbuf)
+
+ return new_messages
--- /dev/null
+# vim: expandtab ts=4 sw=4:
+# ==================================================================================
+# Copyright (c) 2019-2020 Nokia
+# Copyright (c) 2018-2020 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+import uuid
+import json
+from ctypes import RTLD_GLOBAL, Structure, c_int, POINTER, c_char, c_char_p, c_void_p, memmove, cast
+from ctypes import CDLL
+from ctypes import create_string_buffer
+from ricxappframe.rmr.exceptions import BadBufferAllocation, MeidSizeOutOfRange, InitFailed
+
+# https://docs.python.org/3.7/library/ctypes.html
+# https://stackoverflow.com/questions/2327344/ctypes-loading-a-c-shared-library-that-has-dependencies/30845750#30845750
+# make sure you do a set -x LD_LIBRARY_PATH /usr/local/lib/;
+
+# even though we don't use these directly, they contain symbols we need
+rmr_c_lib = CDLL("librmr_si.so", mode=RTLD_GLOBAL)
+
+
+# Internal Helpers (not a part of public api)
+
+
+_rmr_const = rmr_c_lib.rmr_get_consts
+_rmr_const.argtypes = []
+_rmr_const.restype = c_char_p
+
+
+def _get_constants(cache={}):
+ """
+ Get or build needed constants from rmr
+ TODO: are there constants that end user applications need?
+ """
+ if cache:
+ return cache
+
+ js = _rmr_const() # read json string
+ cache = json.loads(str(js.decode())) # create constants value object as a hash
+ return cache
+
+
+def _get_mapping_dict(cache={}):
+ """
+ Get or build the state mapping dict
+
+ RMR_OK 0 state is good
+ RMR_ERR_BADARG 1 argument passd to function was unusable
+ RMR_ERR_NOENDPT 2 send/call could not find an endpoint based on msg type
+ RMR_ERR_EMPTY 3 msg received had no payload; attempt to send an empty message
+ RMR_ERR_NOHDR 4 message didn't contain a valid header
+ RMR_ERR_SENDFAILED 5 send failed; errno has nano reason
+ RMR_ERR_CALLFAILED 6 unable to send call() message
+ RMR_ERR_NOWHOPEN 7 no wormholes are open
+ RMR_ERR_WHID 8 wormhole id was invalid
+ RMR_ERR_OVERFLOW 9 operation would have busted through a buffer/field size
+ RMR_ERR_RETRY 10 request (send/call/rts) failed, but caller should retry (EAGAIN for wrappers)
+ RMR_ERR_RCVFAILED 11 receive failed (hard error)
+ RMR_ERR_TIMEOUT 12 message processing call timed out
+ RMR_ERR_UNSET 13 the message hasn't been populated with a transport buffer
+ RMR_ERR_TRUNC 14 received message likely truncated
+ RMR_ERR_INITFAILED 15 initialization of something (probably message) failed
+
+ """
+ if cache:
+ return cache
+
+ rmr_consts = _get_constants()
+ for key in rmr_consts: # build the state mapping dict
+ if key[:7] in ["RMR_ERR", "RMR_OK"]:
+ en = int(rmr_consts[key])
+ cache[en] = key
+
+ return cache
+
+
+def _state_to_status(stateno):
+ """
+ Convert a msg state to status
+
+ """
+ sdict = _get_mapping_dict()
+ return sdict.get(stateno, "UNKNOWN STATE")
+
+
+_RCONST = _get_constants()
+
+
+##############
+# PUBLIC API
+##############
+
+
+# These constants are directly usable by importers of this library
+# TODO: Are there others that will be useful?
+
+RMR_MAX_RCV_BYTES = _RCONST["RMR_MAX_RCV_BYTES"]
+RMRFL_MTCALL = _RCONST.get("RMRFL_MTCALL", 0x02) # initialization flags
+RMRFL_NONE = _RCONST.get("RMRFL_NONE", 0x0)
+RMR_OK = _RCONST["RMR_OK"] # useful state constants
+RMR_ERR_TIMEOUT = _RCONST["RMR_ERR_TIMEOUT"]
+RMR_ERR_RETRY = _RCONST["RMR_ERR_RETRY"]
+
+
+class rmr_mbuf_t(Structure):
+ """
+ Reimplementation of rmr_mbuf_t which is in an unaccessible header file (src/common/include/rmr.h)
+
+ | typedef struct {
+ | int state; // state of processing
+ | int mtype; // message type
+ | int len; // length of data in the payload (send or received)
+ | unsigned char* payload; // transported data
+ | unsigned char* xaction; // pointer to fixed length transaction id bytes
+ | int sub_id; // subscription id
+ | int tp_state; // transport state (a.k.a errno)
+ |
+ | these things are off limits to the user application
+ |
+ | void* tp_buf; // underlying transport allocated pointer (e.g. nng message)
+ | void* header; // internal message header (whole buffer: header+payload)
+ | unsigned char* id; // if we need an ID in the message separate from the xaction id
+ | int flags; // various MFL (private) flags as needed
+ | int alloc_len; // the length of the allocated space (hdr+payload)
+ | } rmr_mbuf_t;
+
+ We do not include the fields we are not supposed to mess with
+
+ RE PAYLOADs type below, see the documentation for c_char_p:
+ class ctypes.c_char_p
+ Represents the C char * datatype when it points to a zero-terminated string. For a general character pointer that may also point to binary data, POINTER(c_char) must be used. The constructor accepts an integer address, or a bytes object.
+ """
+
+ _fields_ = [
+ ("state", c_int),
+ ("mtype", c_int),
+ ("len", c_int),
+ (
+ "payload",
+ POINTER(c_char),
+ ), # according to th following the python bytes are already unsinged https://bytes.com/topic/python/answers/695078-ctypes-unsigned-char
+ ("xaction", POINTER(c_char)),
+ ("sub_id", c_int),
+ ("tp_state", c_int),
+ ]
+
+
+# argtypes and restype are important: https://stackoverflow.com/questions/24377845/ctype-why-specify-argtypes
+
+
+_rmr_init = rmr_c_lib.rmr_init
+_rmr_init.argtypes = [c_char_p, c_int, c_int]
+_rmr_init.restype = c_void_p
+
+
+def rmr_init(uproto_port, max_msg_size, flags):
+ """
+ Refer to rmr C documentation for rmr_init
+ extern void* rmr_init(char* uproto_port, int max_msg_size, int flags)
+
+ This python function checks that the context is not None and raises
+ an excption if it is.
+ """
+ mrc = _rmr_init(uproto_port, max_msg_size, flags)
+ if mrc is None:
+ raise InitFailed()
+ return mrc
+
+
+_rmr_ready = rmr_c_lib.rmr_ready
+_rmr_ready.argtypes = [c_void_p]
+_rmr_ready.restype = c_int
+
+
+def rmr_ready(vctx):
+ """
+ Refer to rmr C documentation for rmr_ready
+ extern int rmr_ready(void* vctx)
+ """
+ return _rmr_ready(vctx)
+
+
+_rmr_close = rmr_c_lib.rmr_close
+_rmr_close.argtypes = [c_void_p]
+
+
+def rmr_close(vctx):
+ """
+ Refer to rmr C documentation for rmr_close
+ extern void rmr_close(void* vctx)
+ """
+ return _rmr_close(vctx)
+
+
+_rmr_set_stimeout = rmr_c_lib.rmr_set_stimeout
+_rmr_set_stimeout.argtypes = [c_void_p, c_int]
+_rmr_set_stimeout.restype = c_int
+
+
+def rmr_set_stimeout(vctx, time):
+ """
+ Refer to the rmr C documentation for rmr_set_stimeout
+ extern int rmr_set_stimeout(void* vctx, int time)
+ """
+ return _rmr_set_stimeout(vctx, time)
+
+
+_rmr_alloc_msg = rmr_c_lib.rmr_alloc_msg
+_rmr_alloc_msg.argtypes = [c_void_p, c_int]
+_rmr_alloc_msg.restype = POINTER(rmr_mbuf_t)
+
+
+def rmr_alloc_msg(
+ vctx, size, payload=None, gen_transaction_id=False, mtype=None, meid=None, sub_id=None, fixed_transaction_id=None
+):
+ """
+ Refer to the rmr C documentation for rmr_alloc_msg
+ extern rmr_mbuf_t* rmr_alloc_msg(void* vctx, int size)
+ TODO: on next API break, clean up transaction_id ugliness. Kept for now to preserve API.
+
+ if payload is not None, attempts to set the payload
+ if gen_transaction_id is True, it generates and sets a transaction id. Note, fixed_transaction_id supersedes this option
+ if mtype is not None, sets the sbuf's message type
+ if meid is not None, sets the sbuf's meid
+ if sub_id is not None, sets the sbud's subscription id
+ if fixed_transaction_id is set, it deterministically sets the transaction_id. This overrides the option gen_transation_id
+
+ """
+ sbuf = _rmr_alloc_msg(vctx, size)
+ try:
+ # make sure the alloc worked
+ sbuf.contents
+
+ # set specified fields
+ if payload:
+ set_payload_and_length(payload, sbuf)
+
+ if fixed_transaction_id:
+ set_transaction_id(sbuf, fixed_transaction_id)
+ elif gen_transaction_id:
+ generate_and_set_transaction_id(sbuf)
+
+ if mtype:
+ sbuf.contents.mtype = mtype
+
+ if meid:
+ rmr_set_meid(sbuf, meid)
+
+ if sub_id:
+ sbuf.contents.sub_id = sub_id
+
+ return sbuf
+
+ except ValueError:
+ raise BadBufferAllocation
+
+
+_rmr_realloc_payload = rmr_c_lib.rmr_realloc_payload
+_rmr_realloc_payload.argtypes = [POINTER(rmr_mbuf_t), c_int, c_int, c_int] # new_len, copy, clone
+_rmr_realloc_payload.restype = POINTER(rmr_mbuf_t)
+
+
+def rmr_realloc_payload(ptr_mbuf, new_len, copy=False, clone=False):
+ """
+ Refer to the rmr C documentation for rmr_realloc_payload().
+ extern rmr_mbuf_t* rmr_realloc_payload(rmr_mbuf_t*, int, int, int)
+ """
+ return _rmr_realloc_payload(ptr_mbuf, new_len, copy, clone)
+
+
+_rmr_free_msg = rmr_c_lib.rmr_free_msg
+_rmr_free_msg.argtypes = [c_void_p]
+_rmr_free_msg.restype = None
+
+
+def rmr_free_msg(mbuf):
+ """
+ Refer to the rmr C documentation for rmr_free_msg
+ extern void rmr_free_msg(rmr_mbuf_t* mbuf )
+ """
+ if mbuf is not None:
+ _rmr_free_msg(mbuf)
+
+
+_rmr_payload_size = rmr_c_lib.rmr_payload_size
+_rmr_payload_size.argtypes = [POINTER(rmr_mbuf_t)]
+_rmr_payload_size.restype = c_int
+
+
+def rmr_payload_size(ptr_mbuf):
+ """
+ Refer to the rmr C documentation for rmr_payload_size
+ extern int rmr_payload_size(rmr_mbuf_t* msg)
+ """
+ return _rmr_payload_size(ptr_mbuf)
+
+
+"""
+The following functions all seem to have the same interface
+"""
+
+_rmr_send_msg = rmr_c_lib.rmr_send_msg
+_rmr_send_msg.argtypes = [c_void_p, POINTER(rmr_mbuf_t)]
+_rmr_send_msg.restype = POINTER(rmr_mbuf_t)
+
+
+def rmr_send_msg(vctx, ptr_mbuf):
+ """
+ Refer to the rmr C documentation for rmr_send_msg
+ extern rmr_mbuf_t* rmr_send_msg(void* vctx, rmr_mbuf_t* msg)
+ """
+ return _rmr_send_msg(vctx, ptr_mbuf)
+
+
+# TODO: the old message (Send param) is actually optional, but I don't know how to specify that in Ctypes.
+_rmr_rcv_msg = rmr_c_lib.rmr_rcv_msg
+_rmr_rcv_msg.argtypes = [c_void_p, POINTER(rmr_mbuf_t)]
+_rmr_rcv_msg.restype = POINTER(rmr_mbuf_t)
+
+
+def rmr_rcv_msg(vctx, ptr_mbuf):
+ """
+ Refer to the rmr C documentation for rmr_rcv_msg
+ extern rmr_mbuf_t* rmr_rcv_msg(void* vctx, rmr_mbuf_t* old_msg)
+ """
+ return _rmr_rcv_msg(vctx, ptr_mbuf)
+
+
+_rmr_torcv_msg = rmr_c_lib.rmr_torcv_msg
+_rmr_torcv_msg.argtypes = [c_void_p, POINTER(rmr_mbuf_t), c_int]
+_rmr_torcv_msg.restype = POINTER(rmr_mbuf_t)
+
+
+def rmr_torcv_msg(vctx, ptr_mbuf, ms_to):
+ """
+ Refer to the rmr C documentation for rmr_torcv_msg
+ extern rmr_mbuf_t* rmr_torcv_msg(void* vctx, rmr_mbuf_t* old_msg, int ms_to)
+ """
+ return _rmr_torcv_msg(vctx, ptr_mbuf, ms_to)
+
+
+_rmr_rts_msg = rmr_c_lib.rmr_rts_msg
+_rmr_rts_msg.argtypes = [c_void_p, POINTER(rmr_mbuf_t)]
+_rmr_rts_msg.restype = POINTER(rmr_mbuf_t)
+
+
+def rmr_rts_msg(vctx, ptr_mbuf, payload=None, mtype=None):
+ """
+ Refer to the rmr C documentation for rmr_rts_msg
+ extern rmr_mbuf_t* rmr_rts_msg(void* vctx, rmr_mbuf_t* msg)
+
+ additional features beyond c-rmr:
+ if payload is not None, attempts to set the payload
+ if mtype is not None, sets the sbuf's message type
+ """
+
+ if payload:
+ set_payload_and_length(payload, ptr_mbuf)
+
+ if mtype:
+ ptr_mbuf.contents.mtype = mtype
+
+ return _rmr_rts_msg(vctx, ptr_mbuf)
+
+
+_rmr_call = rmr_c_lib.rmr_call
+_rmr_call.argtypes = [c_void_p, POINTER(rmr_mbuf_t)]
+_rmr_call.restype = POINTER(rmr_mbuf_t)
+
+
+def rmr_call(vctx, ptr_mbuf):
+ """
+ Refer to the rmr C documentation for rmr_call
+ extern rmr_mbuf_t* rmr_call(void* vctx, rmr_mbuf_t* msg)
+ """
+ return _rmr_call(vctx, ptr_mbuf)
+
+
+_rmr_bytes2meid = rmr_c_lib.rmr_bytes2meid
+_rmr_bytes2meid.argtypes = [POINTER(rmr_mbuf_t), c_char_p, c_int]
+_rmr_bytes2meid.restype = c_int
+
+
+def rmr_set_meid(ptr_mbuf, byte_str):
+ """
+ Refer to the rmr C documentation for rmr_bytes2meid
+ extern int rmr_bytes2meid(rmr_mbuf_t* mbuf, unsigned char const* src, int len);
+
+ Caution: the meid length supported in an RMR message is 32 bytes, but C applications
+ expect this to be a nil terminated string and thus only 31 bytes are actually available.
+
+ Raises: exceptions.MeidSizeOutOfRang
+ """
+ max = _get_constants().get("RMR_MAX_MEID", 32)
+ if len(byte_str) >= max:
+ raise MeidSizeOutOfRange
+
+ return _rmr_bytes2meid(ptr_mbuf, byte_str, len(byte_str))
+
+
+# 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):
+ """
+ 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
+ """
+ sz = _get_constants().get("RMR_MAX_MEID", 32) # size for buffer to fill
+ buf = create_string_buffer(sz)
+ _rmr_get_meid(ptr_mbuf, buf)
+ return buf.value
+
+
+_rmr_get_src = rmr_c_lib.rmr_get_src
+_rmr_get_src.argtypes = [POINTER(rmr_mbuf_t), c_char_p]
+_rmr_get_src.restype = c_char_p
+
+
+def rmr_get_src(ptr_mbuf, dest):
+ """
+ Refer to the rmr C documentation for rmr_get_src
+ extern unsigned char* rmr_get_src(rmr_mbuf_t* mbuf, unsigned char* dest);
+ """
+ return _rmr_get_src(ptr_mbuf, dest)
+
+
+# Methods that exist ONLY in rmr-python, and are not wrapped methods
+# In hindsight, I wish i put these in a seperate module, but leaving this here to prevent api breakage.
+
+
+def get_payload(ptr_mbuf):
+ """
+ Given a rmr_buf_t*, get it's binary payload as a bytes object
+
+ Parameters
+ ----------
+ ptr_mbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+
+ Returns
+ -------
+ bytes:
+ the message payload
+ """
+ # Logic came from the answer here: https://stackoverflow.com/questions/55103298/python-ctypes-read-pointerc-char-in-python
+ sz = ptr_mbuf.contents.len
+ CharArr = c_char * sz
+ return CharArr(*ptr_mbuf.contents.payload[:sz]).raw
+
+
+def get_xaction(ptr_mbuf):
+ """
+ given a rmr_buf_t*, get it's transaction id
+
+ Parameters
+ ----------
+ ptr_mbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+
+ Returns
+ -------
+ bytes:
+ the transaction id
+ """
+ val = cast(ptr_mbuf.contents.xaction, c_char_p).value
+ sz = _get_constants().get("RMR_MAX_XID", 0)
+ return val[:sz]
+
+
+def message_summary(ptr_mbuf):
+ """
+ Returns a dict that contains the fields of a message
+
+ Parameters
+ ----------
+ ptr_mbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+
+ Returns
+ -------
+ dict:
+ dict message summary
+ """
+ return {
+ "payload": get_payload(ptr_mbuf) if ptr_mbuf.contents.state == RMR_OK else None,
+ "payload length": ptr_mbuf.contents.len,
+ "message type": ptr_mbuf.contents.mtype,
+ "subscription id": ptr_mbuf.contents.sub_id,
+ "transaction id": get_xaction(ptr_mbuf),
+ "message state": ptr_mbuf.contents.state,
+ "message status": _state_to_status(ptr_mbuf.contents.state),
+ "payload max size": rmr_payload_size(ptr_mbuf),
+ "meid": rmr_get_meid(ptr_mbuf),
+ "message source": get_src(ptr_mbuf),
+ "errno": ptr_mbuf.contents.tp_state,
+ }
+
+
+def set_payload_and_length(byte_str, ptr_mbuf):
+ """
+ | Set an rmr payload and content length
+ | In place method, no return
+
+ Parameters
+ ----------
+ byte_str: bytes
+ the bytes to set the payload to
+ ptr_mbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+ """
+ if rmr_payload_size(ptr_mbuf) < len(byte_str): # existing message payload too small
+ ptr_mbuf = rmr_realloc_payload(ptr_mbuf, len(byte_str), True)
+
+ memmove(ptr_mbuf.contents.payload, byte_str, len(byte_str))
+ ptr_mbuf.contents.len = len(byte_str)
+
+
+def generate_and_set_transaction_id(ptr_mbuf):
+ """
+ Generate a UUID and Set an rmr transaction id to it
+
+ Parameters
+ ----------
+ ptr_mbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+ """
+ set_transaction_id(ptr_mbuf, uuid.uuid1().hex.encode("utf-8"))
+
+
+def set_transaction_id(ptr_mbuf, tid_bytes):
+ """
+ Set an rmr transaction id
+ TODO: on next API break, merge these two functions. Not done now to preserve API.
+
+ Parameters
+ ----------
+ ptr_mbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+ tid_bytes: bytes
+ bytes of the desired transaction id
+ """
+ sz = _get_constants().get("RMR_MAX_XID", 0)
+ memmove(ptr_mbuf.contents.xaction, tid_bytes, sz)
+
+
+def get_src(ptr_mbuf):
+ """
+ Get the message source (likely host:port)
+
+ Parameters
+ ----------
+ ptr_mbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+
+ Returns
+ -------
+ string:
+ message source
+ """
+ sz = _get_constants().get("RMR_MAX_SRC", 64) # size to fill
+ buf = create_string_buffer(sz)
+ rmr_get_src(ptr_mbuf, buf)
+ return buf.value.decode()
--- /dev/null
+# ==================================================================================
+# Copyright (c) 2019 Nokia
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+
+"""
+Provides mocks that are useful for end applications unit testing
+"""
+
+import json
+import uuid
+
+
+def rcv_mock_generator(msg_payload, msg_type, msg_state, jsonb, timeout=0):
+ """
+ generates a mock function that can be used to monkeypatch rmr_torcv_msg or rmr_rcv_msg
+ """
+
+ def f(_mrc, sbuf, _timeout=timeout): # last param is needed for calls to rmr_torcv_msg, but not in rmr_rcv_msg
+ sbuf.contents.mtype = msg_type
+ payload = json.dumps(msg_payload).encode("utf-8") if jsonb else msg_payload
+ sbuf.contents.payload = payload
+ sbuf.contents.len = len(payload)
+ sbuf.contents.state = msg_state
+ if msg_state != 0: # set something in transport state if 'error'
+ sbuf.contents.tp_state = 99
+ else:
+ sbuf.contents.tp_state = 0
+ return sbuf
+
+ return f
+
+
+def send_mock_generator(msg_state):
+ """
+ generates a mock function that can be used to monkeypatch rmr_send_msg
+ usage example:
+ monkeypatch.setattr('ricxappframe.rmr.rmr.rmr_send_msg', rmr_mocks.send_mock_generator(0))
+ """
+
+ def f(_unused, sbuf):
+ sbuf.contents.state = msg_state
+ if msg_state != 0: # set something in transport state if 'error'
+ sbuf.contents.tp_state = 99
+ else:
+ sbuf.contents.tp_state = 0
+ return sbuf
+
+ return f
+
+
+class _Sbuf_Contents:
+ """fake version of how pointers work (ctype pointer access is done by accessing a magical attrivute called "contents"""
+
+ def __init__(self):
+ self.state = 0
+ self.mtype = 0
+ self.len = 0
+ self.payload = ""
+ self.xaction = uuid.uuid1().hex.encode("utf-8")
+ self.sub_id = 0
+ self.tp_state = 0
+ self.meid = None
+
+ def __str__(self):
+ return str(
+ {
+ "state": self.state,
+ "mtype": self.mtype,
+ "len": self.len,
+ "payload": self.payload,
+ "xaction": self.xaction,
+ "sub_id": self.sub_id,
+ "tp_state": self.tp_state,
+ "meid": self.meid,
+ }
+ )
+
+
+class Rmr_mbuf_t:
+ """fake version of rmr.rmr_mbuf_t"""
+
+ def __init__(self):
+ self.contents = _Sbuf_Contents()
+
+
+def patch_rmr(monkeypatch):
+ """
+ Patch rmr; requires a monkeypatch (pytest) object to be passed in
+ """
+
+ def fake_alloc(
+ _vctx, _sz, payload=None, gen_transaction_id=False, mtype=None, meid=None, sub_id=None, fixed_transaction_id=None
+ ):
+ sbuf = Rmr_mbuf_t()
+ if payload:
+ sbuf.contents.payload = payload
+
+ if fixed_transaction_id:
+ sbuf.contents.xaction = fixed_transaction_id
+ elif gen_transaction_id:
+ sbuf.contents.xaction = uuid.uuid1().hex.encode("utf-8")
+
+ if mtype:
+ sbuf.contents.mtype = mtype
+
+ if meid:
+ sbuf.contents.meid = meid
+
+ if sub_id:
+ sbuf.contents.sub_id = sub_id
+
+ return sbuf
+
+ def fake_set_payload_and_length(payload, sbuf):
+ sbuf.contents.payload = payload
+ sbuf.contents.len = len(payload)
+
+ def fake_generate_and_set_transaction_id(sbuf):
+ sbuf.contents.xaction = uuid.uuid1().hex.encode("utf-8")
+
+ def fake_get_payload(sbuf):
+ return sbuf.contents.payload
+
+ def fake_get_meid(sbuf):
+ return sbuf.contents.meid
+
+ def fake_get_src(_sbuf):
+ return "localtest:80" # this is not a part of rmr_mbuf_t
+
+ def fake_rmr_payload_size(_sbuf):
+ return 4096
+
+ def fake_free(_sbuf):
+ pass
+
+ monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_free_msg", fake_free)
+ monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_alloc_msg", fake_alloc)
+ monkeypatch.setattr("ricxappframe.rmr.rmr.set_payload_and_length", fake_set_payload_and_length)
+ monkeypatch.setattr("ricxappframe.rmr.rmr.generate_and_set_transaction_id", fake_generate_and_set_transaction_id)
+ monkeypatch.setattr("ricxappframe.rmr.rmr.get_payload", fake_get_payload)
+ monkeypatch.setattr("ricxappframe.rmr.rmr.get_src", fake_get_src)
+ monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_get_meid", fake_get_meid)
+ monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_payload_size", fake_rmr_payload_size)
from threading import Thread
from ricxappframe import xapp_rmr
from ricxappframe.xapp_sdl import SDLWrapper
-from rmr import rmr
+from ricxappframe.rmr import rmr
from mdclogpy import Logger
# constants
import queue
from threading import Thread
from mdclogpy import Logger
-from rmr import rmr, helpers
+from ricxappframe.rmr import rmr, helpers
mdc_logger = Logger(name=__name__)
setup(
name="ricxappframe",
- version="0.7.0",
+ version="1.0.0",
packages=find_packages(exclude=["tests.*", "tests"]),
- author="Tommy Carpenter",
- description="Xapp framework for python",
+ author="Tommy Carpenter, E. Scott Daniels",
+ description="Xapp and rmr framework for python",
url="https://gerrit.o-ran-sc.org/r/admin/repos/ric-plt/xapp-frame-py",
- install_requires=["msgpack", "rmr>=4.0.0, <5.0.0", "mdclogpy", "ricsdl>=2.0.3,<3.0.0"],
+ install_requires=["msgpack", "mdclogpy", "ricsdl>=2.0.3,<3.0.0"],
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Telecommunications Industry",
--- /dev/null
+# vim: ts=4 sw=4 expandtab:
+# ==================================================================================
+# Copyright (c) 2019 Nokia
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+import pytest
+
+
+# These are here just to reduce the size of the code in test_rmr so those (important) tests are more readable; in theory these dicts could be large
+# The actual value of the constants should be ignored by the tests; all we should care
+# about is that the constant value was returned by the RMR function. Further, we should
+# not consider it an error if RMR returns more than what is listed here; these are the
+# list of what is/could be used by this package.
+@pytest.fixture
+def expected_constants():
+ return {
+ "RMR_MAX_XID": 32,
+ "RMR_MAX_SID": 32,
+ "RMR_MAX_MEID": 32,
+ "RMR_MAX_SRC": 64,
+ "RMR_MAX_RCV_BYTES": 4096,
+ "RMRFL_NONE": 0,
+ "RMRFL_MTCALL": 2, # can't be added here until jenkins version >= 1.8.3
+ "RMRFL_AUTO_ALLOC": 3,
+ "RMR_DEF_SIZE": 0,
+ "RMR_VOID_MSGTYPE": -1,
+ "RMR_VOID_SUBID": -1,
+ "RMR_OK": 0,
+ "RMR_ERR_BADARG": 1,
+ "RMR_ERR_NOENDPT": 2,
+ "RMR_ERR_EMPTY": 3,
+ "RMR_ERR_NOHDR": 4,
+ "RMR_ERR_SENDFAILED": 5,
+ "RMR_ERR_CALLFAILED": 6,
+ "RMR_ERR_NOWHOPEN": 7,
+ "RMR_ERR_WHID": 8,
+ "RMR_ERR_OVERFLOW": 9,
+ "RMR_ERR_RETRY": 10,
+ "RMR_ERR_RCVFAILED": 11,
+ "RMR_ERR_TIMEOUT": 12,
+ "RMR_ERR_UNSET": 13,
+ "RMR_ERR_TRUNC": 14,
+ "RMR_ERR_INITFAILED": 15,
+ }
+
+
+@pytest.fixture
+def expected_states():
+ return {
+ 0: "RMR_OK",
+ 1: "RMR_ERR_BADARG",
+ 2: "RMR_ERR_NOENDPT",
+ 3: "RMR_ERR_EMPTY",
+ 4: "RMR_ERR_NOHDR",
+ 5: "RMR_ERR_SENDFAILED",
+ 6: "RMR_ERR_CALLFAILED",
+ 7: "RMR_ERR_NOWHOPEN",
+ 8: "RMR_ERR_WHID",
+ 9: "RMR_ERR_OVERFLOW",
+ 10: "RMR_ERR_RETRY",
+ 11: "RMR_ERR_RCVFAILED",
+ 12: "RMR_ERR_TIMEOUT",
+ 13: "RMR_ERR_UNSET",
+ 14: "RMR_ERR_TRUNC",
+ 15: "RMR_ERR_INITFAILED",
+ }
# do NOT use localhost, seems unresolved on jenkins VMs
+# first 3 lines are used for xapp frame tests
+# last four lines are used in the rmr submodule
newrt|start
mse| 60000 | -1 | 127.0.0.1:4564
mse| 60001 | -1 | 127.0.0.1:4564
mse| 100 | -1 | 127.0.0.1:4564
+mse| 0 | -1 | 127.0.0.1:3563
+mse| 46656 | 777 | 127.0.0.1:3563
+mse| 1 | -1 | 127.0.0.1:3564
+mse| 2 | -1 | 127.0.0.1:3564
newrt|end
# ==================================================================================
import time
import pytest
-from rmr.exceptions import InitFailed
+from ricxappframe.rmr.exceptions import InitFailed
from ricxappframe.xapp_frame import Xapp, RMRXapp
--- /dev/null
+# vim: ts=4 sw=4 expandtab:
+# =================================================================================2
+# Copyright (c) 2019-2020 Nokia
+# Copyright (c) 2018-2020 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+import time
+import pytest
+from ricxappframe.rmr import rmr, helpers, exceptions
+
+
+SIZE = 256
+MRC_SEND = None
+MRC_RCV = None
+
+
+def setup_module():
+ """
+ test_rmr module setup
+ """
+ global MRC_SEND
+ MRC_SEND = rmr.rmr_init(b"3562", 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"3563", rmr.RMR_MAX_RCV_BYTES, 0x00)
+ while rmr.rmr_ready(MRC_RCV) == 0:
+ time.sleep(1)
+
+ global MRC_BUF_RCV
+ MRC_BUF_RCV = rmr.rmr_init(b"3564", rmr.RMR_MAX_RCV_BYTES, 0x02)
+ while rmr.rmr_ready(MRC_BUF_RCV) == 0:
+ time.sleep(1)
+
+
+def teardown_module():
+ """
+ test rmr module teardown
+ """
+ rmr.rmr_close(MRC_SEND)
+ rmr.rmr_close(MRC_RCV)
+
+
+def _assert_new_sbuf(sbuf):
+ """
+ verify the initial state of an alloced message is what we expect
+ """
+ summary = rmr.message_summary(sbuf)
+ assert summary["payload"] == b""
+ assert summary["payload length"] == 0
+ assert summary["subscription id"] == -1
+ assert summary["transaction id"] == b""
+ assert summary["message state"] == 0
+ assert summary["message status"] == "RMR_OK"
+ assert summary["meid"] == b""
+ assert summary["errno"] == 0
+
+
+def test_get_constants(expected_constants):
+ """
+ test getting constants. We don't care what values are returned as those
+ should be meaningful only to RMR. We do care that all of the constants
+ which are defined in expected_contents are returned. Further, we don't
+ consider it to be an error if the returned list has more constants than
+ what are in our list.
+
+ To avoid frustration, this should list all missing keys, not fail on the
+ first missing key.
+ """
+ errors = 0
+ econst = expected_constants
+ rconst = rmr._get_constants()
+ for key in econst: # test all expected constants
+ if key not in rconst: # expected value not listed by rmr
+ errors += 1
+ print("did not find required constant in list from RMR: %s" % key)
+
+ assert errors == 0
+
+
+def test_get_mapping_dict(expected_states):
+ """
+ test getting mapping string
+ """
+ assert rmr._get_mapping_dict() == expected_states
+ assert rmr._state_to_status(0) == "RMR_OK"
+ assert rmr._state_to_status(12) == "RMR_ERR_TIMEOUT"
+ assert rmr._state_to_status(666) == "UNKNOWN STATE"
+
+
+def test_meid():
+ """
+ test meid stringification
+ """
+ sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
+
+ rmr.rmr_set_meid(sbuf, b"\x01\x02")
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"\x01\x02"
+ assert len(rmr.rmr_get_meid(sbuf)) == 2
+
+ rmr.rmr_set_meid(sbuf, b"\x00" * 31)
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"" # NULL bytes get truncated
+
+ rmr.rmr_set_meid(sbuf, b"6" * 31)
+ assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"6" * 31 # string in string out
+
+ rmr.rmr_set_meid(sbuf, b"\x01\x02")
+ assert (
+ rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)["meid"] == b"\x01\x02"
+ ) # Ctypes will chop at first nil, so expect only 2 bytes back
+
+ assert len(rmr.rmr_get_meid(sbuf)) == 2
+
+ # test that an exception is raised for buffers which are too long
+ with pytest.raises(exceptions.MeidSizeOutOfRange):
+ rmr.rmr_set_meid(sbuf, b"8" * 32)
+
+
+def test_rmr_set_get():
+ """
+ test set functions
+ """
+ sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
+ _assert_new_sbuf(sbuf)
+
+ # test payload
+ pay = b"\x01\x00\x80"
+ rmr.set_payload_and_length(pay, sbuf)
+ summary = rmr.message_summary(sbuf)
+ assert summary["payload"] == pay
+ assert summary["payload length"] == 3
+
+ # test transid (note we cant test payload because it's randomly gen)
+ assert summary["transaction id"] == b""
+ assert len(summary["transaction id"]) == 0
+ rmr.generate_and_set_transaction_id(sbuf)
+ summary = rmr.message_summary(sbuf)
+ assert summary["transaction id"] != b""
+ assert len(summary["transaction id"]) == 32
+
+ # test meid
+ assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b""
+ rmr.rmr_set_meid(sbuf, b"666\x01\x00\x01")
+ summary = rmr.message_summary(sbuf)
+ assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b"666\x01"
+ assert (len(summary["meid"])) == 4
+
+
+def test_alloc_fancy():
+ """test allocation with setting payload, trans, mtype, subid"""
+ pay = b"yoo\x01\x00\x80"
+ sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE, payload=pay, gen_transaction_id=True, mtype=14, meid=b"asdf", sub_id=654321)
+ summary = rmr.message_summary(sbuf)
+ assert summary["payload"] == pay
+ assert summary["payload length"] == 6
+ assert summary["transaction id"] != b"" # hard to test what it will be, but make sure not empty
+ assert len(summary["transaction id"]) == 32
+ assert summary["message state"] == 0
+ assert summary["message type"] == sbuf.contents.mtype == 14
+ assert rmr.rmr_get_meid(sbuf) == summary["meid"] == b"asdf"
+ assert sbuf.contents.sub_id == summary["subscription id"] == 654321
+
+
+def test_alloc_overlapping_flags():
+ """test allocation with setting the transaction id"""
+ sbuf = rmr.rmr_alloc_msg(MRC_SEND, SIZE, gen_transaction_id=True, fixed_transaction_id=b"6" * 32)
+ summary = rmr.message_summary(sbuf)
+ assert summary["transaction id"] == b"66666666666666666666666666666666"
+
+
+def test_rcv_timeout():
+ """
+ test torcv; this is a scary test because if it fails... it doesn't fail, it will run forever!
+ We receive 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)
+ assert send_summary["message state"] == 0 # if send fails don't attempt receive
+ assert send_summary["message status"] == "RMR_OK"
+ time.sleep(0.5)
+
+ # 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["message state"] == 0
+ assert rcv_summary["message status"] == "RMR_OK"
+ assert rcv_summary["message type"] == 0
+ assert rcv_summary["payload"] == pay
+
+ # send an ACK back
+ ack_pay = b"message received"
+ sbuf_rcv = rmr.rmr_rts_msg(MRC_RCV, sbuf_rcv, payload=ack_pay, mtype=6666)
+ rcv_ack_summary = rmr.message_summary(sbuf_rcv)
+
+ # have the sender receive it
+ sbuf_send = rmr.rmr_torcv_msg(MRC_SEND, sbuf_send, 2000)
+ send_ack_summary = rmr.message_summary(sbuf_send)
+
+ 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"
+ assert send_ack_summary["payload"] == ack_pay
+ assert send_ack_summary["message type"] == 6666
+
+
+def test_send_rcv_subid_good():
+ """
+ test send and receive where subid is used for routing
+ """
+ pay = b"\x01\x00\x80"
+ test_mtype = 46656
+ test_subid = 777
+
+ # send a message
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, pay, mtype=test_mtype, sub_id=test_subid)
+ pre_send_summary = rmr.message_summary(sbuf_send)
+ sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
+ send_summary = rmr.message_summary(sbuf_send)
+
+ # receive it in other context
+ time.sleep(0.5)
+ sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, 3)
+ sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
+ rcv_summary = rmr.message_summary(sbuf_rcv)
+
+ # asserts
+ assert send_summary["message state"] == rcv_summary["message state"] == 0
+ assert send_summary["message status"] == rcv_summary["message status"] == "RMR_OK"
+ assert pre_send_summary["payload"] == rcv_summary["payload"] == pay
+ assert pre_send_summary["message type"] == rcv_summary["message type"] == test_mtype
+ assert pre_send_summary["subscription id"] == rcv_summary["subscription id"] == test_subid
+
+
+def test_send_rcv_subid_bad_subid():
+ """
+ test send and receive where subid is used for routing but nobody recieves this subid
+ """
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, b"\x01\x00\x80", mtype=46656, sub_id=778)
+ sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
+ assert rmr.message_summary(sbuf_send)["message state"] == 2
+ assert rmr.message_summary(sbuf_send)["message status"] == "RMR_ERR_NOENDPT"
+
+
+def test_send_rcv_subid_bad_mtype():
+ """
+ test send and receive where subid is used for routing but nobody recieves this mtype
+ """
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, 3, b"\x01\x00\x80", mtype=46657, sub_id=777)
+ sbuf_send = rmr.rmr_send_msg(MRC_SEND, sbuf_send)
+ assert rmr.message_summary(sbuf_send)["message state"] == 2
+ assert rmr.message_summary(sbuf_send)["message status"] == "RMR_ERR_NOENDPT"
+
+
+def send_burst(mrc, fmt, mtype=1, num=13, counter=0):
+ """
+ Internal function to support test_rcv_all.
+ Send a burst of messages optionally giving the type, payload
+ and number to send.
+ """
+ sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE) # seed message buffer
+
+ for i in range(num):
+ payload = bytes(fmt % counter, "UTF-8")
+ counter += 1
+
+ rmr.set_payload_and_length(payload, sbuf_send)
+ sbuf_send.contents.mtype = mtype
+
+ max_retries = 5
+ while max_retries > 0:
+ sbuf_send = rmr.rmr_send_msg(mrc, sbuf_send)
+ ms = rmr.message_summary(sbuf_send)
+ if ms["message state"] != 10: # 10 is retry
+ break
+ max_retries -= 1
+ time.sleep(0.75)
+
+ assert ms["message state"] == 0
+ assert max_retries > 0
+
+
+def test_rcv_all():
+ """
+ test the ability to receive a batch of queued messages.
+ """
+ pay_fmt = "send to ring msg: %d" # dynamic message format with counter
+
+ send_burst(MRC_SEND, pay_fmt) # send a bunch of 13 messages that should queue
+ time.sleep(1) # ensure underlying transport gets cycles to send/receive
+
+ bundle = helpers.rmr_rcvall_msgs(MRC_BUF_RCV) # use the buffered receiver to read all with a single call
+ assert len(bundle) == 13
+
+ for i, ms in enumerate(bundle):
+ ms = bundle[i] # validate each summary returned, and ordering preserved
+ assert ms["message state"] == 0
+ expected_pay = bytes(pay_fmt % i, "UTF-8")
+ assert ms["payload"] == expected_pay
+
+ send_burst(MRC_SEND, pay_fmt, mtype=1, num=10) # send a second round with msg types 1 and 2 to test filter
+ send_burst(MRC_SEND, pay_fmt, mtype=2, num=8)
+ send_burst(MRC_SEND, pay_fmt, mtype=1, num=5)
+ send_burst(MRC_SEND, pay_fmt, mtype=2, num=4, counter=8) # total of 12 messages with type 2 should be queued
+ time.sleep(1) # ensure underlying transport gets cycles to send/receive
+
+ bundle = helpers.rmr_rcvall_msgs_raw(MRC_BUF_RCV, [2]) # receive only message type 2 messages
+ assert len(bundle) == 12 # we should only get the second batch of 12 messages
+
+ for i, (ms, sbuf) in enumerate(bundle): # test the raw version
+ test_summary = rmr.message_summary(sbuf)
+ assert test_summary == ms
+ assert ms["message state"] == 0 # all should be OK
+ assert ms["message type"] == 2 # only mtype 2 should have been received
+ expected_pay = bytes(pay_fmt % i, "UTF-8") # ordering should still jive with the counter
+ assert ms["payload"] == expected_pay
+ rmr.rmr_free_msg(sbuf)
+
+
+def test_bad_buffer():
+ """test that we get a proper exception when the buffer has a null pointer"""
+ with pytest.raises(exceptions.BadBufferAllocation):
+ rmr.rmr_alloc_msg(None, 4096)
+
+
+def test_resize_payload():
+ """test the ability to insert a larger payload into an existing message"""
+ mtype = 99
+ subid = 100
+
+ mbuf = rmr.rmr_alloc_msg(MRC_SEND, 25) # allocate buffer with small payload
+ mbuf.contents.mtype = mtype # type and sub-id should not change
+ mbuf.contents.sub_id = subid
+
+ long_payload = b"This is a long payload that should force the message buffer to be reallocated"
+ rmr.set_payload_and_length(long_payload, mbuf)
+ summary = rmr.message_summary(mbuf)
+ assert summary["payload max size"] >= len(long_payload) # RMR may allocate a larger payload space
+ assert summary["payload length"] == len(long_payload) # however, the length must be exactly the same
+ assert summary["message type"] == mtype # both mtype and sub-id should be preserved in new
+ assert summary["subscription id"] == subid
--- /dev/null
+# ==================================================================================
+# Copyright (c) 2019 Nokia
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+from ricxappframe.rmr import rmr
+from ricxappframe.rmr.rmr_mocks import rmr_mocks
+
+
+MRC = None
+SIZE = 256
+
+
+def _partial_dict_comparison(subset_dict, target_dict):
+ """
+ Compares that target_dict[k] == subset_dict[k] for all k <- subset_dict
+ """
+ for k, v in subset_dict.items():
+ assert k in target_dict
+ assert target_dict[k] == subset_dict[k]
+
+
+def test_send_mock(monkeypatch):
+ """
+ tests the send mock
+ """
+ monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(12))
+ rmr_mocks.patch_rmr(monkeypatch)
+ sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
+ rmr.set_payload_and_length("testttt".encode("utf8"), sbuf)
+
+ expected = {
+ "meid": None,
+ "message source": "localtest:80",
+ "message state": 0,
+ "message type": 0,
+ "message status": "RMR_OK",
+ "payload": b"testttt",
+ "payload length": 7,
+ "payload max size": 4096,
+ "subscription id": 0,
+ }
+ _partial_dict_comparison(expected, rmr.message_summary(sbuf))
+
+ # set the mtype
+ sbuf.contents.mtype = 666
+
+ # send it (the fake send sets the state, and touches nothing else)
+ sbuf = rmr.rmr_send_msg(MRC, sbuf)
+
+ expected = {
+ "meid": None,
+ "message source": "localtest:80",
+ "message state": 12,
+ "message type": 666,
+ "message status": "RMR_ERR_TIMEOUT",
+ "payload": None,
+ "payload length": 7,
+ "payload max size": 4096,
+ "subscription id": 0,
+ }
+ _partial_dict_comparison(expected, rmr.message_summary(sbuf))
+
+
+def test_rcv_mock(monkeypatch):
+ """
+ tests the rmr recieve mocking generator
+ """
+ rmr_mocks.patch_rmr(monkeypatch)
+ sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
+
+ # test rcv
+ monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_rcv_msg", rmr_mocks.rcv_mock_generator({"foo": "bar"}, 666, 0, True))
+ sbuf = rmr.rmr_rcv_msg(MRC, sbuf)
+ assert rmr.get_payload(sbuf) == b'{"foo": "bar"}'
+ assert sbuf.contents.mtype == 666
+ assert sbuf.contents.state == 0
+ assert sbuf.contents.len == 14
+
+ # test torcv, although the timeout portion is not currently mocked or tested
+ monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_torcv_msg", rmr_mocks.rcv_mock_generator({"foo": "bar"}, 666, 0, True, 50))
+ sbuf = rmr.rmr_torcv_msg(MRC, sbuf, 5)
+ assert rmr.get_payload(sbuf) == b'{"foo": "bar"}'
+ assert sbuf.contents.mtype == 666
+ assert sbuf.contents.state == 0
+ assert sbuf.contents.len == 14
+
+
+def test_alloc(monkeypatch):
+ """
+ test alloc with all fields set
+ """
+ rmr_mocks.patch_rmr(monkeypatch)
+ sbuf = rmr.rmr_alloc_msg(
+ MRC, SIZE, payload=b"foo", gen_transaction_id=True, mtype=5, meid=b"mee", sub_id=234, fixed_transaction_id=b"t" * 32
+ )
+ summary = rmr.message_summary(sbuf)
+ assert summary["payload"] == b"foo"
+ assert summary["transaction id"] == b"t" * 32
+ assert summary["message type"] == 5
+ assert summary["meid"] == b"mee"
+ assert summary["subscription id"] == 234
Code Review / ric-plt / xapp-frame-py.git / commitdiff