src/bindings/rmr-python/tests/test_rmr.py

   1 # ==================================================================================
   2 #       Copyright (c) 2019 Nokia
   3 #       Copyright (c) 2018-2019 AT&T Intellectual Property.
   4 #
   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
   8 #
   9 #          http://www.apache.org/licenses/LICENSE-2.0
  10 #
  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 # ==================================================================================
  17 import time
  18 import pytest
  19 from rmr import rmr
  20
  21
  22 SIZE = 256
  23 MRC = None
  24
  25
  26 def setup_module():
  27     global MRC
  28     MRC = rmr.rmr_init(b"4562", rmr.RMR_MAX_RCV_BYTES, 0x00)
  29     while rmr.rmr_ready(MRC) == 0:
  30         time.sleep(1)
  31
  32
  33 def teardown_module():
  34     rmr.rmr_close(MRC)
  35
  36
  37 def _assert_new_sbuf(sbuf):
  38     """
  39     verify the initial state of an alloced message is what we expect
  40     """
  41     summary = rmr.message_summary(sbuf)
  42     assert summary["payload"] == b""
  43     assert summary["payload length"] == 0
  44     assert summary["transaction id"] == b""
  45     assert summary["message state"] == 0
  46     assert summary["message status"] == "RMR_OK"
  47     assert summary["meid"] is None
  48     assert summary["errno"] == 0
  49
  50
  51 def test_get_constants(expected_constants):
  52     """
  53     test getting constants
  54     """
  55     assert rmr._get_constants() == expected_constants
  56
  57
  58 def test_get_mapping_dict(expected_states):
  59     """
  60     test getting mapping string
  61     """
  62     assert rmr._get_mapping_dict() == expected_states
  63     assert rmr._state_to_status(0) == "RMR_OK"
  64     assert rmr._state_to_status(12) == "RMR_ERR_TIMEOUT"
  65     assert rmr._state_to_status(666) == "UNKNOWN STATE"
  66
  67
  68 def test_meid_prettify():
  69     """
  70     test the printing of meid based on it's value
  71     """
  72     # TODO?? weirdness: setting it takes bytes, but getting it returns a string. This does NOT happen for payload; bytes in, bytes come out.
  73     sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
  74     rmr.rmr_set_meid(sbuf, b"\x00" * 32, 32)
  75     summary = rmr.message_summary(sbuf)
  76     assert summary["meid"] is None  # summary does a pretty print" of 32 null bytes
  77     assert rmr.get_meid(sbuf) == "\x00" * 32  # real underlying value
  78
  79
  80 def test_rmr_set_get():
  81     """
  82     test set functions
  83     """
  84     sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
  85     _assert_new_sbuf(sbuf)
  86
  87     # test payload
  88     pay = b"\x01\x00\x80"
  89     rmr.set_payload_and_length(pay, sbuf)
  90     summary = rmr.message_summary(sbuf)
  91     assert summary["payload"] == pay
  92     assert summary["payload length"] == 3
  93
  94     # test transid (note we cant test payload because it's randomly gen)
  95     assert summary["transaction id"] == b""
  96     assert len(summary["transaction id"]) == 0
  97     rmr.generate_and_set_transaction_id(sbuf)
  98     summary = rmr.message_summary(sbuf)
  99     assert summary["transaction id"] != b""
 100     assert len(summary["transaction id"]) == 32
 101
 102     # test meid
 103     assert rmr.get_meid(sbuf) == "\x00" * 32
 104     assert summary["meid"] is None  # the summary printing function shows the above horridness as None
 105     rmr.rmr_set_meid(sbuf, b"666", 3)
 106     summary = rmr.message_summary(sbuf)
 107     # TODO?? weirdness: setting it takes bytes, but getting it returns a string. This does NOT happen for payload; bytes in, bytes come out.
 108     assert rmr.get_meid(sbuf) == summary["meid"] == "666" + "\x00" * 29