test: Verify LLDP admin status

test/case/infix_services/Readme.adoc

@@ -9,6 +9,7 @@ include::mdns_allow_deny/Readme.adoc[]
 
 include::lldp_enable_disable/Readme.adoc[]
 
+include::lldp_admin_status/Readme.adoc[]
 
 include::ssh_server_config/Readme.adoc[]
 

test/case/infix_services/infix_services.yaml

@@ -2,6 +2,8 @@
 - name: lldp_enable_disable
   case: lldp_enable_disable/test.py
 
+- name: lldp_admin_status
+  case: lldp_admin_status/test.py
 
 - name: mdns_enable_disable
   case: mdns_enable_disable/test.py

test/case/infix_services/lldp_admin_status/Readme.adoc

@@ -0,0 +1 @@
+lldp_admin_status.adoc

test/case/infix_services/lldp_admin_status/lldp_admin_status.adoc

@@ -0,0 +1,26 @@
+=== LLDP admin status
+==== Description
+Verify that LLDP admin status is set properly by lldpd
+
+==== Topology
+ifdef::topdoc[]
+image::{topdoc}../../test/case/infix_services/lldp_admin_status/topology.svg[LLDP admin status topology]
+endif::topdoc[]
+ifndef::topdoc[]
+ifdef::testgroup[]
+image::lldp_admin_status/topology.svg[LLDP admin status topology]
+endif::testgroup[]
+ifndef::testgroup[]
+image::topology.svg[LLDP admin status topology]
+endif::testgroup[]
+endif::topdoc[]
+==== Test sequence
+. Enable target interface and enable LLDP
+. Verify admin-status: 'rx-only'
+. Verify admin-status: 'tx-only'
+. Verify admin-status: 'disabled'
+. Verify admin-status: 'tx-and-rx'
+
+
+<<<
+

