test/infamy: new method, delete_xpaths()

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

test/infamy/netconf.py

@@ -455,3 +455,47 @@ class Device(Transport):
         # Apply the configuration change
         return self.put_config(lyd.print_mem("xml", with_siblings=True,
                                              pretty=False))
+
+    def delete_xpaths(self, xpaths):
+        """Delete several xpaths in a single transaction.
+
+        Needed when the targets reference one another, e.g. both ends of a
+        VETH pair (mutually mandatory leafrefs), and so cannot be removed
+        one at a time.
+        """
+        pattern = r"^/(?P<module>[^:]+):(?P<path>[^/]+)"
+
+        # Group xpaths by their top-level module/container so each is
+        # removed from a single fetched config tree (one diff per module).
+        groups = {}
+        order = []
+        for xpath in xpaths:
+            coverage.track_xpath(xpath)
+            match = re.search(pattern, xpath)
+            if not match:
+                raise ValueError(f"Failed parsing xpath:{xpath}")
+            modpath = f"/{match.group('module')}:{match.group('path')}"
+            if modpath not in groups:
+                groups[modpath] = (match.group('module'), [])
+                order.append(modpath)
+            groups[modpath][1].append(xpath)
+
+        edit = ""
+        for modpath in order:
+            module, paths = groups[modpath]
+            old = self.get_config_dict(modpath)
+            new = copy.deepcopy(old)
+            for xpath in paths:
+                if not libyang.xpath_del(new, xpath):
+                    raise ValueError(f"Failed to delete specified xpath: {xpath}")
+
+            mod = self.ly.get_module(module)
+            oldd = mod.parse_data_dict(old, no_state=True, validate=False)
+            newd = mod.parse_data_dict(new, no_state=True, validate=False)
+
+            lyd = oldd.diff(newd)
+            if lyd is None:
+                raise ValueError(f"Failed generating diff for {modpath}")
+            edit += lyd.print_mem("xml", with_siblings=True, pretty=False)
+
+        return self.put_config(edit)

test/infamy/restconf.py

@@ -502,3 +502,23 @@ class Device(Transport):
         response.raise_for_status()
 
         return True
+
+    def delete_xpaths(self, xpaths):
+        """Delete several xpaths in a single transaction.
+
+        Needed when the targets reference one another, e.g. both ends of a
+        VETH pair (mutually mandatory leafrefs), and so cannot be removed
+        one at a time.  Stage the deletions in the candidate datastore, then
+        copy it to running so they commit and validate together.
+        """
+        self.copy("running", "candidate")
+        for xpath in xpaths:
+            coverage.track_xpath(xpath)
+            path = f"/ds/ietf-datastores:candidate{xpath_to_uri(xpath)}"
+            url = f"{self.restconf_url}{path}"
+            response = requests_workaround_delete(url, headers=self.headers,
+                                                  auth=self.auth, verify=False)
+            response.raise_for_status()
+        self.copy("candidate", "running")
+
+        return True

test/infamy/transport.py

@@ -34,6 +34,10 @@ class Transport(ABC):
     def delete_xpath(self, xpath):
         pass
 
+    @abstractmethod
+    def delete_xpaths(self, xpaths):
+        pass
+
     @abstractmethod
     def copy(self, source, target):
         pass