1 # ==================================================================================
2 # Copyright (c) 2019 Nokia
3 # Copyright (c) 2018-2019 AT&T Intellectual Property.
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
9 # http://www.apache.org/licenses/LICENSE-2.0
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 # ==================================================================================
19 Provides mocks that are useful for end applications unit testing
26 def rcv_mock_generator(msg_payload, msg_type, msg_state, jsonb, timeout=0):
28 generates a mock function that can be used to monkeypatch rmr_torcv_msg or rmr_rcv_msg
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
40 sbuf.contents.tp_state = 0
46 def send_mock_generator(msg_state):
48 generates a mock function that can be used to monkeypatch rmr_send_msg
50 monkeypatch.setattr('rmr.rmr.rmr_send_msg', rmr_mocks.send_mock_generator(0))
54 sbuf.contents.state = msg_state
55 if msg_state != 0: # set something in transport state if 'error'
56 sbuf.contents.tp_state = 99
58 sbuf.contents.tp_state = 0
65 """fake version of how pointers work (ctype pointer access is done by accessing a magical attrivute called "contents"""
72 self.xaction = uuid.uuid1().hex.encode("utf-8")
82 "payload": self.payload,
83 "xaction": self.xaction,
84 "sub_id": self.sub_id,
85 "tp_state": self.tp_state,
91 """fake version of rmr.rmr_mbuf_t"""
94 self.contents = _Sbuf_Contents()
97 def patch_rmr(monkeypatch):
99 Patch rmr; requires a monkeypatch (pytest) object to be passed in
102 def fake_alloc(_unused, _alsounused):
105 def fake_set_payload_and_length(payload, sbuf):
106 sbuf.contents.payload = payload
107 sbuf.contents.len = len(payload)
109 def fake_generate_and_set_transaction_id(sbuf):
110 sbuf.contents.xaction = uuid.uuid1().hex.encode("utf-8")
112 def fake_get_payload(sbuf):
113 return sbuf.contents.payload
115 def fake_get_meid(_sbuf):
116 return None # this is not a part of rmr_mbuf_t
118 def fake_get_src(_sbuf):
119 return "localtest:80" # this is not a part of rmr_mbuf_t
121 def fake_rmr_payload_size(_sbuf):
124 monkeypatch.setattr("rmr.rmr.rmr_alloc_msg", fake_alloc)
125 monkeypatch.setattr("rmr.rmr.set_payload_and_length", fake_set_payload_and_length)
126 monkeypatch.setattr("rmr.rmr.generate_and_set_transaction_id", fake_generate_and_set_transaction_id)
127 monkeypatch.setattr("rmr.rmr.get_payload", fake_get_payload)
128 monkeypatch.setattr("rmr.rmr.get_src", fake_get_src)
129 monkeypatch.setattr("rmr.rmr.rmr_get_meid", fake_get_meid)
130 monkeypatch.setattr("rmr.rmr.rmr_payload_size", fake_rmr_payload_size)