Changes to framework usage:

[ric-plt/xapp-frame-py.git] / tests / test_xapps.py

diff --git a/tests/test_xapps.py b/tests/test_xapps.py
new file mode 100644 (file)
index 0000000..f80d375
--- /dev/null
+++ b/tests/test_xapps.py
@@ -0,0 +1,89 @@
+# ==================================================================================
+#       Copyright (c) 2020 Nokia
+#       Copyright (c) 2020 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.
+# ==================================================================================
+import json
+import time
+from contextlib import suppress
+from ricxappframe.xapp_frame import Xapp, RMRXapp
+
+rmr_xapp = None
+gen_xapp = None
+
+
+def test_flow():
+
+    # test variables
+    def_called = 0
+    sixty_called = 0
+
+    # create rmr app
+    global rmr_xapp
+
+    def post_init(self):
+        print("hey")
+
+    def default_handler(self, summary, sbuf):
+        nonlocal def_called
+        def_called += 1
+        assert json.loads(summary["payload"]) == {"test send 60001": 1}
+        self.rmr_free(sbuf)
+
+    rmr_xapp = RMRXapp(default_handler, post_init=post_init, rmr_port=4564, rmr_wait_for_ready=False, use_fake_sdl=True)
+
+    def sixtythou_handler(self, summary, sbuf):
+        nonlocal sixty_called
+        sixty_called += 1
+        assert json.loads(summary["payload"]) == {"test send 60000": 1}
+        self.rmr_free(sbuf)
+
+    rmr_xapp.register_callback(sixtythou_handler, 60000)
+    rmr_xapp.run()
+
+    time.sleep(1)
+
+    def entry(self):
+
+        self.sdl_set("testns", "mykey", 6)
+        assert self.sdl_get("testns", "mykey") == 6
+        assert self.sdl_find_and_get("testns", "myk") == {"mykey": 6}
+        assert self.healthcheck()
+
+        val = json.dumps({"test send 60000": 1}).encode()
+        self.rmr_send(val, 60000)
+
+        val = json.dumps({"test send 60001": 2}).encode()
+        self.rmr_send(val, 60001)
+
+    global gen_xapp
+    gen_xapp = Xapp(entrypoint=entry, rmr_wait_for_ready=False, use_fake_sdl=True)
+    gen_xapp.run()
+
+    time.sleep(1)
+
+    assert def_called == 1
+    assert sixty_called == 1
+
+
+def teardown_module():
+    """
+    this is like a "finally"; the name of this function is pytest magic
+    safer to put down here since certain failures above can lead to pytest never returning
+    for example if an exception gets raised before stop is called in any test function above, pytest will hang forever
+    """
+    with suppress(Exception):
+        gen_xapp.stop()
+    with suppress(Exception):
+        rmr_xapp.stop()