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")
83 "payload": self.payload,
84 "xaction": self.xaction,
85 "sub_id": self.sub_id,
86 "tp_state": self.tp_state,
93 """fake version of rmr.rmr_mbuf_t"""
96 self.contents = _Sbuf_Contents()
99 def patch_rmr(monkeypatch):
101 Patch rmr; requires a monkeypatch (pytest) object to be passed in
105 _vctx, _sz, payload=None, gen_transaction_id=False, mtype=None, meid=None, sub_id=None, fixed_transaction_id=None
109 sbuf.contents.payload = payload
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")
117 sbuf.contents.mtype = mtype
120 sbuf.contents.meid = meid
123 sbuf.contents.sub_id = sub_id
127 def fake_set_payload_and_length(payload, sbuf):
128 sbuf.contents.payload = payload
129 sbuf.contents.len = len(payload)
131 def fake_generate_and_set_transaction_id(sbuf):
132 sbuf.contents.xaction = uuid.uuid1().hex.encode("utf-8")
134 def fake_get_payload(sbuf):
135 return sbuf.contents.payload
137 def fake_get_meid(sbuf):
138 return sbuf.contents.meid
140 def fake_get_src(_sbuf):
141 return "localtest:80" # this is not a part of rmr_mbuf_t
143 def fake_rmr_payload_size(_sbuf):
146 def fake_free(_sbuf):
149 monkeypatch.setattr("rmr.rmr.rmr_free_msg", fake_free)
150 monkeypatch.setattr("rmr.rmr.rmr_alloc_msg", fake_alloc)
151 monkeypatch.setattr("rmr.rmr.set_payload_and_length", fake_set_payload_and_length)
152 monkeypatch.setattr("rmr.rmr.generate_and_set_transaction_id", fake_generate_and_set_transaction_id)
153 monkeypatch.setattr("rmr.rmr.get_payload", fake_get_payload)
154 monkeypatch.setattr("rmr.rmr.get_src", fake_get_src)
155 monkeypatch.setattr("rmr.rmr.rmr_get_meid", fake_get_meid)
156 monkeypatch.setattr("rmr.rmr.rmr_payload_size", fake_rmr_payload_size)