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 pytest
  18 import json
  19 from rmr import rmr
  20 from rmr.rmr_mocks import rmr_mocks
  21
  22
  23 MRC = None
  24 SIZE = 256
  25
  26
  27 def test_get_mapping_dict(monkeypatch, fake_consts, expected_states):
  28     """
  29     test getting mapping string
  30     """
  31
  32     def fake_rmr_get_consts():
  33         return json.dumps(fake_consts).encode("utf-8")
  34
  35     monkeypatch.setattr("rmr.rmr._rmr_const", fake_rmr_get_consts)
  36     assert rmr._get_mapping_dict() == expected_states
  37     # do again, trigger cache line coverage
  38     assert rmr._get_mapping_dict() == expected_states
  39
  40     assert rmr._state_to_status(0) == "RMR_OK"
  41     assert rmr._state_to_status(12) == "RMR_ERR_TIMEOUT"
  42     assert rmr._state_to_status(666) == "UNKNOWN STATE"
  43
  44
  45 def test_meid_prettify(monkeypatch):
  46     rmr_mocks.patch_rmr(monkeypatch)
  47
  48     # here we re-monkey get_meid
  49     monkeypatch.setattr("rmr.rmr.get_meid", lambda _: "yoooo")
  50     sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
  51     summary = rmr.message_summary(sbuf)
  52     assert summary["meid"] == "yoooo"
  53
  54     # test bytes
  55     monkeypatch.setattr("rmr.rmr.get_meid", lambda _: b"\x01\x00f\x80")
  56     sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
  57     summary = rmr.message_summary(sbuf)
  58     assert summary["meid"] == b"\x01\x00f\x80"
  59
  60     # test the cleanup of null bytes
  61     monkeypatch.setattr(
  62         "rmr.rmr.get_meid",
  63         lambda _: "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
  64     )
  65     sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
  66     summary = rmr.message_summary(sbuf)
  67     assert summary["meid"] == None