diff --git a/clixon/clixon.py b/clixon/clixon.py
index 3e34393..91a9664 100644
--- a/clixon/clixon.py
+++ b/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,
diff --git a/clixon/netconf.py b/clixon/netconf.py
index a199286..30231f1 100644
--- a/clixon/netconf.py
+++ b/clixon/netconf.py
@@ -129,10 +129,6 @@ 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
@@ -140,14 +136,21 @@ def rpc_config_set(
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
diff --git a/tests/test_netconf.py b/tests/test_netconf.py
index a5c5122..151cda1 100644
--- a/tests/test_netconf.py
+++ b/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"""mergetest"""
+
+ 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"""mergetest"""
+
+ 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