Add rmr-python

[ric-plt/lib/rmr.git] / src / bindings / rmr-python / rmr / rmr_mocks / rmr_mocks.py

diff --git a/src/bindings/rmr-python/rmr/rmr_mocks/rmr_mocks.py b/src/bindings/rmr-python/rmr/rmr_mocks/rmr_mocks.py
new file mode 100644 (file)
index 0000000..395bfb3
--- /dev/null
+++ b/src/bindings/rmr-python/rmr/rmr_mocks/rmr_mocks.py
@@ -0,0 +1,120 @@
+# ==================================================================================
+#       Copyright (c) 2019 Nokia
+#       Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#          http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+# ==================================================================================
+
+"""
+Provides mocks that are useful for end applications unit testing
+"""
+
+import json
+import uuid
+
+
+def rcv_mock_generator(msg_payload, msg_type, msg_state, jsonb, timeout=0):
+    """
+    generates a mock function that can be used to monkeypatch rmr_torcv_msg or rmr_rcv_msg
+    """
+
+    def f(_mrc, sbuf, _timeout=timeout):  # last param is needed for calls to rmr_torcv_msg, but not in rmr_rcv_msg
+        sbuf.contents.mtype = msg_type
+        payload = json.dumps(msg_payload).encode("utf-8") if jsonb else msg_payload
+        sbuf.contents.payload = payload
+        sbuf.contents.len = len(payload)
+        sbuf.contents.state = msg_state
+        return sbuf
+
+    return f
+
+
+def send_mock_generator(msg_state):
+    """
+    generates a mock function that can be used to monkeypatch rmr_send_msg
+    usage example:
+        monkeypatch.setattr('rmr.rmr.rmr_send_msg', rmr_mocks.send_mock_generator(0))
+    """
+
+    def f(_unused, sbuf):
+        sbuf.contents.state = msg_state
+        return sbuf
+
+    return f
+
+
+class _Sbuf_Contents:
+    """fake version of how pointers work (ctype pointer access is done by accessing a magical attrivute called "contents"""
+
+    def __init__(self):
+        self.state = 0
+        self.mtype = 0
+        self.len = 0
+        self.payload = ""
+        self.xaction = uuid.uuid1().hex.encode("utf-8")
+        self.sub_id = 0
+
+    def __str__(self):
+        return str(
+            {
+                "state": self.state,
+                "mtype": self.mtype,
+                "len": self.len,
+                "payload": self.payload,
+                "xaction": self.xaction,
+                "sub_id": self.sub_id,
+            }
+        )
+
+
+class Rmr_mbuf_t:
+    """fake version of rmr.rmr_mbuf_t"""
+
+    def __init__(self):
+        self.contents = _Sbuf_Contents()
+
+
+def patch_rmr(monkeypatch):
+    """
+    Patch rmr; requires a monkeypatch (pytest) object to be passed in
+    """
+
+    def fake_alloc(_unused, _alsounused):
+        return Rmr_mbuf_t()
+
+    def fake_set_payload_and_length(payload, sbuf):
+        sbuf.contents.payload = payload
+        sbuf.contents.len = len(payload)
+
+    def fake_generate_and_set_transaction_id(sbuf):
+        sbuf.contents.xaction = uuid.uuid1().hex.encode("utf-8")
+
+    def fake_get_payload(sbuf):
+        return sbuf.contents.payload
+
+    def fake_get_meid(_sbuf):
+        return None  # this is not a part of rmr_mbuf_t
+
+    def fake_get_src(_sbuf):
+        return "localtest:80"  # this is not a part of rmr_mbuf_t
+
+    def fake_rmr_payload_size(_sbuf):
+        return 4096
+
+    monkeypatch.setattr("rmr.rmr.rmr_alloc_msg", fake_alloc)
+    monkeypatch.setattr("rmr.rmr.set_payload_and_length", fake_set_payload_and_length)
+    monkeypatch.setattr("rmr.rmr.generate_and_set_transaction_id", fake_generate_and_set_transaction_id)
+    monkeypatch.setattr("rmr.rmr.get_payload", fake_get_payload)
+    monkeypatch.setattr("rmr.rmr.get_src", fake_get_src)
+    monkeypatch.setattr("rmr.rmr.get_meid", fake_get_meid)
+    monkeypatch.setattr("rmr.rmr.rmr_payload_size", fake_rmr_payload_size)