tests/test_rmr_mocks.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 from ricxappframe.rmr import rmr
  18 from ricxappframe.rmr.rmr_mocks import rmr_mocks
  19
  20
  21 MRC = None
  22 SIZE = 256
  23
  24
  25 def _partial_dict_comparison(subset_dict, target_dict):
  26     """
  27     Compares that target_dict[k] == subset_dict[k] for all k <- subset_dict
  28     """
  29     for k, v in subset_dict.items():
  30         assert k in target_dict
  31         assert target_dict[k] == subset_dict[k]
  32
  33
  34 def test_send_mock(monkeypatch):
  35     """
  36     tests the send mock
  37     """
  38     monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(12))
  39     rmr_mocks.patch_rmr(monkeypatch)
  40     sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
  41     rmr.set_payload_and_length("testttt".encode("utf8"), sbuf)
  42
  43     expected = {
  44         rmr.RMR_MS_MEID: None,
  45         rmr.RMR_MS_MSG_SOURCE: "localtest:80",
  46         rmr.RMR_MS_MSG_STATE: rmr.RMR_OK,
  47         rmr.RMR_MS_MSG_STATUS: "RMR_OK",
  48         rmr.RMR_MS_MSG_TYPE: 0,
  49         rmr.RMR_MS_PAYLOAD: b"testttt",
  50         rmr.RMR_MS_PAYLOAD_LEN: 7,
  51         rmr.RMR_MS_PAYLOAD_MAX: 4096,
  52         rmr.RMR_MS_SUB_ID: 0,
  53     }
  54     _partial_dict_comparison(expected, rmr.message_summary(sbuf))
  55
  56     # set the mtype
  57     sbuf.contents.mtype = 666
  58
  59     # send it (the fake send sets the state, and touches nothing else)
  60     sbuf = rmr.rmr_send_msg(MRC, sbuf)
  61
  62     expected = {
  63         rmr.RMR_MS_MEID: None,
  64         rmr.RMR_MS_MSG_SOURCE: "localtest:80",
  65         rmr.RMR_MS_MSG_STATE: rmr.RMR_ERR_TIMEOUT,
  66         rmr.RMR_MS_MSG_STATUS: "RMR_ERR_TIMEOUT",
  67         rmr.RMR_MS_MSG_TYPE: 666,
  68         rmr.RMR_MS_PAYLOAD: None,
  69         rmr.RMR_MS_PAYLOAD_LEN: 7,
  70         rmr.RMR_MS_PAYLOAD_MAX: 4096,
  71         rmr.RMR_MS_SUB_ID: 0,
  72     }
  73     _partial_dict_comparison(expected, rmr.message_summary(sbuf))
  74
  75
  76 def test_rcv_mock(monkeypatch):
  77     """
  78     tests the rmr recieve mocking generator
  79     """
  80     rmr_mocks.patch_rmr(monkeypatch)
  81     sbuf = rmr.rmr_alloc_msg(MRC, SIZE)
  82
  83     # test rcv
  84     monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_rcv_msg", rmr_mocks.rcv_mock_generator({"foo": "bar"}, 666, 0, True))
  85     sbuf = rmr.rmr_rcv_msg(MRC, sbuf)
  86     assert rmr.get_payload(sbuf) == b'{"foo": "bar"}'
  87     assert sbuf.contents.mtype == 666
  88     assert sbuf.contents.state == rmr.RMR_OK
  89     assert sbuf.contents.len == 14
  90
  91     # test torcv, although the timeout portion is not currently mocked or tested
  92     monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_torcv_msg", rmr_mocks.rcv_mock_generator({"foo": "bar"}, 666, 0, True, 50))
  93     sbuf = rmr.rmr_torcv_msg(MRC, sbuf, 5)
  94     assert rmr.get_payload(sbuf) == b'{"foo": "bar"}'
  95     assert sbuf.contents.mtype == 666
  96     assert sbuf.contents.state == rmr.RMR_OK
  97     assert sbuf.contents.len == 14
  98
  99
 100 def test_alloc(monkeypatch):
 101     """
 102     test alloc with all fields set
 103     """
 104     rmr_mocks.patch_rmr(monkeypatch)
 105     sbuf = rmr.rmr_alloc_msg(
 106         MRC, SIZE, payload=b"foo", gen_transaction_id=True, mtype=5, meid=b"mee", sub_id=234, fixed_transaction_id=b"t" * 32
 107     )
 108     summary = rmr.message_summary(sbuf)
 109     assert summary[rmr.RMR_MS_PAYLOAD] == b"foo"
 110     assert summary[rmr.RMR_MS_TRN_ID] == b"t" * 32
 111     assert summary[rmr.RMR_MS_MSG_TYPE] == 5
 112     assert summary[rmr.RMR_MS_MEID] == b"mee"
 113     assert summary[rmr.RMR_MS_SUB_ID] == 234