ricxappframe/rmr/rmr_mocks/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
  18 """
  19 Provides mocks that are useful for end applications unit testing
  20 """
  21
  22 import json
  23 import uuid
  24 from ricxappframe.rmr import rmr
  25
  26
  27 def rcv_mock_generator(msg_payload, msg_type, msg_state, jsonb, timeout=0):
  28     """
  29     generates a mock function that can be used to monkeypatch rmr_torcv_msg or rmr_rcv_msg
  30     """
  31
  32     def f(_mrc, sbuf, _timeout=timeout):  # last param is needed for calls to rmr_torcv_msg, but not in rmr_rcv_msg
  33         sbuf.contents.mtype = msg_type
  34         payload = json.dumps(msg_payload).encode("utf-8") if jsonb else msg_payload
  35         sbuf.contents.payload = payload
  36         sbuf.contents.len = len(payload)
  37         sbuf.contents.state = msg_state
  38         if msg_state != 0:  # set something in transport state if 'error'
  39             sbuf.contents.tp_state = 99
  40         else:
  41             sbuf.contents.tp_state = 0
  42         return sbuf
  43
  44     return f
  45
  46
  47 def send_mock_generator(msg_state):
  48     """
  49     generates a mock function that can be used to monkeypatch rmr_send_msg
  50     usage example:
  51         monkeypatch.setattr('ricxappframe.rmr.rmr.rmr_send_msg', rmr_mocks.send_mock_generator(0))
  52     """
  53
  54     def f(_unused, sbuf):
  55         sbuf.contents.state = msg_state
  56         if msg_state != 0:  # set something in transport state if 'error'
  57             sbuf.contents.tp_state = 99
  58         else:
  59             sbuf.contents.tp_state = 0
  60         return sbuf
  61
  62     return f
  63
  64
  65 class _Sbuf_Contents:
  66     """fake version of how pointers work (ctype pointer access is done by accessing a magical attribute called "contents"""
  67
  68     def __init__(self):
  69         self.state = 0
  70         self.mtype = 0
  71         self.len = 0
  72         self.payload = ""
  73         self.xaction = uuid.uuid1().hex.encode("utf-8")
  74         self.sub_id = 0
  75         self.tp_state = 0
  76         self.meid = None
  77
  78     def __str__(self):
  79         return str(
  80             {
  81                 rmr.RMR_MS_MSG_STATE: self.state,
  82                 rmr.RMR_MS_MSG_TYPE: self.mtype,
  83                 rmr.RMR_MS_PAYLOAD_LEN: self.len,
  84                 rmr.RMR_MS_PAYLOAD: self.payload,
  85                 rmr.RMR_MS_TRN_ID: self.xaction,
  86                 rmr.RMR_MS_SUB_ID: self.sub_id,
  87                 rmr.RMR_MS_ERRNO: self.tp_state,
  88                 rmr.RMR_MS_MEID: self.meid,
  89             }
  90         )
  91
  92
  93 class Rmr_mbuf_t:
  94     """fake version of rmr.rmr_mbuf_t"""
  95
  96     def __init__(self):
  97         self.contents = _Sbuf_Contents()
  98
  99
 100 def patch_rmr(monkeypatch):
 101     """
 102     Patch rmr; requires a monkeypatch (pytest) object to be passed in
 103     """
 104
 105     def fake_alloc(
 106         _vctx, _sz, payload=None, gen_transaction_id=False, mtype=None, meid=None, sub_id=None, fixed_transaction_id=None
 107     ):
 108         sbuf = Rmr_mbuf_t()
 109         if payload:
 110             sbuf.contents.payload = payload
 111
 112         if fixed_transaction_id:
 113             sbuf.contents.xaction = fixed_transaction_id
 114         elif gen_transaction_id:
 115             sbuf.contents.xaction = uuid.uuid1().hex.encode("utf-8")
 116
 117         if mtype:
 118             sbuf.contents.mtype = mtype
 119
 120         if meid:
 121             sbuf.contents.meid = meid
 122
 123         if sub_id:
 124             sbuf.contents.sub_id = sub_id
 125
 126         return sbuf
 127
 128     def fake_set_payload_and_length(payload, sbuf):
 129         sbuf.contents.payload = payload
 130         sbuf.contents.len = len(payload)
 131
 132     def fake_generate_and_set_transaction_id(sbuf):
 133         sbuf.contents.xaction = uuid.uuid1().hex.encode("utf-8")
 134
 135     def fake_get_payload(sbuf):
 136         return sbuf.contents.payload
 137
 138     def fake_get_meid(sbuf):
 139         return sbuf.contents.meid
 140
 141     def fake_get_src(_sbuf):
 142         return "localtest:80"  # this is not a part of rmr_mbuf_t
 143
 144     def fake_rmr_payload_size(_sbuf):
 145         return 4096
 146
 147     def fake_free(_sbuf):
 148         pass
 149
 150     monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_free_msg", fake_free)
 151     monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_alloc_msg", fake_alloc)
 152     monkeypatch.setattr("ricxappframe.rmr.rmr.set_payload_and_length", fake_set_payload_and_length)
 153     monkeypatch.setattr("ricxappframe.rmr.rmr.generate_and_set_transaction_id", fake_generate_and_set_transaction_id)
 154     monkeypatch.setattr("ricxappframe.rmr.rmr.get_payload", fake_get_payload)
 155     monkeypatch.setattr("ricxappframe.rmr.rmr.get_src", fake_get_src)
 156     monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_get_meid", fake_get_meid)
 157     monkeypatch.setattr("ricxappframe.rmr.rmr.rmr_payload_size", fake_rmr_payload_size)