Fix read_only enforcement in __exit__ and guard devices access in rpc…

clixon/clixon.py

@@ -141,31 +141,33 @@ def __exit__(self, *args: object) -> None:
 
         """
         if self.__read_only:
-            logger.info("Read only mode enabled")
+            logger.info("Read only mode enabled, skipping config set")
+        else:
+            try:
+                if self.__root is None:
+                    self.__root = self.get_root()
 
-        try:
-            if self.__root is None:
-                self.__root = self.get_root()
+                for child in self.__root:
+                    config = rpc_config_set(
+                        child, user=self.__user, target=self.__target
+                    )
+                    send(self.__socket, config, pp)
+                    data = read(self.__socket, pp)
 
-            for child in self.__root:
-                config = rpc_config_set(child, user=self.__user, target=self.__target)
-                send(self.__socket, config, pp)
-                data = read(self.__socket, pp)
+                    self.__handle_errors(data)
 
-                self.__handle_errors(data)
+                if self.__commit:
+                    self.commit()
 
-            if self.__commit:
-                self.commit()
+            except Exception as e:
+                logger.error(f"Got exception from Clixon.__exit__: {e}")
+                raise Exception(f"{e}")
 
-        except Exception as e:
-            logger.error(f"Got exception from Clixon.__exit__: {e}")
-            raise Exception(f"{e}")
-        finally:
-            if not self.__from_server:
-                try:
-                    self.close_session()
-                except Exception as e:
-                    logger.error(f"Failed to send close-session: {e}")
+        if not self.__from_server:
+            try:
+                self.close_session()
+            except Exception as e:
+                logger.error(f"Failed to send close-session: {e}")
 
     def commit(self) -> None:
         """
@@ -511,6 +513,10 @@ def apply_template(
         :rtype: None
         """
 
+        if self.__read_only:
+            logger.info("Read only mode enabled")
+            return
+
         if inline:
             rpc = rpc_apply_template(
                 devname,

clixon/netconf.py

@@ -129,25 +129,28 @@ def rpc_config_set(
     root.rpc.edit_config.default_operation.cdata = "merge"
     root.rpc.edit_config.create("config")
 
-    if root.rpc.edit_config.config.get_elements("devices") == []:
-        root.rpc.edit_config.config.create("devices", attributes=CONTROLLER_NS)
-        logger.debug("Created configuration node devices.")
-
     for node in config.get_elements():
         if node.get_name() == "devices":
             continue
 
         root.rpc.edit_config.config.add(node)
         logger.debug(f"Added node {node.get_name()} to configuration.")
 
-    for device in config.devices.device:
-        if device.find_modified():
-            logger.debug(
-                f"Modifications found on device {device.name.get_data()}, added to configuration."
-            )
-            root.rpc.edit_config.config.devices.add(device)
-        else:
-            logger.debug(f"No modifications found on device {device.name.get_data()}")
+    if config.get_elements("devices"):
+        if root.rpc.edit_config.config.get_elements("devices") == []:
+            root.rpc.edit_config.config.create("devices", attributes=CONTROLLER_NS)
+            logger.debug("Created configuration node devices.")
+
+        for device in config.devices.device:
+            if device.find_modified():
+                logger.debug(
+                    f"Modifications found on device {device.name.get_data()}, added to configuration."
+                )
+                root.rpc.edit_config.config.devices.add(device)
+            else:
+                logger.debug(
+                    f"No modifications found on device {device.name.get_data()}"
+                )
 
     return root
 

tests/test_netconf.py

@@ -511,3 +511,34 @@ def test_rpc_config_get_with_multiple_namespaces():
     )
 
     assert root.dumps() == xmlstr
+
+
+def test_rpc_config_set_without_devices():
+    """
+    Test the rpc_config_set function without devices in config.
+    Verifies no crash when config has no devices node (e.g. xpath="/services").
+    """
+
+    xmlstr = f"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="{user}" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42" xmlns:cl="http://clicon.org/lib"><edit-config><target><actions xmlns="http://clicon.org/controller"/></target><default-operation>merge</default-operation><config><services><service-name>test</service-name></services></config></edit-config></rpc>"""
+
+    config = Element("config", {})
+    config.create("services").create("service-name", data="test")
+    root = netconf.rpc_config_set(config)
+
+    assert root.dumps() == xmlstr
+
+
+def test_rpc_config_set_with_services_and_devices():
+    """
+    Test the rpc_config_set function with both services and devices.
+    Verifies services are included and devices block is handled correctly.
+    """
+
+    xmlstr = f"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="{user}" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42" xmlns:cl="http://clicon.org/lib"><edit-config><target><actions xmlns="http://clicon.org/controller"/></target><default-operation>merge</default-operation><config><services><service-name>test</service-name></services><devices xmlns="http://clicon.org/controller"/></config></edit-config></rpc>"""
+
+    config = Element("config", {})
+    config.create("services").create("service-name", data="test")
+    config.create("devices").create("device").create("name", data="foo")
+    root = netconf.rpc_config_set(config)
+
+    assert root.dumps() == xmlstr