Merge pull request #723 from kernelkit/verify-iface-status

Verify Interface Status

test/case/ietf_interfaces/Readme.adoc

@@ -38,3 +38,5 @@ include::static_multicast_filters/Readme.adoc[]
 include::vlan_qos/Readme.adoc[]
 
 include::verify_all_interface_types/Readme.adoc[]
+
+include::iface_enable_disable/Readme.adoc[]

test/case/ietf_interfaces/ietf_interfaces.yaml

@@ -55,3 +55,6 @@
 
 - name: verify_all_interface_types
   case: verify_all_interface_types/test.py
+
+- name: iface_enable_disable
+  case: iface_enable_disable/test.py

test/case/ietf_interfaces/iface_enable_disable/Readme.adoc

@@ -0,0 +1,31 @@
+=== Interface status
+==== Description
+Verify interface status properly propagate changes when an interface
+is disabled and then re-enabled.
+
+Both admin-status and oper-status are verified.
+
+==== Topology
+ifdef::topdoc[]
+image::../../test/case/ietf_interfaces/iface_enable_disable/topology.png[Interface status topology]
+endif::topdoc[]
+ifndef::topdoc[]
+ifdef::testgroup[]
+image::iface_enable_disable/topology.png[Interface status topology]
+endif::testgroup[]
+ifndef::testgroup[]
+image::topology.png[Interface status topology]
+endif::testgroup[]
+endif::topdoc[]
+==== Test sequence
+. Set up topology and attach to target DUTs
+. Configure bridge and associated interfaces in target1
+. Disable interface in target2
+. Verify the interface is disabled
+. Enable the interface and assign an IP address
+. Verify the interface is enabled
+. Verify it is possible to ping the interface
+
+
+<<<
+

