tests/test_rmr.py

# =================================================================================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
MRC_BUF_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, rmr.RMRFL_MTCALL)
    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)
    rmr.rmr_close(MRC_BUF_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[rmr.RMR_MS_PAYLOAD] == b""
    assert summary[rmr.RMR_MS_PAYLOAD_LEN] == 0
    assert summary[rmr.RMR_MS_SUB_ID] == -1
    assert summary[rmr.RMR_MS_TRN_ID] == b""
    assert summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
    assert summary[rmr.RMR_MS_MSG_STATUS] == "RMR_OK"
    assert summary[rmr.RMR_MS_MEID] == b""
    assert summary[rmr.RMR_MS_ERRNO] == 0


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)[rmr.RMR_MS_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)[rmr.RMR_MS_MEID] == b""  # NULL bytes get truncated

    rmr.rmr_set_meid(sbuf, b"6" * 31)
    assert rmr.rmr_get_meid(sbuf) == rmr.message_summary(sbuf)[rmr.RMR_MS_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)[rmr.RMR_MS_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[rmr.RMR_MS_PAYLOAD] == pay
    assert summary[rmr.RMR_MS_PAYLOAD_LEN] == 3

    # test transid (note we cant test payload because it's randomly gen)
    assert summary[rmr.RMR_MS_TRN_ID] == b""
    assert len(summary[rmr.RMR_MS_TRN_ID]) == 0
    rmr.generate_and_set_transaction_id(sbuf)
    summary = rmr.message_summary(sbuf)
    assert summary[rmr.RMR_MS_TRN_ID] != b""
    assert len(summary[rmr.RMR_MS_TRN_ID]) == 32

    # test meid
    assert rmr.rmr_get_meid(sbuf) == summary[rmr.RMR_MS_MEID] == b""
    rmr.rmr_set_meid(sbuf, b"666\x01\x00\x01")
    summary = rmr.message_summary(sbuf)
    assert rmr.rmr_get_meid(sbuf) == summary[rmr.RMR_MS_MEID] == b"666\x01"
    assert (len(summary[rmr.RMR_MS_MEID])) == 4

    # exercise the logging setter; cannot test the result
    rmr.rmr_set_vlevel(0)


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[rmr.RMR_MS_PAYLOAD] == pay
    assert summary[rmr.RMR_MS_PAYLOAD_LEN] == 6
    assert summary[rmr.RMR_MS_TRN_ID] != b""  # hard to test what it will be, but make sure not empty
    assert len(summary[rmr.RMR_MS_TRN_ID]) == 32
    assert summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
    assert summary[rmr.RMR_MS_MSG_TYPE] == sbuf.contents.mtype == 14
    assert rmr.rmr_get_meid(sbuf) == summary[rmr.RMR_MS_MEID] == b"asdf"
    assert sbuf.contents.sub_id == summary[rmr.RMR_MS_SUB_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[rmr.RMR_MS_TRN_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)
    start_rcv_sec = time.time()
    sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 500)  # should wait a bit before returning
    summary = rmr.message_summary(sbuf_rcv)
    assert summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_ERR_TIMEOUT
    assert summary[rmr.RMR_MS_MSG_STATUS] == "RMR_ERR_TIMEOUT"
    assert (time.time() - start_rcv_sec > 0.5)  # test duration should be longer than the 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[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK  # if send fails don't attempt receive
    assert send_summary[rmr.RMR_MS_MSG_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[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
    assert rcv_summary[rmr.RMR_MS_MSG_STATUS] == "RMR_OK"
    assert rcv_summary[rmr.RMR_MS_MSG_TYPE] == 0
    assert rcv_summary[rmr.RMR_MS_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[rmr.RMR_MS_MSG_STATE] == rcv_ack_summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
    assert send_ack_summary[rmr.RMR_MS_MSG_STATUS] == rcv_ack_summary[rmr.RMR_MS_MSG_STATUS] == "RMR_OK"
    assert send_ack_summary[rmr.RMR_MS_PAYLOAD] == ack_pay
    assert send_ack_summary[rmr.RMR_MS_MSG_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[rmr.RMR_MS_MSG_STATE] == rcv_summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
    assert send_summary[rmr.RMR_MS_MSG_STATUS] == rcv_summary[rmr.RMR_MS_MSG_STATUS] == "RMR_OK"
    assert pre_send_summary[rmr.RMR_MS_PAYLOAD] == rcv_summary[rmr.RMR_MS_PAYLOAD] == pay
    assert pre_send_summary[rmr.RMR_MS_MSG_TYPE] == rcv_summary[rmr.RMR_MS_MSG_TYPE] == test_mtype
    assert pre_send_summary[rmr.RMR_MS_SUB_ID] == rcv_summary[rmr.RMR_MS_SUB_ID] == test_subid


def test_send_rcv_subid_bad_subid():
    """
    test send and receive where subid is used for routing but nobody receives 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)[rmr.RMR_MS_MSG_STATE] == rmr.RMR_ERR_NOENDPT
    assert rmr.message_summary(sbuf_send)[rmr.RMR_MS_MSG_STATUS] == "RMR_ERR_NOENDPT"


def test_send_rcv_subid_bad_mtype():
    """
    test send and receive where subid is used for routing but nobody receives 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)[rmr.RMR_MS_MSG_STATE] == rmr.RMR_ERR_NOENDPT
    assert rmr.message_summary(sbuf_send)[rmr.RMR_MS_MSG_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[rmr.RMR_MS_MSG_STATE] != rmr.RMR_ERR_RETRY:
                break
            max_retries -= 1
            time.sleep(0.75)

        assert ms[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
        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[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
        expected_pay = bytes(pay_fmt % i, "UTF-8")
        assert ms[rmr.RMR_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 type 2 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[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK  # all should be OK
        assert ms[rmr.RMR_MS_MSG_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[rmr.RMR_MS_PAYLOAD] == expected_pay
        rmr.rmr_free_msg(sbuf)

    # check the timeout scenarios
    start_rcv_sec = time.time()
    bundle = helpers.rmr_rcvall_msgs(MRC_RCV, timeout=1001)  # non-zero timeout means wait
    assert len(bundle) == 0  # we should get none
    assert (time.time() - start_rcv_sec > 1)  # test duration should be longer than 1 second

    start_rcv_sec = time.time()
    bundle = helpers.rmr_rcvall_msgs_raw(MRC_RCV, timeout=1002)  # non-zero timeout means wait
    assert len(bundle) == 0  # we should get none
    assert (time.time() - start_rcv_sec > 1)  # test duration should be longer than 1 second


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[rmr.RMR_MS_PAYLOAD_MAX] >= len(long_payload)  # RMR may allocate a larger payload space
    assert summary[rmr.RMR_MS_PAYLOAD_LEN] == len(long_payload)  # however, the length must be exactly the same
    assert summary[rmr.RMR_MS_MSG_TYPE] == mtype  # both mtype and sub-id should be preserved in new
    assert summary[rmr.RMR_MS_SUB_ID] == subid


def test_wh():
    """test the ability to send a message directly, without routing, via a wormhole"""
    state = rmr.rmr_wh_state(MRC_SEND, 1)
    assert state != rmr.RMR_OK
    whid = rmr.rmr_wh_open(MRC_SEND, b"127.0.0.1:3563")
    assert whid >= 0
    state = rmr.rmr_wh_state(MRC_SEND, whid)
    assert state == rmr.RMR_OK

    sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
    _assert_new_sbuf(sbuf_send)
    mtype = 1
    sbuf_send.contents.mtype = mtype
    payload = b"Birds like worms"
    rmr.set_payload_and_length(payload, sbuf_send)
    send_summary = rmr.message_summary(sbuf_send)

    # send via call, but don't wait long for a response
    rmr.rmr_wh_call(MRC_SEND, whid, sbuf_send, 1, 100)

    # receive message in other context
    time.sleep(0.5)
    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)

    # asserts
    assert send_summary[rmr.RMR_MS_MSG_STATE] == rcv_summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
    assert send_summary[rmr.RMR_MS_MSG_TYPE] == rcv_summary[rmr.RMR_MS_MSG_TYPE] == mtype
    assert send_summary[rmr.RMR_MS_PAYLOAD] == rcv_summary[rmr.RMR_MS_PAYLOAD] == payload

    sbuf_send = rmr.rmr_alloc_msg(MRC_SEND, SIZE)
    _assert_new_sbuf(sbuf_send)
    mtype = 1
    sbuf_send.contents.mtype = mtype
    payload = b"Birds like worms"
    rmr.set_payload_and_length(payload, sbuf_send)

    # send without waiting for a response
    rmr.rmr_wh_send_msg(MRC_SEND, whid, sbuf_send)

    # receive message in other context
    time.sleep(0.5)
    sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
    rcv_summary = rmr.message_summary(sbuf_rcv)

    # asserts
    assert send_summary[rmr.RMR_MS_MSG_STATE] == rcv_summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
    assert send_summary[rmr.RMR_MS_MSG_TYPE] == rcv_summary[rmr.RMR_MS_MSG_TYPE] == mtype
    assert send_summary[rmr.RMR_MS_PAYLOAD] == rcv_summary[rmr.RMR_MS_PAYLOAD] == payload

    rmr.rmr_wh_close(MRC_SEND, whid)
    # don't check state after close, connection is always reported as open