About test_subs_oper.py

[fourwilling]

Hi rjarry,

I tried to modify one of your tests as below. However, it seems that this can only run successfully for one time. When I ran it for the second time, it gave me "Module "sysrepo-example" was not found in sysrepo"

Do you know where I went wrong? Thank you!

import logging
import os
import unittest
import sysrepo

YANG_FILE = os.path.join(
    os.path.dirname(os.path.dirname(__file__)), "examples/sysrepo-example.yang"
)
sysrepo.configure_logging(stderr_level=logging.ERROR)

# ------------------------------------------------------------------------------
class OperSubscriptionTest(unittest.TestCase):
    def test_oper_sub(self):
        priv = object()
        state = None

        def oper_data_cb(xpath, private_data):
            self.assertEqual(xpath, "/sysrepo-example:state")
            self.assertEqual(private_data, priv)
            return state

        with sysrepo.SysrepoConnection() as conn:
            # conn.remove_module("sysrepo-example")
            conn.install_module(YANG_FILE, enabled_features=["turbo"])
            with conn.start_session("operational") as sess:
                sess.subscribe_oper_data_request(
                    "sysrepo-example",
                    "/sysrepo-example:state",
                    oper_data_cb,
                    private_data=priv,
                    strict=True,
                )
                with conn.start_session("operational") as op_sess:
                    state = {
                        "state": {
                            "system": {"hostname": "foo"},
                            "network": {
                                "interface": [
                                    {
                                        "name": "eth0",
                                        "address": "1.2.3.4/24",
                                        "up": True,
                                        "stats": {"rx": 42, "tx": 42},
                                    }
                                ]
                            },
                        }
                    }
                    self.assertEqual(op_sess.get_data("/sysrepo-example:state"), state)
                    state = {"state": {"invalid": True}}
                    with self.assertRaises(sysrepo.SysrepoCallbackFailedError):
                        op_sess.get_data("/sysrepo-example:state")

            conn.disconnect()
            conn.remove_module("sysrepo-example")

unittest.main()

[rjarry]

I think you need to completely disconnect all sysrepo applications and reconnect once to apply all datastore schema changes. See https://netopeer.liberouter.org/doc/sysrepo/master/html/schema.html for more details.

[fourwilling]

Hi rjarry,

Doesn't "conn.disconnect()" fully disconnect sysrepo applications? I don't know why it works if I move "conn.disconnect()" in front of "conn.install_module()"

import logging
import os
import unittest

import sysrepo


YANG_FILE = os.path.join(
    os.path.dirname(os.path.dirname(__file__)), "examples/sysrepo-example.yang"
)
sysrepo.configure_logging(stderr_level=logging.ERROR)

# ------------------------------------------------------------------------------
class OperSubscriptionTest(unittest.TestCase):
    # @classmethod
    # def setUpClass(cls):
    #     with sysrepo.SysrepoConnection() as conn:
    #         conn.install_module(YANG_FILE, enabled_features=["turbo"])
    #     cls.conn = sysrepo.SysrepoConnection(err_on_sched_fail=True)
    #     cls.sess = cls.conn.start_session()

    # @classmethod
    # def tearDownClass(cls):
    #     cls.sess.stop()
    #     cls.conn.remove_module("sysrepo-example")
    #     cls.conn.disconnect()
    #     # reconnect to make sure module is removed
    #     with sysrepo.SysrepoConnection(err_on_sched_fail=True):
    #         pass
    def test_oper_sub(self):
        priv = object()
        state = None

        def oper_data_cb(xpath, private_data):
            self.assertEqual(xpath, "/sysrepo-example:state")
            self.assertEqual(private_data, priv)
            return state

        with sysrepo.SysrepoConnection() as conn:
            conn.remove_module("sysrepo-example")
            conn.install_module(YANG_FILE)
            with conn.start_session() as sess:
                sess.subscribe_oper_data_request(
                    "sysrepo-example",
                    "/sysrepo-example:state",
                    oper_data_cb,
                    private_data=priv,
                    strict=True,
                )
                with conn.start_session("operational") as op_sess:
                    state = {
                        "state": {
                            "system": {"hostname": "foo"},
                            "network": {
                                "interface": [
                                    {
                                        "name": "eth0",
                                        "address": "1.2.3.4/24",
                                        "up": True,
                                        "stats": {"rx": 42, "tx": 42},
                                    }
                                ]
                            },
                        }
                    }
                    print(op_sess.get_data("/sysrepo-example:state"), state)
                    self.assertEqual(op_sess.get_data("/sysrepo-example:state"), state)
                    state = {"state": {"invalid": True}}
                    with self.assertRaises(sysrepo.SysrepoCallbackFailedError):
                        op_sess.get_data("/sysrepo-example:state")

            # conn.remove_module("sysrepo-example") 
            conn.disconnect()

unittest.main()

Also, is there a function to show all connected sysrepo applications?

Thank you!

[rjarry]

conn.disconnect() only disconnects the current process, not the other connected sysrepo applications.

There is an API function to get the number of current connections but I did not make it available in the python bindings. I do not think there is any API to show all connected processes.

Maybe you can list processes that have open files in /dev/shm/sr*:

lsof /dev/shm/sr*