test/case/ietf_interfaces/iface_enable_disable/test.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""
+Interface status
+
+Verify interface status properly propagate changes when an interface
+is disabled and then re-enabled.
+
+Both admin-status and oper-status are verified.
+"""
+
+import infamy
+import infamy.iface as iface
+
+from infamy import until
+
+def print_error_message(iface, param, exp_val, act_val):
+    return f"'{param}' failure for interface '{iface}'. Expected '{exp_val}', Actual: '{act_val}'"
+
+def assert_param(target, interface, parameter, expected_value):
+    def check_param():
+        actual_value = iface.get_param(target, interface, parameter)
+        if actual_value is None:
+            raise ValueError(f"Failed to retrieve '{parameter}' for interface '{interface}'")
+        return actual_value == expected_value
+
+    until(check_param)
+
+    actual_value = iface.get_param(target, interface, parameter)
+    assert (expected_value == actual_value), print_error_message(
+        iface=interface,
+        param = parameter,
+        exp_val = expected_value,
+        act_val = actual_value
+        )
+
+def configure_interface(target, iface_name, iface_type=None, enabled=True, ip_address=None, bridge=None):
+
+    interface_config = {
+        "name": iface_name,
+        "enabled": enabled
+    }
+
+    if iface_type:
+        interface_config["type"] = iface_type
+
+    if ip_address:
+        interface_config["ipv4"] = {
+            "address": [
+                {
+                    "ip": ip_address, 
+                    "prefix-length": 24
+                }]}
+
+    if bridge:
+        interface_config["infix-interfaces:bridge-port"] = {
+            "bridge": bridge
+        }
+
+    target.put_config_dict( "ietf-interfaces", {
+        "interfaces": {
+            "interface": [
+                interface_config
+            ]}})
+
+with infamy.Test() as test:
+    with test.step("Set up topology and attach to target DUTs"):
+        env = infamy.Env()
+        target1 = env.attach("target1", "mgmt")
+        target2 = env.attach("target2", "mgmt")
+
+        _, data1 = env.ltop.xlate("target1", "data")
+        _, link1 = env.ltop.xlate("target1", "link")
+
+        _, iface_under_test = env.ltop.xlate("target2", "link")
+        _, host_send_iface = env.ltop.xlate("host", "data")
+        _bridge = "br_0"
+
+        target_address = "10.10.10.2"
+        host_address = "10.10.10.1"        
+
+    with test.step("Configure bridge and associated interfaces in target1"):
+        configure_interface(target1, _bridge, enabled=True, iface_type="infix-if-type:bridge")
+        configure_interface(target1, data1, enabled=True, bridge=_bridge)
+        configure_interface(target1, link1, enabled=True, bridge=_bridge)
+
+    with test.step("Disable interface in target2"):
+        configure_interface(target2, iface_under_test, enabled=False)
+
+    with test.step("Verify the interface is disabled"):
+        assert_param(target2, iface_under_test, "admin-status", "down")
+        assert_param(target2, iface_under_test, "oper-status", "down")
+
+    with test.step("Enable the interface and assign an IP address"):
+        configure_interface(target2, iface_under_test, enabled=True, ip_address=target_address)
+
+    with test.step("Verify the interface is enabled"):
+        assert_param(target2, iface_under_test, "admin-status", "up")
+        assert_param(target2, iface_under_test, "oper-status", "up")
+
+    with infamy.IsolatedMacVlan(host_send_iface) as send_ns:
+        with test.step("Verify it is possible to ping the interface"):
+            send_ns.addip(host_address)
+            send_ns.must_reach(target_address)
+
+    test.succeed()

test/case/ietf_interfaces/iface_enable_disable/topology.dot

@@ -0,0 +1,35 @@
+graph "2x4" {
+	layout="neato";
+	overlap="false";
+	esep="+40";
+
+        node [shape=record, fontname="monospace"];
+	edge [color="cornflowerblue", penwidth="2"];
+
+	host [
+	    label="host | { <mgmt1> mgmt1 | <data> data | <mgmt2> mgmt2 }",
+	    pos="0,15!",
+	    kind="controller",
+	];
+
+        target1 [
+	    label="{ <mgmt> mgmt | <data> data | <link> link } | { \n dut1 \n\n }",
+	    pos="8,15!",
+
+	    kind="infix",
+	];
+
+        target2 [
+            label="{ <link> link | <mgmt> mgmt } | { \n dut2 \n\n }",
+	    pos="8,12!",
+
+	    kind="infix",
+	];
+
+	host:mgmt1 -- target1:mgmt [kind=mgmt, color="lightgrey"]
+	host:data -- target1:data [color=black, fontcolor=black, taillabel="10.10.10.1/24"]
+
+	host:mgmt2 -- target2:mgmt [kind=mgmt, color="lightgrey"]
+
+	target1:link -- target2:link [color=black, fontcolor=black, headlabel="10.10.10.2/24"]
+}

test/case/ietf_interfaces/iface_phys_address/test.py

@@ -30,7 +30,7 @@ def calc_mac(base_mac, mac_offset):
 def reset_mac(tgt, port, mac):
     """Reset DUT interface MAC address to default."""
     node = "infix-interfaces:custom-phys-address"
-    xpath = iface.get_iface_xpath(port, node)
+    xpath = iface.get_xpath(port, node)
     tgt.delete_xpath(xpath)
     with test.step("Verify target:data MAC address is reset to default"):
         until(lambda: iface.get_phys_address(tgt, tport) == mac)

test/case/ietf_interfaces/verify_all_interface_types/test.py

@@ -22,10 +22,10 @@
 
 
 def verify_interface(target, interface, expected_type):
-    assert iface.interface_exist(target, interface), f"Interface <{interface}> does not exist."
+    assert iface.exist(target, interface), f"Interface <{interface}> does not exist."
 
     expected_type = f"infix-if-type:{expected_type}"
-    actual_type = iface._iface_get_param(target, interface, "type")
+    actual_type = iface.get_param(target, interface, "type")
 
     if expected_type == "infix-if-type:etherlike"  and actual_type == "infix-if-type:ethernet":
         return  # Allow 'etherlike' to match 'ethernet'

test/case/ietf_interfaces/veth_delete/test.py

@@ -51,9 +51,9 @@
         })
 
     with test.step("Verify interfaces 'veth0a' and 'veth0b' exists"):
-        assert iface.interface_exist(target, veth0a), \
+        assert iface.exist(target, veth0a), \
             f"Interface <{veth0a}> does not exist."
-        assert iface.interface_exist(target, veth0b), \
+        assert iface.exist(target, veth0b), \
             f"Interface <{veth0b}> does not exist."
 
     with test.step("Set IP address on target:eth0 (dummy op)"):
@@ -97,15 +97,15 @@
         target = env.attach("target", "mgmt")
 
     with test.step("Verify target:eth0 and target:eth1 still exist"):
-        assert iface.interface_exist(target, eth0), \
+        assert iface.exist(target, eth0), \
             f"Interface {eth0} missing!"
-        assert iface.interface_exist(target, eth1), \
+        assert iface.exist(target, eth1), \
             f"Interface {eth1} missing!"
 
     with test.step("Verify VETH pair have been removed"):
-        assert not iface.interface_exist(target, veth0a), \
+        assert not iface.exist(target, veth0a), \
             f"Interface <{veth0a}> still exists!"
-        assert not iface.interface_exist(target, veth0b), \
+        assert not iface.exist(target, veth0b), \
             f"Interface <{veth0b}> still exists!"
 
     test.succeed()

test/case/ietf_interfaces/vlan_ping/test.py

@@ -4,26 +4,31 @@
 
 Very basic test if the VLAN interface configuration works.
 """
+
 import infamy
 import infamy.iface as iface
-import copy
 
 from infamy import until
 
 def test_ping(hport, should_pass):
