session: add locking support

Author: tbsuht

README.rst

@@ -153,6 +153,7 @@ Supported Features
    ``sr_edit_batch()``, ``sr_validate()``, ``sr_apply_changes()``,
    ``sr_discard_changes()``, ``sr_replace_config()``)
 -  Get data (``sr_get_data()``, ``sr_get_item()``, ``sr_get_items()``)
+-  Module locking (``sr_*lock*``)
 
 __ https://pypi.org/project/libyang/
 .. _async: https://docs.python.org/3/library/asyncio-task.html#coroutine
@@ -163,7 +164,6 @@ Not Yet Supported Features
 All other features are not yet supported by sysrepo-python. The most notable
 are:
 
--  Module locking (``sr_*lock*``)
 -  Module management (``sr_*_module_*``)
 
 Contributing

cffi/cdefs.h

@@ -101,6 +101,9 @@ int sr_session_set_orig_name(sr_session_ctx_t *session, const char *);
 int sr_session_get_orig_data(sr_session_ctx_t *session, uint32_t idx, uint32_t *size, const void **data);
 int sr_session_push_orig_data(sr_session_ctx_t * session, uint32_t size, const void *data);
 
+int sr_lock(sr_session_ctx_t *session, const char *module_name, uint32_t timeout_ms);
+int sr_unlock(sr_session_ctx_t *session, const char *module_name);
+
 typedef enum sr_val_type_e {
 	SR_UNKNOWN_T,
 	SR_LIST_T,

sysrepo/session.py

@@ -111,6 +111,42 @@ def get_datastore(self) -> str:
         """
         return datastore_name(lib.sr_session_get_ds(self.cdata))
 
+    def lock(self, module_name: str = "", timeout_ms: int = 0) -> None:
+        """
+        Locks the data of the specified module or the whole datastore.
+        :arg module_name:
+            Optional name of the module to be locked.
+        :arg timeout_ms:
+            Optional timeout in ms for waiting. If 0, no waiting is performed.
+        """
+        module = str2c(module_name) if len(module_name) > 0 else ffi.NULL
+        check_call(lib.sr_lock, self.cdata, module, timeout_ms)
+
+    def unlock(self, module_name: str = "") -> None:
+        """
+        Unlocks the data of the specified module or the whole datastore.
+        :arg module_name:
+            Optional name of the module to be locked.
+        """
+        module = str2c(module_name) if len(module_name) > 0 else ffi.NULL
+        check_call(lib.sr_unlock, self.cdata, module)
+
+    @contextmanager
+    def locked(self, module_name: str = "", timeout_ms: int = 0):
+        """
+        Convenience method which manages un-/locking the data of the specified module
+        or the whole datastore.
+        :arg module_name:
+            Optional name of the module to be locked.
+        :arg timeout_ms:
+            Optional timeout in ms for waiting. If 0, no waiting is performed.
+        """
+        try:
+            self.lock(module_name, timeout_ms)
+            yield
+        finally:
+            self.unlock(module_name)
+
     def switch_datastore(self, datastore: str) -> None:
         """
         Change datastore which the session operates on. All subsequent calls will be

tests/test_session.py

@@ -3,6 +3,8 @@
 
 import logging
 import os
+import threading
+import time
 import types
 import unittest
 
@@ -162,3 +164,43 @@ def test_get_netconf_id_and_get_user_are_only_available_in_implicit_session(self
 
             with self.assertRaises(sysrepo.SysrepoUnsupportedError):
                 sess.get_user()
+
+    def test_basic_session_lock(self):
+        # lock whole datastore
+        with self.conn.start_session("running") as sess:
+            with sess.locked():
+                config = {"conf": {"system": {"hostname": "foobar1"}}}
+                sess.replace_config(config, "sysrepo-example")
+                sess.apply_changes()
+
+        # lock specific module name with a timeout
+        with self.conn.start_session("running") as sess:
+            with sess.locked("sysrepo-example", 1000):
+                config = {"conf": {"system": {"hostname": "foobar2"}}}
+                sess.replace_config(config, "sysrepo-example")
+                sess.apply_changes()
+
+    def test_concurrent_session_lock(self):
+        def function_thread_one():
+            with self.conn.start_session("running") as sess:
+                with sess.locked():
+                    config = {"conf": {"system": {"hostname": "foobar1"}}}
+                    sess.replace_config(config, "sysrepo-example")
+                    sess.apply_changes()
+                    time.sleep(3)  # keep the lock active
+
+        def function_thread_two():
+            time.sleep(1)
+            with self.conn.start_session("running") as sess:
+                with self.assertRaises(sysrepo.SysrepoLockedError):
+                    with sess.locked():
+                        pass
+
+        thread_one = threading.Thread(target=function_thread_one)
+        thread_two = threading.Thread(target=function_thread_two)
+
+        thread_one.start()
+        thread_two.start()
+
+        thread_one.join()
+        thread_two.join()