test/case/infix_services/lldp_admin_status/test.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""LLDP admin status
+
+Verify that LLDP admin status is set properly by lldpd
+
+"""
+import time
+import infamy
+import infamy.lldp as lldp
+
+from scapy.all import Ether, sendp
+from scapy.contrib.lldp import (
+    LLDPDU, LLDPDUChassisID, LLDPDUPortID, LLDPDUTimeToLive, LLDPDUEndOfLLDPDU
+)
+
+def capture_traffic(iface, sec):
+    with infamy.IsolatedMacVlan(iface) as netns:
+        sniffer = infamy.Sniffer(netns, "ether proto 0x88cc")               
+        with sniffer:
+            print("Capturing network traffic ...")
+            time.sleep(sec)
+        return sniffer.output()
+
+def send_lldp_packet(iface, chassis_id, chassis_id_subtype, ttl=3):
+    eth = Ether(dst="01:80:c2:00:00:0e", type=0x88cc)
+    lldpdu = eth / LLDPDU() 
+    lldpdu /= LLDPDUChassisID(subtype=chassis_id_subtype, id=chassis_id)
+    lldpdu /= LLDPDUPortID(subtype=5, id=iface) 
+    lldpdu /= LLDPDUTimeToLive(ttl=ttl) / LLDPDUEndOfLLDPDU()
+    sendp(lldpdu, iface=iface, verbose=False)
+
+def verify_neigh_presence(test, target, port, expect_neighbor):
+    """Verify neighbor (host) presence on the target system"""
+    neighbors = lldp.get_remote_systems_data(target, port)
+    if expect_neighbor and not neighbors:
+        print("Expected LLDP neighbor but found none.")
+        test.fail()
+    if not expect_neighbor and neighbors:
+        print("Unexpected LLDP neighbor found.")
+        test.fail()    
+
+def verify_admin_status(test, target, port, admin_status, local_capture, remote_detect):
+    target.put_config_dicts({
+        "ieee802-dot1ab-lldp": {
+            "lldp": {
+                "port": [{
+                    "name": target["data"],
+                    "dest-mac-address": "00-00-00-00-00-00",
+                    "admin-status": admin_status
+                }]
+            }
+        }
+    })
+
+    rc = capture_traffic(port, 5)
+
+    if local_capture and "LLDP" not in rc.stdout:
+        test.fail()
+    if not local_capture and "LLDP" in rc.stdout:
+        test.fail()
+
+    send_lldp_packet(port, "Chassis ID 007", 7)
+    verify_neigh_presence(test, target, target["data"], remote_detect)
+
+with infamy.Test() as test:
+    env = infamy.Env()
+    target = env.attach("target", "mgmt")
+
+    lldp_link = env.ltop.get_link("host", "target", flt=lambda e: "ieee-mc" in e.get("requires", "").split())
+    if not lldp_link:
+        print("Skipping test: No link expecting 'ieee-mc' found in the topology.")
+        test.skip()
+
+    log_host_port, _ = lldp_link
+    _, hport = env.ltop.xlate("host", log_host_port)
+
+    test.step("Enable target interface and enable LLDP")
+    target.put_config_dicts({
+        "ietf-interfaces": {
+            "interfaces": {
+                "interface": [{
+                    "name": target["data"], 
+                    "enabled": True
+                }]
+            }
+        }
+    })
+
+    target.put_config_dicts({
+        "ieee802-dot1ab-lldp": {
+            "lldp": {
+                "enabled": True, 
+                "message-tx-interval": 1
+            }
+        }
+    })
+
+    with test.step("Verify admin-status: 'rx-only'"):
+        verify_admin_status(test, target, hport, "rx-only", False, True)
+    with test.step("Verify admin-status: 'tx-only'"):
+        verify_admin_status(test, target, hport, "tx-only", True, False)
+    with test.step("Verify admin-status: 'disabled'"):
+        verify_admin_status(test, target, hport, "disabled", False, False)
+    with test.step("Verify admin-status: 'tx-and-rx'"): 
+        verify_admin_status(test, target, hport, "tx-and-rx", True, True)
+
+    test.succeed()

test/case/infix_services/lldp_admin_status/topology.dot

@@ -0,0 +1,23 @@
+graph "1x2" {
+	layout="neato";
+	overlap="false";
+	esep="+80";
+
+        node [shape=record, fontname="DejaVu Sans Mono, Book"];
+	edge [color="cornflowerblue", penwidth="2", fontname="DejaVu Serif, Book"];
+
+	host [
+	    label="host | { <mgmt> mgmt | <data> data }",
+	    pos="0,12!",
+	    requires="controller",
+	];
+
+        target [
+	    label="{ <mgmt> mgmt | <data> data } | target",
+	    pos="10,12!",
+	    requires="infix",
+	];
+
+	host:mgmt -- target:mgmt [requires="mgmt", color="lightgrey"]
+	host:data -- target:data [color=black, fontcolor=black, fontsize=12, taillabel="10.0.0.1/24", headlabel="10.0.0.10/24", requires="ieee-mc"]
+}

test/infamy/lldp.py

@@ -0,0 +1,40 @@
+"""
+Fetch LLDP local system data from remote device.
+"""
+
+reverse_subtype_mapping = {
+    "chassis-component": 1,
+    "interface-alias": 2,
+    "port-component": 3,
+    "mac-address": 4,
+    "network-address": 5,
+    "interface-name": 6,
+    "local": 7  # 'local' maps to 7 as per LLDP spec
+}
+
+def get_remote_systems_data(target, port):
+    """Fetch the full remote-systems-data list for a specific port"""
+    content = target.get_data("/ieee802-dot1ab-lldp:lldp")
+
+    if not content:
+        return []
+
+    for port_entry in content.get("lldp", {}).get("port", []):
+        if port_entry.get("name") == port:
+            return port_entry.get("remote-systems-data", [])
+
+    return []
+
+def get_chassis_ids(target, port):
+    """Fetch all LLDP chassis IDs for neighbors on a specific port"""
+    neighbors = get_remote_systems_data(target, port)
+    return [neighbor.get("chassis-id") for neighbor in neighbors if "chassis-id" in neighbor]
+
+def get_chassis_ids_subtype(target, port):
+    """Fetch all LLDP chassis ID subtypes for neighbors on a specific port and convert them to numbers."""
+    neighbors = get_remote_systems_data(target, port)
+
+    return [
+        reverse_subtype_mapping.get(neighbor.get("chassis-id-subtype"), 0)  # Default to 0 if unknown
+        for neighbor in neighbors if "chassis-id-subtype" in neighbor
+    ]