-      with infamy.IsolatedMacVlan(hport) as ns:
-            pingtest = ns.runsh("""
-            set -ex
+    with infamy.IsolatedMacVlan(hport) as ns:
+        try:
+            ns.runsh("""
+                  set -ex
+                  ip link set iface up
+                  ip link add dev vlan10 link iface up type vlan id 10
+                  ip addr add 10.0.0.1/24 dev vlan10
+                  """)
 
-            ip link set iface up
-            ip link add dev vlan10 link iface up type vlan id 10
-            ip addr add 10.0.0.1/24 dev vlan10
-            """)
-            if(should_pass):
+            if should_pass:
                 ns.must_reach("10.0.0.2")
             else:
                 ns.must_not_reach("10.0.0.2")
 
+        except Exception as e:
+            print(f"An error occurred during the VLAN setup or ping test: {e}")
+            raise
+
 with infamy.Test() as test:
       with test.step("Set up topology and attach to target DUT"):
             env = infamy.Env()
@@ -60,7 +65,7 @@ def test_ping(hport, should_pass):
             })
 
       with test.step("Waiting for links to come up"):
-            until(lambda: iface.get_oper_up(target, tport))
+            until(lambda: iface.get_param(target, tport, "oper-status") == "up")
 
       with test.step("Ping 10.0.0.2 from VLAN 10 on host:data with IP 10.0.0.1"):
             _, hport = env.ltop.xlate("host", "data")

test/infamy/iface.py

@@ -2,14 +2,14 @@
 Fetch interface status from remote device.
 """
 
-def get_iface_xpath(iface, path=None):
+def get_xpath(iface, path=None):
     """Compose complete XPath to a YANG node in /ietf-interfaces"""
     xpath=f"/ietf-interfaces:interfaces/interface[name='{iface}']"
     if not path is None:
         xpath=f"{xpath}/{path}"
     return xpath
 
-def _iface_extract_param(json_content, param):
+def _extract_param(json_content, param):
     """Returns (extracted) value for parameter 'param'"""
     interfaces = json_content.get('interfaces')
     if not interfaces:
@@ -22,16 +22,16 @@ def _iface_extract_param(json_content, param):
 
     return None
 
-def _iface_get_param(target, iface, param=None):
+def get_param(target, iface, param=None):
     """Fetch target dict for iface and extract param from JSON"""
-    content = target.get_data(get_iface_xpath(iface, param))
+    content = target.get_data(get_xpath(iface, param))
     if content is None:
         return None
-    return _iface_extract_param(content, param)
+    return _extract_param(content, param)
 
-def interface_exist(target, iface):
+def exist(target, iface):
     """Verify that the target interface exists"""
-    return _iface_get_param(target, iface, "name") is not None
+    return get_param(target, iface, "name") is not None
 
 def address_exist(target, iface, address, prefix_length = 24, proto="dhcp"):
     """Check if 'address' is set on iface"""
@@ -56,47 +56,9 @@ def get_ipv4_address(target, iface):
         return None
     return ipv4['address']
 
-def get_if_index(target, iface):
-    """Fetch interface 'if-index' (operational status)"""
-    return _iface_get_param(target, iface, "if-index")
-
-def get_oper_status(target, iface):
-    """Fetch interface 'oper-status' (operational status)"""
-    return _iface_get_param(target, iface, "oper-status")
-
 def get_phys_address(target, iface):
     """Fetch interface MAC address (operational status)"""
-    return _iface_get_param(target, iface, "phys-address")
-
-def get_oper_up(target,iface):
-    state=get_oper_status(target,iface)
-    return state == "up"
-
-def print_iface(target, iface):
-    data = target.get_data(_iface_xpath(iface, None))
-    print(data)
-
-def print_all(target):
-    """Print status parameters for all target interfaces"""
-    try:
-        content = target.get_dict("/ietf-interfaces:interfaces")
-        interfaces = content.get('interfaces')
-        if interfaces:
-            interface_list = interfaces.get('interface')
-            if interface_list and isinstance(interface_list, list):
-                col1 = "name"
-                col2 = "if-index"
-                col3 = "oper-status"
-                print('-'*36)
-                print(f"{col1: <12}{col2: <12}{col3: <12}")
-                print('-'*36)
-                for interface in interface_list:
-                    print(f"{interface['name']: <12}"
-                          f"{interface['if-index']: <12}"
-                          f"{interface['oper-status']: <12}")
-                print('-'*36)
-    except:
-        print(f"Failed to get interfaces' status from target {target}")
+    return get_param(target, iface, "phys-address")
 
 def exist_bridge_multicast_filter(target, group, iface, bridge):
     # The interface array is different in restconf/netconf, netconf has a keyed list but

test/infamy/netconf.py

@@ -374,7 +374,7 @@ def get_schema(self, schema, outdir):
 
     def get_iface(self, name):
         """Fetch target dict for iface and extract param from JSON"""
-        content = self.get_data(iface.get_iface_xpath(name))
+        content = self.get_data(iface.get_xpath(name))
         interface = content.get("interfaces", {}).get("interface", None)
 
         if interface is None: