test/infamy: Use PATCH for RESTCONF merge semantics

Switch RESTCONF from PUT to PATCH for configuration updates. This gives
RESTCONF the same merge behavior as NETCONF edit-config, allowing NACM
to correctly enforce path-specific permissions instead of requiring
write access to the entire datastore.

Fixes #617

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>

test/infamy/netconf.py

@@ -281,7 +281,7 @@ class Device(Transport):
         """Get Python dictionary version of XML configuration"""
         return self.get_config(xpath).print_dict()
 
-    def put_config(self, edit):
+    def put_config(self, edit, retries=3):
         """Send XML configuration over NETCONF"""
         yang2nc = {
             "none": None,
@@ -298,7 +298,7 @@ class Device(Transport):
                               f"nc:operation=\"{dst}\"" if dst else "")
 
         last_error = None
-        for _ in range(0, 3):
+        for _ in range(0, retries):
             try:
                 self.ncc.edit_config(xml, default_operation='merge')
                 last_error = None
@@ -313,7 +313,7 @@ class Device(Transport):
         if last_error is not None:
             raise last_error
 
-    def put_config_dicts(self, models):
+    def put_config_dicts(self, models, retries=3):
         """PUT full configuration of all models to running-config"""
         config = ""
         infer_put_dict(self.name, models)
@@ -328,9 +328,9 @@ class Device(Transport):
             lyd = mod.parse_data_dict(models[model], no_state=True, validate=False)
             config += lyd.print_mem("xml", with_siblings=True, pretty=False) + "\n"
         # print(f"Send new XML config: {config}")
-        return self.put_config(config)
+        return self.put_config(config, retries=retries)
 
-    def put_config_dict(self, modname, edit):
+    def put_config_dict(self, modname, edit, retries=3):
         """Convert Python dictionary to XMl and send as configuration"""
         try:
             mod = self.ly.get_module(modname)
@@ -341,7 +341,7 @@ class Device(Transport):
         lyd = mod.parse_data_dict(edit, no_state=True, validate=False)
         config = lyd.print_mem("xml", with_siblings=True, pretty=False)
         # print(f"Send new XML config: {config}")
-        return self.put_config(config)
+        return self.put_config(config, retries=retries)
 
     def call(self, call):
         """Call RPC, XML version"""

test/infamy/restconf.py

@@ -51,8 +51,8 @@ def requests_workaround(method, url, json, headers, auth, verify=False, retry=0)
     request = requests.Request(method, url, json=json, headers=headers,
                                auth=auth)
     prepared_request = session.prepare_request(request)
-    prepared_request.url = prepared_request.url.replace('%25', '%')
-    prepared_request.url = prepared_request.url.replace('%3A', ':')
+    prepared_request.url = re.sub(r'%25', '%', prepared_request.url)
+    prepared_request.url = re.sub(r'%3a', ':', prepared_request.url, flags=re.IGNORECASE)
     response = session.send(prepared_request, verify=verify)
     try:
         # Raise exceptions for HTTP errors
@@ -83,6 +83,10 @@ def requests_workaround_post(url, json, headers, auth, verify=False):
     return requests_workaround('POST', url, json, headers, auth, verify=False)
 
 
+def requests_workaround_patch(url, json, headers, auth, verify=False):
+    return requests_workaround('PATCH', url, json, headers, auth, verify=False)
+
+
 def requests_workaround_get(url, headers, auth, verify=False):
     return requests_workaround('GET', url, None, headers, auth, verify=False)
 
@@ -201,7 +205,7 @@ class Device(Transport):
     def get_datastore(self, datastore="operational", path="", parse=True):
         """Get a datastore"""
         dspath = f"/ds/ietf-datastores:{datastore}"
-        if path is not None:
+        if path is not None and path != "":
             dspath = f"{dspath}{path}"
 
         url = f"{self.restconf_url}{dspath}"
@@ -254,36 +258,102 @@ class Device(Transport):
 
     def get_config_dict(self, modname):
         """Get all configuration for module @modname as dictionary"""
-        ds = self.get_running(modname)
+        # Strip leading slash if present (for compatibility with NETCONF xpath style)
+        modname = modname.lstrip('/')
+
+        # Get the whole module's configuration by requesting the module root
+        # The modname parameter might be in format "module:container" or just "module"
+        if ":" in modname:
+            model, container = modname.split(":", 1)  # Split only on first colon
+            path = f"/{model}:{container}"  # This creates something like /ietf-syslog:syslog
+        else:
+            # If no colon, assume the whole thing is the module name
+            model = modname
+            path = f"/{model}"  # This creates something like /ietf-syslog
+
+        ds = self.get_running(path)
+        if ds is None:
+            return None
         ds = json.loads(ds.print_mem("json", with_siblings=True, pretty=False))
-        model, container = modname.split(":")
-        for k, v in ds.items():
-            return {container: v}
+        # If we have module:container format, extract the container part from the result
+        if ":" in modname:
+            _, container = modname.split(":", 1)
+            for k, v in ds.items():
+                return {container: v}
+        else:
+            # Return the whole result if no specific container was specified
+            return ds
 
