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