--- /dev/null
+# misc cruft
+*.log
+log.txt
+rmr/*
+docs_and_diagrams/
+
+# documentation
+.tox
+docs/_build/
+
+# standard python ignore template
+.pytest_cache/
+xunit-results.xml
+.DS_Store
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+# Usually these files are written by a python script from a template
+# before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+venv-tox/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*,cover
+.hypothesis/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# IPython Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule
+
+# dotenv
+.env
+
+# virtualenv
+venv/
+ENV/
+
+# Spyder project settings
+.spyderproject
+
+# Rope project settings
+.ropeproject
+
+# Test report
+xunit-reports
+coverage-reports
--- /dev/null
+# Python xapp frame
+
+Running the two examples (adjust for your shell notation)
+
+ set -x LD_LIBRARY_PATH /usr/local/lib/:/usr/local/lib64; set -x RMR_SEED_RT test_route.rt; python pong_xapp.py
+
+ set -x LD_LIBRARY_PATH /usr/local/lib/:/usr/local/lib64; set -x RMR_SEED_RT test_route.rt; python ping_xapp.py
--- /dev/null
+"""
+Test xapp 1
+"""
+# ==================================================================================
+# 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 time
+import json
+from threading import Thread
+from rmr import rmr
+from ricxappframe.xapp_frame import Xapp
+
+# Now we use the framework to echo back the acks
+class MyXapp(Xapp):
+ def loop(self):
+ my_ns = "myxapp"
+ number = 0
+ while True:
+ # test healthcheck
+ print("Healthy? {}".format(xapp.healthcheck()))
+
+ # rmr send
+ val = json.dumps({"test_send": number}).encode()
+ self.rmr_send(val, 60000)
+ number += 1
+
+ # store it in SDL and read it back; delete and read
+ self.sdl_set(my_ns, "numba", number)
+ print((self.sdl_get(my_ns, "numba"), self.sdl_find_and_get(my_ns, "num")))
+ self.sdl_delete(my_ns, "numba")
+ print(self.sdl_get(my_ns, "numba"))
+
+ # rmr receive
+ for (summary, sbuf) in self.rmr_get_messages():
+ print(summary)
+ self.rmr_free(sbuf)
+
+ time.sleep(1)
+
+
+xapp = MyXapp(4564, use_fake_sdl=True)
+xapp.run()
--- /dev/null
+"""
+Test xapp 2 that works with 1
+"""
+# ==================================================================================
+# 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
+from ricxappframe.xapp_frame import RMRXapp
+
+
+# Note, this is an OOP pattern for this that I find slightly more natural
+# The problem is we want the client xapp to be able to call methods defined in the RMRXapp
+# Another exactly equivelent way would have been to use Closures like
+# def consume(summary, sbuf):
+# xapp.rts()
+# xapp = RMRXapp(consume)
+# However, the subclass looks slightly more natural. Open to the alternative.
+
+
+class MyXapp(RMRXapp):
+ def consume(self, summary, sbuf):
+ """callbnack called for each new message"""
+ print(summary)
+ jpay = json.loads(summary["payload"])
+ self.rmr_rts(sbuf, new_payload=json.dumps({"ACK": jpay["test_send"]}).encode(), new_mtype=60001, retries=100)
+ self.rmr_free(sbuf)
+
+
+xapp = MyXapp(use_fake_sdl=True)
+xapp.run()
--- /dev/null
+newrt|start
+rte|60000|localhost:4562
+newrt|end
--- /dev/null
+"""
+Framework for python xapps
+Framework here means Xapp classes that can be subclassed
+"""
+# ==================================================================================
+# 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.
+# ==================================================================================
+
+
+from ricxappframe import xapp_rmr
+from ricxappframe.xapp_sdl import SDLWrapper
+from rmr import rmr
+from mdclogpy import Logger
+
+
+mdc_logger = Logger(name=__name__)
+
+
+# Private base class; not for direct client use
+
+
+class _BaseXapp:
+ """
+ Base xapp; not for client use directly
+ """
+
+ def __init__(self, rmr_port=4562, rmr_wait_for_ready=True, use_fake_sdl=False):
+ """
+ Init
+
+ Parameters
+ ----------
+ rmr_port: int
+ port to listen on
+
+ rmr_wait_for_ready: bool (optional)
+ if this is True, then init waits until rmr is ready to send, which includes having a valid routing file.
+ this can be set to False if the client only wants to *receive only*
+
+ use_fake_sdl: bool (optional)
+ if this is True, it uses dbaas' "fake dict backend" instead of Redis or other backends.
+ Set this to true when developing your xapp or during unit testing to completely avoid needing a dbaas running or any network at all
+ """
+
+ # Start rmr rcv thread
+ self._rmr_loop = xapp_rmr.RmrLoop(port=rmr_port, wait_for_ready=rmr_wait_for_ready)
+ self._mrc = self._rmr_loop.mrc # for convenience
+
+ # SDL
+ self._sdl = SDLWrapper(use_fake_sdl)
+
+ def rmr_get_messages(self):
+ """
+ returns a generator iterable over all current messages in the queue that have not yet been read by the client xapp
+ """
+ while not self._rmr_loop.rcv_queue.empty():
+ (summary, sbuf) = self._rmr_loop.rcv_queue.get()
+ yield (summary, sbuf)
+
+ def rmr_send(self, payload, mtype, retries=100):
+ """
+ Allocates a buffer, sets payload and mtype, and sends
+
+ Parameters
+ ----------
+ payload: bytes
+ payload to set
+ mtype: int
+ message type
+ retries: int (optional)
+ Number of times to retry at the application level before excepting RMRFailure
+
+ Returns
+ -------
+ bool
+ whether or not the send worked after retries attempts
+ """
+ sbuf = rmr.rmr_alloc_msg(vctx=self._mrc, size=len(payload), payload=payload, gen_transaction_id=True, mtype=mtype)
+
+ for _ in range(retries):
+ sbuf = rmr.rmr_send_msg(self._mrc, sbuf)
+ if sbuf.contents.state == 0:
+ self.rmr_free(sbuf)
+ return True
+
+ self.rmr_free(sbuf)
+ return False
+
+ def rmr_rts(self, sbuf, new_payload=None, new_mtype=None, retries=100):
+ """
+ Allows the xapp to return to sender, possibly adjusting the payload and message type before doing so
+
+ This does NOT free the sbuf for the caller as the caller may wish to perform multiple rts per buffer.
+ The client needs to free.
+
+ Parameters
+ ----------
+ sbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+ new_payload: bytes (optional)
+ New payload to set
+ new_mtype: int (optional)
+ New message type (replaces the received message)
+ retries: int (optional)
+ Number of times to retry at the application level before excepting RMRFailure
+
+ Returns
+ -------
+ bool
+ whether or not the send worked after retries attempts
+ """
+ for _ in range(retries):
+ sbuf = rmr.rmr_rts_msg(self._mrc, sbuf, payload=new_payload, mtype=new_mtype)
+ if sbuf.contents.state == 0:
+ return True
+
+ return False
+
+ def rmr_free(self, sbuf):
+ """
+ Free an rmr message buffer after use
+
+ Note: this does not need to be a class method, self is not used. However if we break it out as a function we need a home for it.
+ Parameters
+ ----------
+ sbuf: ctypes c_void_p
+ Pointer to an rmr message buffer
+ """
+ rmr.rmr_free_msg(sbuf)
+
+ # SDL
+ # NOTE, even though these are passthroughs, the seperate SDL wrapper is useful for other applications like A1.
+ # Therefore, we don't embed that SDLWrapper functionality here so that it can be instantiated on it's own.
+
+ def sdl_set(self, ns, key, value, usemsgpack=True):
+ """
+ set a key
+
+ Parameters
+ ----------
+ ns: string
+ the sdl namespace
+ key: string
+ the sdl key
+ value:
+ if usemsgpack is True, value can be anything serializable by msgpack
+ if usemsgpack is False, value must be bytes
+ usemsgpack: boolean (optional)
+ determines whether the value is serialized using msgpack
+ """
+ self._sdl.set(ns, key, value, usemsgpack)
+
+ def sdl_get(self, ns, key, usemsgpack=True):
+ """
+ get a key
+
+ Parameters
+ ----------
+ ns: string
+ the sdl namespace
+ key: string
+ the sdl key
+ usemsgpack: boolean (optional)
+ if usemsgpack is True, the value is deserialized using msgpack
+ if usemsgpack is False, the value is returned as raw bytes
+
+ Returns
+ -------
+ None (if not exist) or see above; depends on usemsgpack
+ """
+ return self._sdl.get(ns, key, usemsgpack)
+
+ def sdl_find_and_get(self, ns, prefix, usemsgpack=True):
+ """
+ get all k v pairs that start with prefix
+
+ Parameters
+ ----------
+ ns: string
+ the sdl namespace
+ key: string
+ the sdl key
+ prefix: string
+ the prefix
+ usemsgpack: boolean (optional)
+ if usemsgpack is True, the value returned is a dict where each value has been deserialized using msgpack
+ if usemsgpack is False, the value returned is as a dict mapping keys to raw bytes
+
+ Returns
+ -------
+ {} (if no keys match) or see above; depends on usemsgpack
+ """
+ return self._sdl.find_and_get(ns, prefix, usemsgpack)
+
+ def sdl_delete(self, ns, key):
+ """
+ delete a key
+
+ Parameters
+ ----------
+ ns: string
+ the sdl namespace
+ key: string
+ the sdl key
+ """
+ self._sdl.delete(ns, key)
+
+ # Health
+
+ def healthcheck(self):
+ """
+ this needs to be understood how this is supposed to work
+ """
+ return self._rmr_loop.healthcheck() and self._sdl.healthcheck()
+
+ def stop(self):
+ """
+ cleans up and stops the xapp.
+ Currently this only stops the rmr thread
+ This is critical for unit testing as pytest will never return if the thread is running.
+
+ TODO: can we register a ctrl-c handler so this gets called on ctrl-c? Because currently two ctrl-c are needed to stop
+ """
+ self._rmr_loop.stop()
+
+
+# Public Classes to subclass (these subclass _BaseXapp)
+
+
+class RMRXapp(_BaseXapp):
+ """
+ Represents an xapp that is purely driven by rmr messages (i.e., when messages are received, the xapp does something
+ When run is called, the xapp framework waits for rmr messages, and calls the client provided consume callback on every one
+ """
+
+ def consume(self, summary, sbuf):
+ """
+ This function is to be implemented by the client and is called whenever a new rmr message is received.
+ It is expected to take two parameters (besides self):
+
+ Parameters
+ ----------
+ summary: dict
+ the rmr message summary
+ sbuf: ctypes c_void_p
+ Pointer to an rmr message buffer. The user must call free on this when done.
+ """
+ raise NotImplementedError()
+
+ def run(self):
+ """
+ This function should be called when the client xapp is ready to wait for consume to be called on received messages
+
+ TODO: should we run this in a thread too? We can't currently call "stop" on rmr xapps at an arbitrary time because this doesn't return control
+ Running the below in a thread probably makes the most sense.
+ """
+ while True:
+ if not self._rmr_loop.rcv_queue.empty():
+ (summary, sbuf) = self._rmr_loop.rcv_queue.get()
+ self.consume(summary, sbuf)
+
+
+class Xapp(_BaseXapp):
+ """
+ Represents an xapp where the client provides a generic function to call, which is mostly likely a loop-forever loop
+ """
+
+ def loop(self):
+ """
+ This function is to be implemented by the client and is called
+ """
+ raise NotImplementedError()
+
+ def run(self):
+ """
+ This function should be called when the client xapp is ready to start their loop
+ This is simple and the client could just call self.loop(), however this gives a consistent interface as the other xapps
+ """
+ self.loop()
--- /dev/null
+"""
+Contains rmr functionality specific to the xapp
+The general rmr API is via "rmr"
+"""
+# ==================================================================================
+# 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 time
+import queue
+from threading import Thread
+from mdclogpy import Logger
+from rmr import rmr, helpers
+
+
+mdc_logger = Logger(name=__name__)
+
+
+class RmrLoop:
+ """
+ Class represents an rmr loop that constantly reads from rmr
+
+ Note, we use a queue here, and a thread, rather than the xapp frame just looping and calling consume, so that a possibly slow running consume function does not block the reading of new messages
+ """
+
+ def __init__(self, port, wait_for_ready=True):
+ """
+ sets up rmr, then launches a thread that reads and injects messages into a queue
+
+ Parameters
+ ----------
+ port: int
+ port to listen on
+
+ wait_for_ready: bool (optional)
+ if this is True, then this function hangs until rmr is ready to send, which includes having a valid routing file.
+ this can be set to False if the client only wants to *receive only*
+ """
+
+ # Public
+ # thread safe queue https://docs.python.org/3/library/queue.html
+ # We use a thread and a queue so that a long running consume callback function can never block reads.
+ # IE a consume implementation could take a long time and the ring size for rmr blows up here and messages are lost
+ self.rcv_queue = queue.Queue()
+
+ # rmr context; RMRFL_MTCALL puts RMR into a multithreaded mode, where a thread populates a ring of messages that receive calls read from
+ self.mrc = rmr.rmr_init(str(port).encode(), rmr.RMR_MAX_RCV_BYTES, rmr.RMRFL_MTCALL)
+
+ if wait_for_ready:
+ mdc_logger.debug("Waiting for rmr to init on port {}..".format(port))
+ while rmr.rmr_ready(self.mrc) == 0:
+ time.sleep(0.1)
+
+ # Private
+ self._keep_going = True
+ self._last_ran = time.time()
+
+ # start the work loop
+ mdc_logger.debug("Starting loop thread")
+
+ def loop():
+ mdc_logger.debug("Work loop starting")
+ while self._keep_going:
+
+ # read our mailbox
+ # TODO: take a flag as to whether RAW is needed or not
+ # RAW allows for RTS however the caller must free, and the caller may not need RTS.
+ # Currently after consuming, callers should do rmr.rmr_free_msg(sbuf)
+
+ for (msg, sbuf) in helpers.rmr_rcvall_msgs_raw(self.mrc):
+ self.rcv_queue.put((msg, sbuf))
+
+ self._last_ran = time.time()
+
+ self._thread = Thread(target=loop)
+ self._thread.start()
+
+ def stop(self):
+ """
+ sets a flag that will cleanly stop the thread
+ note, this does not yet have a use yet for xapps to call, however this is very handy during unit testing.
+ """
+ self._keep_going = False
+
+ def healthcheck(self, seconds=30):
+ """
+ returns a boolean representing whether the rmr loop is healthy, by checking two attributes:
+ 1. is it running?,
+ 2. is it stuck in a long (> seconds) loop?
+
+ Parameters
+ ----------
+ seconds: int (optional)
+ the rmr loop is determined healthy if it has completed in the last (seconds)
+ """
+ return self._thread.is_alive() and ((time.time() - self._last_ran) < seconds)
--- /dev/null
+"""
+sdl functionality
+"""
+# ==================================================================================
+# 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 msgpack
+from ricsdl.syncstorage import SyncStorage
+
+
+class SDLWrapper:
+ """
+ This is a wrapper around the SDL Python interface.
+
+ We do not embed the below directly in the Xapp classes because this SDL wrapper is useful for other python apps, for example A1 Mediator uses this verbatim.
+ Therefore, we leave this here as a seperate instantiable object so it can be used outside of xapps too
+ One could argue this get moved into *sdl itself*.
+
+ We currently use msgpack for binary (de)serialization: https://msgpack.org/index.html
+ """
+
+ def __init__(self, use_fake_sdl=False):
+ """
+ init
+
+ Parameters
+ ----------
+ use_fake_sdl: bool
+ if this is True (default: False), then SDLs "fake dict backend" is used, which is very useful for testing since it allows you to use SDL without any SDL or Redis deployed at all.
+ This can be used while developing your xapp, and also for monkeypatching during unit testing (e.g., the xapp framework unit tests do this)
+ """
+ if use_fake_sdl:
+ self._sdl = SyncStorage(fake_db_backend="dict")
+ else:
+ self._sdl = SyncStorage()
+
+ def set(self, ns, key, value, usemsgpack=True):
+ """
+ set a key
+
+ NOTE: I am down for a discussion about whether usemsgpack should *default* to True or False here. This seems like a usage statistic question (that we don't have enough data for yet). Are more uses for an xapp to write/read their own data, or will more xapps end up reading data written by some other thing? I think it's too early to know this. So we go with True as the very first user of this, a1, does this. I'm open to changing this default to False later with evidence.
+
+ Parameters
+ ----------
+ ns: string
+ the sdl namespace
+ key: string
+ the sdl key
+ value:
+ if usemsgpack is True, value can be anything serializable by msgpack
+ if usemsgpack is False, value must be bytes
+ usemsgpack: boolean (optional)
+ determines whether the value is serialized using msgpack
+ """
+ if usemsgpack:
+ self._sdl.set(ns, {key: msgpack.packb(value, use_bin_type=True)})
+ else:
+ self._sdl.set(ns, {key: value})
+
+ def get(self, ns, key, usemsgpack=True):
+ """
+ get a key
+
+ Parameters
+ ----------
+ ns: string
+ the sdl namespace
+ key: string
+ the sdl key
+ usemsgpack: boolean (optional)
+ if usemsgpack is True, the value is deserialized using msgpack
+ if usemsgpack is False, the value is returned as raw bytes
+
+ Returns
+ -------
+ None (if not exist) or see above; depends on usemsgpack
+ """
+ ret_dict = self._sdl.get(ns, {key})
+ if key in ret_dict:
+ if usemsgpack:
+ return msgpack.unpackb(ret_dict[key], raw=False)
+ return ret_dict[key]
+
+ return None
+
+ def find_and_get(self, ns, prefix, usemsgpack=True):
+ """
+ get all k v pairs that start with prefix
+
+ Parameters
+ ----------
+ ns: string
+ the sdl namespace
+ key: string
+ the sdl key
+ prefix: string
+ the prefix
+ usemsgpack: boolean (optional)
+ if usemsgpack is True, the value returned is a dict where each value has been deserialized using msgpack
+ if usemsgpack is False, the value returned is as a dict mapping keys to raw bytes
+
+ Returns
+ -------
+ {} (if no keys match) or see above; depends on usemsgpack
+ """
+
+ # note: SDL "*" usage is inconsistent with real python regex, where it would be ".*"
+ ret_dict = self._sdl.find_and_get(ns, "{0}*".format(prefix))
+ if usemsgpack:
+ return {k: msgpack.unpackb(v, raw=False) for k, v in ret_dict.items()}
+ return ret_dict
+
+ def delete(self, ns, key):
+ """
+ delete a key
+
+ Parameters
+ ----------
+ ns: string
+ the sdl namespace
+ key: string
+ the sdl key
+ """
+ self._sdl.remove(ns, {key})
+
+ def healthcheck(self):
+ """
+ checks if the sdl connection is healthy
+
+ Returns
+ -------
+ bool
+ """
+ return self._sdl.is_active()
--- /dev/null
+# CI script installs RMR from PackageCloud using this version
+---
+version: 1.13.1
--- /dev/null
+# ==================================================================================
+# 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.
+# ==================================================================================
+from setuptools import setup, find_packages
+
+setup(
+ name="ricxappframe",
+ version="0.1.0",
+ packages=find_packages(exclude=["tests.*", "tests"]),
+ author="Tommy Carpenter",
+ description="Xapp framework for python",
+ url="https://gerrit.o-ran-sc.org/r/admin/repos/ric-plt/xapp-frame-py",
+ install_requires=["msgpack", "rmr>=2.2.0, <3.0.0", "mdclogpy", "ricsdl>=2.0.3,<3.0.0"],
+)
--- /dev/null
+"""
+tests data functions
+"""
+# ==================================================================================
+# Copyright (c) 2019-2020 Nokia
+# Copyright (c) 2018-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.
+# ==================================================================================
+from ricxappframe.xapp_sdl import SDLWrapper
+
+
+NS = "testns"
+
+
+def test_sdl():
+ """
+ test raw sdl functions
+ """
+ sdl = SDLWrapper(use_fake_sdl=True)
+ sdl.set(NS, "as.df1", "data")
+ sdl.set(NS, "as.df2", "data2")
+ assert sdl.get(NS, "as.df1") == "data"
+ assert sdl.get(NS, "as.df2") == "data2"
+ assert sdl.find_and_get(NS, "as.df1") == {"as.df1": "data"}
+ assert sdl.find_and_get(NS, "as.df2") == {"as.df2": "data2"}
+ assert sdl.find_and_get(NS, "as.df") == {"as.df1": "data", "as.df2": "data2"}
+ assert sdl.find_and_get(NS, "as.d") == {"as.df1": "data", "as.df2": "data2"}
+ assert sdl.find_and_get(NS, "as.") == {"as.df1": "data", "as.df2": "data2"}
+ assert sdl.find_and_get(NS, "asd") == {}
+
+ # delete 1
+ sdl.delete(NS, "as.df1")
+ assert sdl.get(NS, "as.df1") is None
+ assert sdl.get(NS, "as.df2") == "data2"
+
+ # delete 2
+ sdl.delete(NS, "as.df2")
+ assert sdl.get(NS, "as.df2") is None
+
+ assert sdl.find_and_get(NS, "as.df") == {}
+ assert sdl.find_and_get(NS, "") == {}
--- /dev/null
+# ==================================================================================
+# 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.
+# ==================================================================================
+from ricxappframe.xapp_frame import Xapp
+
+gen_xapp = None
+rmr_xapp = None
+
+
+def test_init_general_xapp():
+ class MyXapp(Xapp):
+ # TODO: obviouslly a lot more is needed here. For now this tests that the class is instantiable.
+ def loop(self):
+ print("ok")
+
+ global gen_xapp
+ gen_xapp = MyXapp(rmr_wait_for_ready=False)
+ gen_xapp.run()
+
+
+def teardown_module():
+ """
+ module teardown
+
+ pytest will never return without this.
+ """
+ gen_xapp.stop()
--- /dev/null
+# ==================================================================================
+# 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.
+# ==================================================================================
+[tox]
+envlist = py37,flake8
+# ,docs,docs-linkcheck
+minversion = 2.0
+
+[testenv]
+basepython = python3.7
+deps=
+ pytest
+ coverage
+ pytest-cov
+setenv =
+ LD_LIBRARY_PATH = /usr/local/lib/:/usr/local/lib64
+
+commands =
+ pytest --junitxml xunit-results.xml --cov ricxappframe --cov-report xml --cov-report term-missing --cov-report html --cov-fail-under=50
+ coverage xml -i
+
+[testenv:flake8]
+basepython = python3.7
+skip_install = true
+deps = flake8
+commands = flake8 setup.py ricxappframe tests
+
+[flake8]
+extend-ignore = E501,E741,E731
+
+# verbatim as asked for by the docs instructions: https://wiki.o-ran-sc.org/display/DOC/Configure+Repo+for+Documentation
+# [testenv:docs]
+# basepython = python3.7
+# deps =
+# sphinx
+# sphinx-rtd-theme
+# sphinxcontrib-httpdomain
+# recommonmark
+# lfdocs-conf
+
+#commands =
+# sphinx-build -W -b html -n -d {envtmpdir}/doctrees ./docs/ {toxinidir}/docs/_build/html
+# echo "Generated docs available in {toxinidir}/docs/_build/html"
+#whitelist_externals = echo
+
+#[testenv:docs-linkcheck]
+#basepython = python3.7
+#deps = sphinx
+# sphinx-rtd-theme
+# sphinxcontrib-httpdomain
+# recommonmark
+# lfdocs-conf
+#commands = sphinx-build -W -b linkcheck -d {envtmpdir}/doctrees ./docs/ {toxinidir}/docs/_build/linkcheck
Code Review / ric-plt / xapp-frame-py.git / commitdiff