-    def put_config_dicts(self, models):
-        """PUT full configuration of all models to running-config"""
+    def put_config_dicts(self, models, retries=3):
+        """PATCH configuration of all models to running-config
+
+        Uses candidate datastore + copy to running to trigger sysrepo
+        change callbacks, similar to how NETCONF edit-config + commit works.
+
+        Args:
+            models: Dictionary of models to configure
+            retries: Number of retry attempts on failure (default 3)
+        """
         infer_put_dict(self.name, models)
-        running = self.get_running()
 
-        for model in models.keys():
+        # Copy running to candidate first (to preserve existing config)
+        self.copy("running", "candidate")
+
+        # PATCH each model to candidate datastore
+        for model, config in models.items():
             try:
                 mod = self.lyctx.get_module(model)
             except libyang.util.LibyangError:
                 raise Exception(f"YANG model '{model}' not found on device. "
                                f"Model may not be installed or enabled. "
                                f"Available models can be checked with get_schema_list()") from None
-            lyd = mod.parse_data_dict(models[model], no_state=True, validate=False)
-            running.merge(lyd)
 
-        cfg = running.print_mem("json", with_siblings=True, pretty=True)
-        # print(f"PUT new running-config: {cfg}")
-        return self.put_datastore("running", json.loads(cfg))
+            # Parse and convert to get proper structure with module prefix
+            lyd = mod.parse_data_dict(config, no_state=True, validate=False)
+            patch_data = json.loads(lyd.print_mem("json", with_siblings=True, pretty=False))
 
-    def put_config_dict(self, modname, edit):
-        """Add @edit to running config and put the whole configuration"""
+            # PATCH to candidate datastore
+            url = f"{self.restconf_url}/ds/ietf-datastores:candidate"
 
-        # This is hacky, refactor when rousette have PATCH support.
-        running = self.get_running()
+            last_error = None
+            for attempt in range(0, retries):
+                try:
+                    response = requests_workaround_patch(
+                        url,
+                        json=patch_data,
+                        headers=self.headers,
+                        auth=self.auth,
+                        verify=False
+                    )
+                    response.raise_for_status()
+                    last_error = None
+                    break
+                except Exception as e:
+                    last_error = e
+                    if attempt < retries - 1:
+                        print(f"Failed PATCH to {url}: {e}  Retrying ...")
+                        time.sleep(1)
+                    else:
+                        print(f"Failed PATCH to {url}: {e}")
+                    continue
+
+            if last_error is not None:
+                raise last_error
+
+        # Copy candidate to running (acts as "commit", triggers sysrepo callbacks)
+        self.copy("candidate", "running")
+
+    def put_config_dict(self, modname, edit, retries=3):
+        """PATCH configuration for a single model to running-config
+
+        Uses candidate datastore + copy to running to trigger sysrepo
+        change callbacks, similar to how NETCONF edit-config + commit works.
+
+        Args:
+            modname: YANG module name
+            edit: Configuration dictionary
+            retries: Number of retry attempts on failure (default 3)
+        """
         try:
             mod = self.lyctx.get_module(modname)
         except libyang.util.LibyangError:
@@ -291,26 +361,42 @@ class Device(Transport):
                            f"Model may not be installed or enabled. "
                            f"Available models can be checked with get_schema_list()") from None
 
-        for k, _ in edit.items():
-            module = modname + ":" + k
-            break
+        # Copy running to candidate first (to preserve existing config)
+        self.copy("running", "candidate")
 
-        # Ugly hack, but this function should be refactored when patch
-        # is available in rousette anyway.
-        rundict = json.loads(running.print_mem("json", with_siblings=True,
-                                               pretty=False))
-        if rundict.get(module) is None:
-            rundict[module] = {}
-            running = self.lyctx.parse_data_mem(json.dumps(rundict), "json",
-                                                parse_only=True)
+        # Parse and convert to get proper structure with module prefix
+        lyd = mod.parse_data_dict(edit, no_state=True, validate=False)
+        patch_data = json.loads(lyd.print_mem("json", with_siblings=True, pretty=False))
 
-        change = mod.parse_data_dict(edit, no_state=True, validate=False)
-        running.merge_module(change)
-        cfg = running.print_mem("json", with_siblings=True, pretty=False)
-        # print(f"PUT new running-config: {cfg}")
-        data = json.loads(cfg)
+        # PATCH to candidate datastore
+        url = f"{self.restconf_url}/ds/ietf-datastores:candidate"
+        last_error = None
+        for attempt in range(0, retries):
+            try:
+                response = requests_workaround_patch(
+                    url,
+                    json=patch_data,
+                    headers=self.headers,
+                    auth=self.auth,
+                    verify=False
+                )
+                response.raise_for_status()
+                last_error = None
+                break
+            except Exception as e:
+                last_error = e
+                if attempt < retries - 1:
+                    print(f"Failed PATCH to {url}: {e}  Retrying ...")
+                    time.sleep(1)
+                else:
+                    print(f"Failed PATCH to {url}: {e}")
+                continue
 
-        return self.put_datastore("running", data)
+        if last_error is not None:
+            raise last_error
+
+        # Copy candidate to running (acts as "commit", triggers sysrepo callbacks)
+        self.copy("candidate", "running")
 
     def call_dict(self, model, call):
         pass # Need implementation