infamy: Adapt for operational changes between netconf and restconf

test/case/ietf_hardware/usb.py

@@ -82,7 +82,10 @@ with infamy.Test() as test:
             until(lambda: usb.get_usb_state(target, available[0]) == "unlocked")
 
     with test.step("Remove all hardware configuration"):
-        target.delete_xpath("/ietf-hardware:hardware/component")
+        xpath=target.get_xpath("/ietf-hardware:hardware/component", "name", "USB")
+        target.delete_xpath(xpath)
+        xpath=target.get_xpath("/ietf-hardware:hardware/component", "name", "USB2")
+        target.delete_xpath(xpath)
 
     with test.step("Verify USB ports locked"):
         for port in available:
@@ -112,7 +115,10 @@ with infamy.Test() as test:
             until(lambda: usb.get_usb_state(target, port) == "unlocked")
 
     with test.step("Remove USB configuration, and reboot"):
-        target.delete_xpath("/ietf-hardware:hardware/component")
+        xpath=target.get_xpath("/ietf-hardware:hardware/component", "name", "USB")
+        target.delete_xpath(xpath)
+        xpath=target.get_xpath("/ietf-hardware:hardware/component", "name", "USB2")
+        target.delete_xpath(xpath)
         target.copy("running", "startup")
         target.reboot()
         if wait_boot(target) == False:

test/infamy/iface.py

@@ -2,13 +2,6 @@
 Fetch interface status from remote device.
 """
 
-def _iface_xpath(iface, path=None):
-    """Compose complete XPath to a YANG node in /ietf-interfaces"""
-    xpath = f"/ietf-interfaces:interfaces/interface[name='{iface}']"
-    if path:
-        xpath.join(f"/{path}")
-    return xpath
-
 def _iface_extract_param(json_content, param):
     """Returns (extracted) value for parameter 'param'"""
     interfaces = json_content.get('interfaces')
@@ -24,11 +17,8 @@ def _iface_extract_param(json_content, param):
 
 def _iface_get_param(target, iface, param=None):
     """Fetch target dict for iface and extract param from JSON"""
-    try:
-        content = target.get_data(_iface_xpath(iface, param))
-        return _iface_extract_param(content, param)
-    except:
-        return None
+    content = target.get_data(target.get_iface_xpath(iface, param))
+    return _iface_extract_param(content, param)
 
 def interface_exist(target, iface):
     """Verify that the target interface exists"""
@@ -45,7 +35,14 @@ def address_exist(target, iface, address, prefix_length = 24, proto="dhcp"):
 
 def get_ipv4_address(target, iface):
     """Fetch interface IPv4 addresses from (operational status)"""
-    ipv4 = _iface_get_param(target, iface, "ipv4")
+    # The interface array is different in restconf/netconf, netconf has a keyed list but
+    # restconf has a numbered list, i think i read that this was a bug in rousette, but
+    # have not found it.
+    interface=target.get_iface(iface)
+    if interface is None:
+        raise "Interface not found"
+
+    ipv4 = interface.get("ipv4") or interface.get("ietf-ip:ipv4")
     if ipv4 is None or 'address' not in ipv4:
         return None
     return ipv4['address']
@@ -93,8 +90,18 @@ def print_all(target):
         print(f"Failed to get interfaces' status from target {target}")
 
 def exist_bridge_multicast_filter(target, group, iface, bridge):
-    bridge = _iface_get_param(target, bridge, "bridge")
-    for filter in bridge.get("multicast-filters", {}).get("multicast-filter", {}):
+    # The interface array is different in restconf/netconf, netconf has a keyed list but
+    # restconf has a numbered list, i think i read that this was a bug in rousette, but
+    # have not found it.
+    interface=target.get_iface(bridge)
+    if interface is None:
+        raise "Interface not found"
+
+    brif = interface.get("bridge") or interface.get("infix-interfaces:bridge")
+    if brif is None:
+        return False
+
+    for filter in brif.get("multicast-filters", {}).get("multicast-filter", {}):
         if filter.get("group") == group:
             for p in filter.get("ports"):
                 if p["port"] == iface:

test/infamy/route.py

@@ -11,7 +11,9 @@ def _exist_route(target, prefix, nexthop, version, source_protocol):
     routes = _get_routes(target, version)
     route_found = False
     for r in routes:
-        if r["destination-prefix"] != prefix:
+        # netconf presents destination-prefix, restconf prefix with model
+        p=r.get("destination-prefix") or (version == "ipv4" and r.get("ietf-ipv4-unicast-routing:destination-prefix")) or (version == "ipv6" and r.get("ietf-ipv6-unicast-routing:destination-prefix"))
+        if p != prefix:
             continue
 
         if source_protocol and r.get("source-protocol") != source_protocol:
@@ -27,7 +29,9 @@ def _exist_route(target, prefix, nexthop, version, source_protocol):
                     if address == nexthop:
                         return True
             else:
-                if nh["next-hop-address"] == nexthop:
+                # netconf presents next-hop-address, restconf prefix with model    ietf-ipv4-unicast-routing:next-hop-address
+                nh_addr=nh.get("next-hop-address", None) or (version == "ipv4" and nh.get("ietf-ipv4-unicast-routing:next-hop-address", None)) or (version == "ipv6" and nh.get("ietf-ipv6-unicast-routing:next-hop-address", None))
+                if nh_addr == nexthop:
                     return True
             return False
     return route_found
@@ -44,7 +48,7 @@ def _get_ospf_status(target):
     rib = target.get_data(xpath)["routing"]["control-plane-protocols"].get("control-plane-protocol", {})
     for p in rib:
         if p["type"] == "ietf-ospf:ospfv2":
-            return p["ospf"]
+            return p.get("ospf") or p.get("ietf-ospf:ospf")
 
 def _get_ospf_status_area(target, area_id):
     ospf=_get_ospf_status(target)