From dca9263776ce33c3469d5274b58b7443c7c6a69a Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Wed, 19 Aug 2026 11:24:35 -0700 Subject: [PATCH 01/10] feat: add VPP notification test cases Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 533 +++++++++++++++++++++++++ 1 file changed, 533 insertions(+) create mode 100644 test/sai_test/sai_notification_test.py diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py new file mode 100644 index 00000000..4bb06b4f --- /dev/null +++ b/test/sai_test/sai_notification_test.py @@ -0,0 +1,533 @@ +# Copyright (c) 2026 Microsoft Open Technologies, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# THIS CODE IS PROVIDED ON AN *AS IS* BASIS, WITHOUT WARRANTIES OR +# CONDITIONS OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT +# LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS +# FOR A PARTICULAR PURPOSE, MERCHANTABILITY OR NON-INFRINGEMENT. + +"""Opt-in VPP SAI notification tests.""" + +import re +import os +import subprocess +import threading +import time + +from ptf import config as ptf_config +from ptf.testutils import test_params_get +from unittest import SkipTest + +from scapy.all import Ether, IP, UDP, get_if_hwaddr, sendp, sniff +from scapy.contrib.bfd import BFD + +from sai_thrift.sai_adapter import * +from sai_test_base import T0TestBase +from sai_utils import sai_ipaddress, sai_ipprefix + + +PORT_NOTIFICATION_TYPE = 0 +BFD_NOTIFICATION_TYPE = 1 +NOTIFICATION_TEST_PARAM = "vpp_notification_test" +NOTIFICATION_TEST_VALUE = "true" +NOTIFICATION_TIMEOUT = 5.0 +NOTIFICATION_POLL_INTERVAL = 0.5 +# RFC 5880 holds the control-packet interval at one second until a session is +# up, so a fresh BFD session needs several round trips before it transitions. +# Measured bring-up against the scapy responder is around 5.5s, on top of which +# the notification still has to clear the VPP event poll. +BFD_NOTIFICATION_TIMEOUT = 25.0 +# vppProcessEvents() sleeps this long between drains of the VPP event queue, so +# link changes made just before a test starts can still be in flight. +VPP_EVENT_POLL_SECONDS = 2.0 +VPPCTL_TIMEOUT = 10 +BFD_EPHEMERAL_SRC_PORT = 49152 +BFD_STATE_INIT = 2 +BFD_STATE_UP = 3 +# A scapy responder cannot sustain a sub-second reply cadence reliably, and at +# the SAI default the negotiated interval drops the session within 300ms of a +# missed reply. One second with a multiplier of three keeps the session stable +# while still detecting a broken path in about three seconds. +BFD_INTERVAL_USEC = 1000000 +BFD_MULTIPLIER = 3 + + +class NotificationTestBase(T0TestBase): + """Common setup for the opt-in server-owned notification bridge.""" + + def setUp(self, **kwargs): + self.pending_events = [] + params = test_params_get() or {} + if params.get(NOTIFICATION_TEST_PARAM) != NOTIFICATION_TEST_VALUE: + super().setUp(skip_reason="VPP notification tests are opt-in") + return + + T0TestBase.setUp(self, **kwargs) + status = self.client.sai_thrift_enable_notifications() + self.assertEqual(status, SAI_STATUS_SUCCESS) + self.discard_pending_notifications() + + def tearDown(self): + try: + if self.client is not None: + self.discard_pending_notifications() + finally: + super().tearDown() + + def discard_pending_notifications(self): + self.pending_events = [] + self.client.sai_thrift_drain_notifications() + + def wait_for_notification(self, predicate, timeout=NOTIFICATION_TIMEOUT): + """Return the first queued event matching predicate, keeping the rest. + + Events the caller is not waiting for yet are retained so that a drain + which returns several transitions at once cannot lose the one a later + assertion depends on. + """ + deadline = time.monotonic() + timeout + while True: + for index, event in enumerate(self.pending_events): + if predicate(event): + return self.pending_events.pop(index) + + if time.monotonic() >= deadline: + break + + self.pending_events.extend( + self.client.sai_thrift_drain_notifications() + ) + time.sleep(NOTIFICATION_POLL_INTERVAL) + + self.fail( + "timed out waiting for the expected SAI notification; observed {}".format( + [ + (event.notification_type, hex(event.object_id), event.state) + for event in self.pending_events + ] + ) + ) + + @staticmethod + def peer_interface(port_index): + for _, configured_port, interface_name in ptf_config.get("interfaces", []): + if configured_port == port_index: + if not re.fullmatch(r"OEth[0-9]+_peer", interface_name): + raise AssertionError( + "unexpected VPP PTF peer interface: {}".format(interface_name) + ) + return interface_name + raise AssertionError( + "PTF interface for port {} was not configured".format(port_index) + ) + + @staticmethod + def vpp_interface(peer_name): + match = re.fullmatch(r"OEth([0-9]+)_peer", peer_name) + if match is None: + raise AssertionError( + "unexpected VPP PTF peer interface: {}".format(peer_name) + ) + return "OEthernet{}".format(match.group(1)) + + @staticmethod + def vpp_hwif_name(vpp_interface_name): + return "host-{}".format(vpp_interface_name) + + @staticmethod + def set_link_state(interface_name, is_up): + state = "up" if is_up else "down" + subprocess.run( + ["ip", "link", "set", "dev", interface_name, state], + check=True, + ) + + @staticmethod + def vppctl(*command): + """Return combined vppctl output, used as dataplane-side evidence. + + Reading VPP directly keeps the assertions independent of the SAI object + that the notification itself updates. + """ + result = subprocess.run( + ["vppctl"] + list(command), + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + timeout=VPPCTL_TIMEOUT, + ) + return result.stdout.decode("utf-8", "replace") + + def log_vpp_state(self, *command): + output = self.vppctl(*command) + print("vppctl {}:\n{}".format(" ".join(command), output)) + return output + + +class PortNotificationTestBase(NotificationTestBase): + """Port notification setup without unrelated L3 configuration. + + The link is flapped on the SAI host-interface netdev rather than on the + wire-side ``OEth_peer``. VPP binds ``host-OEthernet`` with an + AF_PACKET socket, and that driver does not watch the underlying netdev's + carrier: downing the veth peer leaves Linux reporting NO-CARRIER while + ``vppctl show hardware-interfaces`` still reports the link up, so no + ``sw_interface_event`` is ever raised and no notification can follow. + Downing the host-interface netdev is still the HLD's ``ip link set down`` + stimulus, and linux-cp propagates it to the paired hardware interface, + which does raise the asynchronous VPP event under test. + """ + + def setUp(self): + super().setUp( + is_remove_default_vlan=False, + is_create_vlan=False, + is_create_fdb=False, + is_create_default_route=False, + is_create_lag=False, + is_create_vlan_itf=False, + is_create_route_for_vlan_itf=False, + is_create_route_for_lag=False, + wait_sec=1, + ) + self.port = self.dut.port_obj_list[0] + self.hostif_dev = self.port.config.name + self.vpp_hwif = self.vpp_hwif_name( + self.vpp_interface(self.peer_interface(self.port.dev_port_index)) + ) + self.set_link_state(self.hostif_dev, True) + # Let the link-up events raised above reach the queue before the + # discard, otherwise they surface mid-test and the port looks flappy. + time.sleep(VPP_EVENT_POLL_SECONDS * 2) + self.discard_pending_notifications() + + def port_event(self, state): + event = self.wait_for_notification( + lambda event: event.notification_type == PORT_NOTIFICATION_TYPE + and event.object_id == self.port.oid + and event.state == state + ) + self.log_vpp_state("show", "hardware-interfaces", self.vpp_hwif) + return event + + +class PortStateChangeTest(PortNotificationTestBase): + """Verify a VPP link-down event reaches the SAI callback bridge.""" + + def runTest(self): + try: + self.set_link_state(self.hostif_dev, False) + self.port_event(SAI_PORT_OPER_STATUS_DOWN) + finally: + self.set_link_state(self.hostif_dev, True) + + +class PortStateRecoveryTest(PortNotificationTestBase): + """Verify a link-down/link-up sequence reaches SAI in order.""" + + def runTest(self): + try: + self.set_link_state(self.hostif_dev, False) + self.port_event(SAI_PORT_OPER_STATUS_DOWN) + self.set_link_state(self.hostif_dev, True) + self.port_event(SAI_PORT_OPER_STATUS_UP) + finally: + self.set_link_state(self.hostif_dev, True) + + +class BfdResponder: + """Small Scapy responder for one single-hop or multihop BFD session. + + The session is anchored on a LAG router interface, and VPP picks the egress + member by hashing the flow, so every member peer is watched rather than + guessing which one carries the control packets. Replies go back out the + member the request arrived on. + """ + + def __init__(self, interface_names, local_ip, remote_ip, udp_port, discriminator): + self.interface_names = list(interface_names) + self.local_ip = local_ip + self.remote_ip = remote_ip + self.udp_port = udp_port + self.discriminator = discriminator + self.source_macs = { + name: get_if_hwaddr(name) for name in self.interface_names + } + self.stop_event = threading.Event() + self.threads = [ + threading.Thread(target=self._run, args=(name,), daemon=True) + for name in self.interface_names + ] + + def start(self): + for thread in self.threads: + thread.start() + + def stop(self): + self.stop_event.set() + for thread in self.threads: + thread.join(timeout=2) + + def _run(self, interface_name): + while not self.stop_event.is_set(): + sniff( + iface=interface_name, + filter="udp", + timeout=0.5, + store=False, + prn=lambda packet: self._respond(interface_name, packet), + ) + + def _respond(self, interface_name, packet): + if not packet.haslayer(Ether) or not packet.haslayer(IP): + return + if not packet.haslayer(UDP): + return + + udp = packet[UDP] + if udp.dport != self.udp_port: + return + + bfd = packet.getlayer(BFD) + if bfd is None: + try: + bfd = BFD(bytes(udp.payload)) + except Exception: + return + + # Follow the RFC 5880 state machine. A peer sitting in Down only leaves + # it when it hears Down or Init, so a responder that always advertises + # Up leaves the session stuck with VPP Down and remote Up forever. + response_state = ( + BFD_STATE_UP + if bfd.sta in (BFD_STATE_INIT, BFD_STATE_UP) + else BFD_STATE_INIT + ) + + response = ( + Ether(src=self.source_macs[interface_name], dst=packet[Ether].src) + # Single-hop BFD is GTSM protected, so the reply has to arrive with + # TTL 255. RFC 5880 also fixes the destination port at 3784 (4784 for + # multihop) in both directions, with an ephemeral source port; VPP + # silently ignores a reply that mirrors the ports instead. + / IP(src=self.remote_ip, dst=self.local_ip, ttl=255) + / UDP(sport=BFD_EPHEMERAL_SRC_PORT, dport=self.udp_port) + / BFD( + version=1, + diag=0, + sta=response_state, + flags=0, + detect_mult=BFD_MULTIPLIER, + my_discriminator=self.discriminator, + your_discriminator=bfd.my_discriminator, + min_tx_interval=BFD_INTERVAL_USEC, + min_rx_interval=BFD_INTERVAL_USEC, + echo_rx_interval=0, + ) + ) + sendp(response, iface=interface_name, verbose=False) + + +class BfdNotificationTestBase(NotificationTestBase): + """Build a LAG-backed BFD session with a real connected local address.""" + + local_ip = "10.1.1.1" + remote_ip = "10.1.1.2" + gateway_ip = "10.1.1.2" + local_discriminator = 0x1001 + remote_discriminator = 0x2001 + udp_port = 3784 + multihop = False + + def setUp(self): + if os.environ.get("SIMULATE_SONIC") != "1": + super().setUp(skip_reason="BFD notification tests require SIMULATE_SONIC=1") + return + + self.bfd_session = None + self.lag_rif = None + self.neighbor_entry = None + self.next_hop = None + self.route_entry = None + self.responder = None + + super().setUp( + is_remove_default_vlan=False, + is_create_vlan=False, + is_create_fdb=False, + is_create_default_route=False, + is_create_lag=True, + is_create_vlan_itf=False, + is_create_route_for_vlan_itf=False, + is_create_route_for_lag=False, + wait_sec=1, + ) + + if not self.dut.default_vrf: + self.route_configer.get_default_virtual_router() + + lag = self.dut.lag_list[0] + self.lag_rif = self.route_configer.create_router_interface(lag) + self.peers = [ + self.peer_interface(port_index) + for port_index in lag.member_port_indexs + ] + self.peer_mac = get_if_hwaddr(self.peers[0]) + + self.neighbor_entry = sai_thrift_neighbor_entry_t( + rif_id=self.lag_rif, + ip_address=sai_ipaddress(self.gateway_ip), + ) + status = sai_thrift_create_neighbor_entry( + self.client, + self.neighbor_entry, + dst_mac_address=self.peer_mac, + no_host_route=False, + ) + self.assertEqual(status, SAI_STATUS_SUCCESS) + + if self.multihop: + self.next_hop = sai_thrift_create_next_hop( + self.client, + ip=sai_ipaddress(self.gateway_ip), + router_interface_id=self.lag_rif, + type=SAI_NEXT_HOP_TYPE_IP, + ) + self.assertEqual(self.status(), SAI_STATUS_SUCCESS) + self.route_entry = sai_thrift_route_entry_t( + vr_id=self.dut.default_vrf, + destination=sai_ipprefix(self.remote_ip + "/32"), + ) + status = sai_thrift_create_route_entry( + self.client, + self.route_entry, + next_hop_id=self.next_hop, + ) + self.assertEqual(status, SAI_STATUS_SUCCESS) + + def start_session(self): + self.responder = BfdResponder( + self.peers, + self.local_ip, + self.remote_ip, + self.udp_port, + self.remote_discriminator, + ) + self.responder.start() + + self.bfd_session = sai_thrift_create_bfd_session( + self.client, + type=SAI_BFD_SESSION_TYPE_ASYNC_ACTIVE, + virtual_router=self.dut.default_vrf, + local_discriminator=self.local_discriminator, + remote_discriminator=self.remote_discriminator, + udp_src_port=49152, + bfd_encapsulation_type=SAI_BFD_ENCAPSULATION_TYPE_NONE, + iphdr_version=4, + src_ip_address=sai_ipaddress(self.local_ip), + dst_ip_address=sai_ipaddress(self.remote_ip), + min_tx=BFD_INTERVAL_USEC, + min_rx=BFD_INTERVAL_USEC, + multiplier=BFD_MULTIPLIER, + hw_lookup_valid=True, + multihop=self.multihop, + cbit=False, + admin_state=True, + ) + self.assertNotEqual(self.bfd_session, SAI_NULL_OBJECT_ID) + self.assertEqual(self.status(), SAI_STATUS_SUCCESS) + + def bfd_event(self, state): + return self.wait_for_notification( + lambda event: event.notification_type == BFD_NOTIFICATION_TYPE + and event.object_id == self.bfd_session + and event.state == state, + timeout=BFD_NOTIFICATION_TIMEOUT, + ) + + def assert_bfd_state(self, state): + attributes = sai_thrift_get_bfd_session_attribute( + self.client, + self.bfd_session, + state=True, + ) + self.assertEqual(attributes["state"], state) + + def assert_vpp_bfd_state(self, expected_state_word): + """Corroborate the SAI state against VPP's own session table. + + SAI_BFD_SESSION_ATTR_STATE is written by the same code path that emits + the notification, so it cannot on its own show that VPP really moved. + """ + output = self.log_vpp_state("show", "bfd", "sessions") + self.assertIn( + self.remote_ip, + output, + "VPP has no BFD session towards {}".format(self.remote_ip), + ) + self.assertRegex( + output, + r"(?i)\b{}\b".format(expected_state_word), + "VPP did not report BFD state {}".format(expected_state_word), + ) + + def tearDown(self): + try: + if self.responder is not None: + self.responder.stop() + if self.bfd_session is not None: + sai_thrift_remove_bfd_session(self.client, self.bfd_session) + if self.route_entry is not None: + sai_thrift_remove_route_entry(self.client, self.route_entry) + if self.next_hop is not None: + sai_thrift_remove_next_hop(self.client, self.next_hop) + if self.neighbor_entry is not None: + sai_thrift_remove_neighbor_entry(self.client, self.neighbor_entry) + finally: + super().tearDown() + + + def bring_session_up(self): + self.start_session() + self.bfd_event(SAI_BFD_SESSION_STATE_UP) + self.assert_bfd_state(SAI_BFD_SESSION_STATE_UP) + self.assert_vpp_bfd_state("up") + + def break_path(self): + """Silence the peer so VPP misses detect_mult consecutive intervals.""" + self.responder.stop() + self.responder = None + self.bfd_event(SAI_BFD_SESSION_STATE_DOWN) + self.assert_bfd_state(SAI_BFD_SESSION_STATE_DOWN) + self.assert_vpp_bfd_state("down") + + +class BfdSessionUpTest(BfdNotificationTestBase): + """Verify that a responder-driven BFD session emits an UP notification.""" + + def runTest(self): + self.bring_session_up() + + +class BfdSessionDownTest(BfdNotificationTestBase): + """Verify that stopping the responder emits a BFD DOWN notification.""" + + def runTest(self): + self.bring_session_up() + self.break_path() + + +class BfdMultihopTest(BfdNotificationTestBase): + """Verify multihop BFD uses UDP/4784 and a routed lookup.""" + + remote_ip = "10.1.2.2" + gateway_ip = "10.1.1.2" + udp_port = 4784 + multihop = True + + def runTest(self): + self.bring_session_up() + self.break_path() From c03e12ea8b205793be321bbc790352d57147da5f Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Thu, 20 Aug 2026 12:17:32 -0700 Subject: [PATCH 02/10] fix: make port notification tests use SAI admin state Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 176 +++++++++++-------------- 1 file changed, 80 insertions(+), 96 deletions(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index 4bb06b4f..f9247f83 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -9,20 +9,14 @@ # LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS # FOR A PARTICULAR PURPOSE, MERCHANTABILITY OR NON-INFRINGEMENT. -"""Opt-in VPP SAI notification tests.""" +"""Opt-in SAI notification tests with a VPP-specific BFD fixture.""" import re import os -import subprocess import threading import time -from ptf import config as ptf_config from ptf.testutils import test_params_get -from unittest import SkipTest - -from scapy.all import Ether, IP, UDP, get_if_hwaddr, sendp, sniff -from scapy.contrib.bfd import BFD from sai_thrift.sai_adapter import * from sai_test_base import T0TestBase @@ -31,8 +25,10 @@ PORT_NOTIFICATION_TYPE = 0 BFD_NOTIFICATION_TYPE = 1 -NOTIFICATION_TEST_PARAM = "vpp_notification_test" +NOTIFICATION_TEST_PARAM = "notification_test" NOTIFICATION_TEST_VALUE = "true" +PLATFORM_PARAM = "platform" +VPP_PLATFORM = "vpp" NOTIFICATION_TIMEOUT = 5.0 NOTIFICATION_POLL_INTERVAL = 0.5 # RFC 5880 holds the control-packet interval at one second until a session is @@ -40,9 +36,6 @@ # Measured bring-up against the scapy responder is around 5.5s, on top of which # the notification still has to clear the VPP event poll. BFD_NOTIFICATION_TIMEOUT = 25.0 -# vppProcessEvents() sleeps this long between drains of the VPP event queue, so -# link changes made just before a test starts can still be in flight. -VPP_EVENT_POLL_SECONDS = 2.0 VPPCTL_TIMEOUT = 10 BFD_EPHEMERAL_SRC_PORT = 49152 BFD_STATE_INIT = 2 @@ -111,75 +104,8 @@ def wait_for_notification(self, predicate, timeout=NOTIFICATION_TIMEOUT): ) ) - @staticmethod - def peer_interface(port_index): - for _, configured_port, interface_name in ptf_config.get("interfaces", []): - if configured_port == port_index: - if not re.fullmatch(r"OEth[0-9]+_peer", interface_name): - raise AssertionError( - "unexpected VPP PTF peer interface: {}".format(interface_name) - ) - return interface_name - raise AssertionError( - "PTF interface for port {} was not configured".format(port_index) - ) - - @staticmethod - def vpp_interface(peer_name): - match = re.fullmatch(r"OEth([0-9]+)_peer", peer_name) - if match is None: - raise AssertionError( - "unexpected VPP PTF peer interface: {}".format(peer_name) - ) - return "OEthernet{}".format(match.group(1)) - - @staticmethod - def vpp_hwif_name(vpp_interface_name): - return "host-{}".format(vpp_interface_name) - - @staticmethod - def set_link_state(interface_name, is_up): - state = "up" if is_up else "down" - subprocess.run( - ["ip", "link", "set", "dev", interface_name, state], - check=True, - ) - - @staticmethod - def vppctl(*command): - """Return combined vppctl output, used as dataplane-side evidence. - - Reading VPP directly keeps the assertions independent of the SAI object - that the notification itself updates. - """ - result = subprocess.run( - ["vppctl"] + list(command), - check=False, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - timeout=VPPCTL_TIMEOUT, - ) - return result.stdout.decode("utf-8", "replace") - - def log_vpp_state(self, *command): - output = self.vppctl(*command) - print("vppctl {}:\n{}".format(" ".join(command), output)) - return output - - class PortNotificationTestBase(NotificationTestBase): - """Port notification setup without unrelated L3 configuration. - - The link is flapped on the SAI host-interface netdev rather than on the - wire-side ``OEth_peer``. VPP binds ``host-OEthernet`` with an - AF_PACKET socket, and that driver does not watch the underlying netdev's - carrier: downing the veth peer leaves Linux reporting NO-CARRIER while - ``vppctl show hardware-interfaces`` still reports the link up, so no - ``sw_interface_event`` is ever raised and no notification can follow. - Downing the host-interface netdev is still the HLD's ``ip link set down`` - stimulus, and linux-cp propagates it to the paired hardware interface, - which does raise the asynchronous VPP event under test. - """ + """Port notification setup without unrelated L3 configuration.""" def setUp(self): super().setUp( @@ -194,48 +120,56 @@ def setUp(self): wait_sec=1, ) self.port = self.dut.port_obj_list[0] - self.hostif_dev = self.port.config.name - self.vpp_hwif = self.vpp_hwif_name( - self.vpp_interface(self.peer_interface(self.port.dev_port_index)) + attributes = sai_thrift_get_port_attribute( + self.client, + port_oid=self.port.oid, + admin_state=True, + ) + self.initial_admin_state = attributes["admin_state"] + self.assertTrue( + self.initial_admin_state, + "test fixture must provide an administratively up port", ) - self.set_link_state(self.hostif_dev, True) - # Let the link-up events raised above reach the queue before the - # discard, otherwise they surface mid-test and the port looks flappy. - time.sleep(VPP_EVENT_POLL_SECONDS * 2) self.discard_pending_notifications() def port_event(self, state): - event = self.wait_for_notification( + return self.wait_for_notification( lambda event: event.notification_type == PORT_NOTIFICATION_TYPE and event.object_id == self.port.oid and event.state == state ) - self.log_vpp_state("show", "hardware-interfaces", self.vpp_hwif) - return event + + def set_admin_state(self, admin_state): + status = sai_thrift_set_port_attribute( + self.client, + port_oid=self.port.oid, + admin_state=admin_state, + ) + self.assertEqual(status, SAI_STATUS_SUCCESS) class PortStateChangeTest(PortNotificationTestBase): - """Verify a VPP link-down event reaches the SAI callback bridge.""" + """Verify a port admin-down event reaches the SAI callback bridge.""" def runTest(self): try: - self.set_link_state(self.hostif_dev, False) + self.set_admin_state(False) self.port_event(SAI_PORT_OPER_STATUS_DOWN) finally: - self.set_link_state(self.hostif_dev, True) + self.set_admin_state(self.initial_admin_state) class PortStateRecoveryTest(PortNotificationTestBase): - """Verify a link-down/link-up sequence reaches SAI in order.""" + """Verify a port admin-down/admin-up sequence reaches SAI in order.""" def runTest(self): try: - self.set_link_state(self.hostif_dev, False) + self.set_admin_state(False) self.port_event(SAI_PORT_OPER_STATUS_DOWN) - self.set_link_state(self.hostif_dev, True) + self.set_admin_state(True) self.port_event(SAI_PORT_OPER_STATUS_UP) finally: - self.set_link_state(self.hostif_dev, True) + self.set_admin_state(self.initial_admin_state) class BfdResponder: @@ -253,6 +187,8 @@ def __init__(self, interface_names, local_ip, remote_ip, udp_port, discriminator self.remote_ip = remote_ip self.udp_port = udp_port self.discriminator = discriminator + from scapy.all import get_if_hwaddr + self.source_macs = { name: get_if_hwaddr(name) for name in self.interface_names } @@ -272,6 +208,8 @@ def stop(self): thread.join(timeout=2) def _run(self, interface_name): + from scapy.all import sniff + while not self.stop_event.is_set(): sniff( iface=interface_name, @@ -282,6 +220,9 @@ def _run(self, interface_name): ) def _respond(self, interface_name, packet): + from scapy.all import Ether, IP, UDP, sendp + from scapy.contrib.bfd import BFD + if not packet.haslayer(Ether) or not packet.haslayer(IP): return if not packet.haslayer(UDP): @@ -342,7 +283,48 @@ class BfdNotificationTestBase(NotificationTestBase): udp_port = 3784 multihop = False + @staticmethod + def peer_interface(port_index): + from ptf import config as ptf_config + + for _, configured_port, interface_name in ptf_config.get("interfaces", []): + if configured_port == port_index: + if not re.fullmatch(r"OEth[0-9]+_peer", interface_name): + raise AssertionError( + "unexpected VPP PTF peer interface: {}".format(interface_name) + ) + return interface_name + raise AssertionError( + "PTF interface for port {} was not configured".format(port_index) + ) + + @staticmethod + def vppctl(*command): + """Return VPP state as independent evidence for the BFD assertion.""" + import subprocess + + result = subprocess.run( + ["vppctl"] + list(command), + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + timeout=VPPCTL_TIMEOUT, + ) + return result.stdout.decode("utf-8", "replace") + + def log_vpp_state(self, *command): + output = self.vppctl(*command) + print("vppctl {}:\n{}".format(" ".join(command), output)) + return output + def setUp(self): + params = test_params_get() or {} + if params.get(PLATFORM_PARAM) != VPP_PLATFORM: + super().setUp( + skip_reason="BFD notification tests require platform='vpp'" + ) + return + if os.environ.get("SIMULATE_SONIC") != "1": super().setUp(skip_reason="BFD notification tests require SIMULATE_SONIC=1") return @@ -375,6 +357,8 @@ def setUp(self): self.peer_interface(port_index) for port_index in lag.member_port_indexs ] + from scapy.all import get_if_hwaddr + self.peer_mac = get_if_hwaddr(self.peers[0]) self.neighbor_entry = sai_thrift_neighbor_entry_t( From 86145e40f4631fb3ce5cebbdfb650dce8fa370c0 Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Thu, 20 Aug 2026 12:48:56 -0700 Subject: [PATCH 03/10] fix: harden notification test fixture cleanup Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index f9247f83..01e6710a 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -17,6 +17,7 @@ import time from ptf.testutils import test_params_get +from unittest import SkipTest from sai_thrift.sai_adapter import * from sai_test_base import T0TestBase @@ -124,14 +125,21 @@ def setUp(self): self.client, port_oid=self.port.oid, admin_state=True, + oper_status=True, ) self.initial_admin_state = attributes["admin_state"] - self.assertTrue( - self.initial_admin_state, - "test fixture must provide an administratively up port", - ) + self.initial_oper_status = attributes["oper_status"] + self.port_skip_reason = None + if not self.initial_admin_state: + self.port_skip_reason = "test fixture must provide an administratively up port" + elif self.initial_oper_status != SAI_PORT_OPER_STATUS_UP: + self.port_skip_reason = "test fixture must provide an operationally up port" self.discard_pending_notifications() + def require_port_ready(self): + if self.port_skip_reason: + raise SkipTest(self.port_skip_reason) + def port_event(self, state): return self.wait_for_notification( lambda event: event.notification_type == PORT_NOTIFICATION_TYPE @@ -152,6 +160,7 @@ class PortStateChangeTest(PortNotificationTestBase): """Verify a port admin-down event reaches the SAI callback bridge.""" def runTest(self): + self.require_port_ready() try: self.set_admin_state(False) self.port_event(SAI_PORT_OPER_STATUS_DOWN) @@ -163,6 +172,7 @@ class PortStateRecoveryTest(PortNotificationTestBase): """Verify a port admin-down/admin-up sequence reaches SAI in order.""" def runTest(self): + self.require_port_ready() try: self.set_admin_state(False) self.port_event(SAI_PORT_OPER_STATUS_DOWN) @@ -331,6 +341,7 @@ def setUp(self): self.bfd_session = None self.lag_rif = None + self.owns_lag_rif = False self.neighbor_entry = None self.next_hop = None self.route_entry = None @@ -352,7 +363,9 @@ def setUp(self): self.route_configer.get_default_virtual_router() lag = self.dut.lag_list[0] + existing_rifs = set(lag.rif_list or []) self.lag_rif = self.route_configer.create_router_interface(lag) + self.owns_lag_rif = self.lag_rif not in existing_rifs self.peers = [ self.peer_interface(port_index) for port_index in lag.member_port_indexs @@ -470,6 +483,8 @@ def tearDown(self): sai_thrift_remove_next_hop(self.client, self.next_hop) if self.neighbor_entry is not None: sai_thrift_remove_neighbor_entry(self.client, self.neighbor_entry) + if self.owns_lag_rif: + sai_thrift_remove_router_interface(self.client, self.lag_rif) finally: super().tearDown() From 787c36d08b265a81f61cc1960f7362e933be3d48 Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Fri, 21 Aug 2026 07:13:41 -0700 Subject: [PATCH 04/10] fix: make notification opt-in message generic Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index 01e6710a..f7dabe5a 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -56,7 +56,7 @@ def setUp(self, **kwargs): self.pending_events = [] params = test_params_get() or {} if params.get(NOTIFICATION_TEST_PARAM) != NOTIFICATION_TEST_VALUE: - super().setUp(skip_reason="VPP notification tests are opt-in") + super().setUp(skip_reason="SAI notification tests are opt-in") return T0TestBase.setUp(self, **kwargs) From 2c4ea58b23dbfbf96475f1de00b7bc32886a5671 Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Tue, 25 Aug 2026 11:40:32 -0700 Subject: [PATCH 05/10] fix: use shared notification kind enum Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index f7dabe5a..0e98ab73 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -24,8 +24,12 @@ from sai_utils import sai_ipaddress, sai_ipprefix -PORT_NOTIFICATION_TYPE = 0 -BFD_NOTIFICATION_TYPE = 1 +PORT_NOTIFICATION_TYPE = ( + sai_thrift_notification_type_t.SAI_THRIFT_NOTIFICATION_TYPE_PORT_STATE_CHANGE +) +BFD_NOTIFICATION_TYPE = ( + sai_thrift_notification_type_t.SAI_THRIFT_NOTIFICATION_TYPE_BFD_SESSION_STATE_CHANGE +) NOTIFICATION_TEST_PARAM = "notification_test" NOTIFICATION_TEST_VALUE = "true" PLATFORM_PARAM = "platform" From 2a0aa3530226cfd9d7ebde08f786909a48a41eb4 Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Tue, 25 Aug 2026 11:41:49 -0700 Subject: [PATCH 06/10] fix: keep BFD capture active during responder startup Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 51 ++++++++++++++++---------- 1 file changed, 32 insertions(+), 19 deletions(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index 0e98ab73..c154eb2d 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -42,6 +42,7 @@ # the notification still has to clear the VPP event poll. BFD_NOTIFICATION_TIMEOUT = 25.0 VPPCTL_TIMEOUT = 10 +SNIFFER_START_TIMEOUT = 5.0 BFD_EPHEMERAL_SRC_PORT = 49152 BFD_STATE_INIT = 2 BFD_STATE_UP = 3 @@ -206,32 +207,44 @@ def __init__(self, interface_names, local_ip, remote_ip, udp_port, discriminator self.source_macs = { name: get_if_hwaddr(name) for name in self.interface_names } - self.stop_event = threading.Event() - self.threads = [ - threading.Thread(target=self._run, args=(name,), daemon=True) - for name in self.interface_names - ] + self.sniffers = [] def start(self): - for thread in self.threads: - thread.start() - - def stop(self): - self.stop_event.set() - for thread in self.threads: - thread.join(timeout=2) + import functools - def _run(self, interface_name): - from scapy.all import sniff + from scapy.all import AsyncSniffer - while not self.stop_event.is_set(): - sniff( + for interface_name in self.interface_names: + ready = threading.Event() + sniffer = AsyncSniffer( iface=interface_name, - filter="udp", - timeout=0.5, + filter="udp dst port {} and src host {}".format( + self.udp_port, self.local_ip + ), store=False, - prn=lambda packet: self._respond(interface_name, packet), + prn=functools.partial(self._respond, interface_name), + started_callback=ready.set, ) + sniffer.start() + self.sniffers.append(sniffer) + + if not ready.wait(timeout=SNIFFER_START_TIMEOUT): + self.stop() + raise AssertionError( + "BFD responder failed to start capture on {}".format( + interface_name + ) + ) + + def stop(self): + for sniffer in self.sniffers: + try: + sniffer.stop() + except Exception: + # stop() raises if the sniffer never reached its run loop, which + # happens when start() aborts partway and unwinds through here. + pass + self.sniffers = [] def _respond(self, interface_name, packet): from scapy.all import Ether, IP, UDP, sendp From f7c395dc3337779fea12af790cb5a46a8e3593a4 Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Tue, 25 Aug 2026 11:42:55 -0700 Subject: [PATCH 07/10] fix: guard VPP-specific BFD corroboration Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index c154eb2d..f72de0ea 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -346,7 +346,8 @@ def log_vpp_state(self, *command): def setUp(self): params = test_params_get() or {} - if params.get(PLATFORM_PARAM) != VPP_PLATFORM: + self.platform = params.get(PLATFORM_PARAM) + if self.platform != VPP_PLATFORM: super().setUp( skip_reason="BFD notification tests require platform='vpp'" ) @@ -510,7 +511,8 @@ def bring_session_up(self): self.start_session() self.bfd_event(SAI_BFD_SESSION_STATE_UP) self.assert_bfd_state(SAI_BFD_SESSION_STATE_UP) - self.assert_vpp_bfd_state("up") + if self.platform == VPP_PLATFORM: + self.assert_vpp_bfd_state("up") def break_path(self): """Silence the peer so VPP misses detect_mult consecutive intervals.""" @@ -518,7 +520,8 @@ def break_path(self): self.responder = None self.bfd_event(SAI_BFD_SESSION_STATE_DOWN) self.assert_bfd_state(SAI_BFD_SESSION_STATE_DOWN) - self.assert_vpp_bfd_state("down") + if self.platform == VPP_PLATFORM: + self.assert_vpp_bfd_state("down") class BfdSessionUpTest(BfdNotificationTestBase): From 15ddcce1e15e436ef3c205745e14ec7186907595 Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Tue, 25 Aug 2026 14:46:48 -0700 Subject: [PATCH 08/10] feat: load BFD notifications through harness fixtures Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 387 ++++--------------------- 1 file changed, 56 insertions(+), 331 deletions(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index f72de0ea..032b313d 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -9,11 +9,9 @@ # LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS # FOR A PARTICULAR PURPOSE, MERCHANTABILITY OR NON-INFRINGEMENT. -"""Opt-in SAI notification tests with a VPP-specific BFD fixture.""" +"""Opt-in SAI notification tests with a provider-backed BFD fixture.""" -import re -import os -import threading +import importlib import time from ptf.testutils import test_params_get @@ -21,7 +19,6 @@ from sai_thrift.sai_adapter import * from sai_test_base import T0TestBase -from sai_utils import sai_ipaddress, sai_ipprefix PORT_NOTIFICATION_TYPE = ( @@ -32,26 +29,9 @@ ) NOTIFICATION_TEST_PARAM = "notification_test" NOTIFICATION_TEST_VALUE = "true" -PLATFORM_PARAM = "platform" -VPP_PLATFORM = "vpp" +BFD_FIXTURE_PARAM = "bfd_fixture" NOTIFICATION_TIMEOUT = 5.0 NOTIFICATION_POLL_INTERVAL = 0.5 -# RFC 5880 holds the control-packet interval at one second until a session is -# up, so a fresh BFD session needs several round trips before it transitions. -# Measured bring-up against the scapy responder is around 5.5s, on top of which -# the notification still has to clear the VPP event poll. -BFD_NOTIFICATION_TIMEOUT = 25.0 -VPPCTL_TIMEOUT = 10 -SNIFFER_START_TIMEOUT = 5.0 -BFD_EPHEMERAL_SRC_PORT = 49152 -BFD_STATE_INIT = 2 -BFD_STATE_UP = 3 -# A scapy responder cannot sustain a sub-second reply cadence reliably, and at -# the SAI default the negotiated interval drops the session within 300ms of a -# missed reply. One second with a multiplier of three keeps the session stable -# while still detecting a broken path in about three seconds. -BFD_INTERVAL_USEC = 1000000 -BFD_MULTIPLIER = 3 class NotificationTestBase(T0TestBase): @@ -110,6 +90,7 @@ def wait_for_notification(self, predicate, timeout=NOTIFICATION_TIMEOUT): ) ) + class PortNotificationTestBase(NotificationTestBase): """Port notification setup without unrelated L3 configuration.""" @@ -187,280 +168,55 @@ def runTest(self): self.set_admin_state(self.initial_admin_state) -class BfdResponder: - """Small Scapy responder for one single-hop or multihop BFD session. - - The session is anchored on a LAG router interface, and VPP picks the egress - member by hashing the flow, so every member peer is watched rather than - guessing which one carries the control packets. Replies go back out the - member the request arrived on. +class BfdNotificationTestBase(NotificationTestBase): + """Run generic BFD notification checks against a platform-supplied fixture. + + The fixture module is named by the ``bfd_fixture`` test parameter and must + expose ``create_fixture(test_obj, multihop)`` returning an object with + ``common_config_kwargs()``, ``setup()``, ``start_session()`` returning the + BFD session OID, ``stop_peer()`` and ``teardown()``. It may also provide + ``notification_timeout`` and ``assert_external_state(state_word)`` to + corroborate the transition against the platform's own state. """ - def __init__(self, interface_names, local_ip, remote_ip, udp_port, discriminator): - self.interface_names = list(interface_names) - self.local_ip = local_ip - self.remote_ip = remote_ip - self.udp_port = udp_port - self.discriminator = discriminator - from scapy.all import get_if_hwaddr - - self.source_macs = { - name: get_if_hwaddr(name) for name in self.interface_names - } - self.sniffers = [] - - def start(self): - import functools - - from scapy.all import AsyncSniffer - - for interface_name in self.interface_names: - ready = threading.Event() - sniffer = AsyncSniffer( - iface=interface_name, - filter="udp dst port {} and src host {}".format( - self.udp_port, self.local_ip - ), - store=False, - prn=functools.partial(self._respond, interface_name), - started_callback=ready.set, - ) - sniffer.start() - self.sniffers.append(sniffer) - - if not ready.wait(timeout=SNIFFER_START_TIMEOUT): - self.stop() - raise AssertionError( - "BFD responder failed to start capture on {}".format( - interface_name - ) - ) - - def stop(self): - for sniffer in self.sniffers: - try: - sniffer.stop() - except Exception: - # stop() raises if the sniffer never reached its run loop, which - # happens when start() aborts partway and unwinds through here. - pass - self.sniffers = [] - - def _respond(self, interface_name, packet): - from scapy.all import Ether, IP, UDP, sendp - from scapy.contrib.bfd import BFD - - if not packet.haslayer(Ether) or not packet.haslayer(IP): - return - if not packet.haslayer(UDP): - return - - udp = packet[UDP] - if udp.dport != self.udp_port: - return - - bfd = packet.getlayer(BFD) - if bfd is None: - try: - bfd = BFD(bytes(udp.payload)) - except Exception: - return - - # Follow the RFC 5880 state machine. A peer sitting in Down only leaves - # it when it hears Down or Init, so a responder that always advertises - # Up leaves the session stuck with VPP Down and remote Up forever. - response_state = ( - BFD_STATE_UP - if bfd.sta in (BFD_STATE_INIT, BFD_STATE_UP) - else BFD_STATE_INIT - ) - - response = ( - Ether(src=self.source_macs[interface_name], dst=packet[Ether].src) - # Single-hop BFD is GTSM protected, so the reply has to arrive with - # TTL 255. RFC 5880 also fixes the destination port at 3784 (4784 for - # multihop) in both directions, with an ephemeral source port; VPP - # silently ignores a reply that mirrors the ports instead. - / IP(src=self.remote_ip, dst=self.local_ip, ttl=255) - / UDP(sport=BFD_EPHEMERAL_SRC_PORT, dport=self.udp_port) - / BFD( - version=1, - diag=0, - sta=response_state, - flags=0, - detect_mult=BFD_MULTIPLIER, - my_discriminator=self.discriminator, - your_discriminator=bfd.my_discriminator, - min_tx_interval=BFD_INTERVAL_USEC, - min_rx_interval=BFD_INTERVAL_USEC, - echo_rx_interval=0, - ) - ) - sendp(response, iface=interface_name, verbose=False) - - -class BfdNotificationTestBase(NotificationTestBase): - """Build a LAG-backed BFD session with a real connected local address.""" - - local_ip = "10.1.1.1" - remote_ip = "10.1.1.2" - gateway_ip = "10.1.1.2" - local_discriminator = 0x1001 - remote_discriminator = 0x2001 - udp_port = 3784 multihop = False - @staticmethod - def peer_interface(port_index): - from ptf import config as ptf_config - - for _, configured_port, interface_name in ptf_config.get("interfaces", []): - if configured_port == port_index: - if not re.fullmatch(r"OEth[0-9]+_peer", interface_name): - raise AssertionError( - "unexpected VPP PTF peer interface: {}".format(interface_name) - ) - return interface_name - raise AssertionError( - "PTF interface for port {} was not configured".format(port_index) - ) - - @staticmethod - def vppctl(*command): - """Return VPP state as independent evidence for the BFD assertion.""" - import subprocess - - result = subprocess.run( - ["vppctl"] + list(command), - check=False, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - timeout=VPPCTL_TIMEOUT, - ) - return result.stdout.decode("utf-8", "replace") - - def log_vpp_state(self, *command): - output = self.vppctl(*command) - print("vppctl {}:\n{}".format(" ".join(command), output)) - return output - def setUp(self): params = test_params_get() or {} - self.platform = params.get(PLATFORM_PARAM) - if self.platform != VPP_PLATFORM: - super().setUp( - skip_reason="BFD notification tests require platform='vpp'" - ) + fixture_name = params.get(BFD_FIXTURE_PARAM) + self.bfd_fixture = None + if not fixture_name: + super().setUp(skip_reason="BFD notification fixture is not configured") return - if os.environ.get("SIMULATE_SONIC") != "1": - super().setUp(skip_reason="BFD notification tests require SIMULATE_SONIC=1") - return - - self.bfd_session = None - self.lag_rif = None - self.owns_lag_rif = False - self.neighbor_entry = None - self.next_hop = None - self.route_entry = None - self.responder = None - - super().setUp( - is_remove_default_vlan=False, - is_create_vlan=False, - is_create_fdb=False, - is_create_default_route=False, - is_create_lag=True, - is_create_vlan_itf=False, - is_create_route_for_vlan_itf=False, - is_create_route_for_lag=False, - wait_sec=1, - ) + fixture_module = importlib.import_module(fixture_name) + self.bfd_fixture = fixture_module.create_fixture(self, self.multihop) - if not self.dut.default_vrf: - self.route_configer.get_default_virtual_router() - - lag = self.dut.lag_list[0] - existing_rifs = set(lag.rif_list or []) - self.lag_rif = self.route_configer.create_router_interface(lag) - self.owns_lag_rif = self.lag_rif not in existing_rifs - self.peers = [ - self.peer_interface(port_index) - for port_index in lag.member_port_indexs - ] - from scapy.all import get_if_hwaddr - - self.peer_mac = get_if_hwaddr(self.peers[0]) - - self.neighbor_entry = sai_thrift_neighbor_entry_t( - rif_id=self.lag_rif, - ip_address=sai_ipaddress(self.gateway_ip), - ) - status = sai_thrift_create_neighbor_entry( - self.client, - self.neighbor_entry, - dst_mac_address=self.peer_mac, - no_host_route=False, - ) - self.assertEqual(status, SAI_STATUS_SUCCESS) - - if self.multihop: - self.next_hop = sai_thrift_create_next_hop( - self.client, - ip=sai_ipaddress(self.gateway_ip), - router_interface_id=self.lag_rif, - type=SAI_NEXT_HOP_TYPE_IP, - ) - self.assertEqual(self.status(), SAI_STATUS_SUCCESS) - self.route_entry = sai_thrift_route_entry_t( - vr_id=self.dut.default_vrf, - destination=sai_ipprefix(self.remote_ip + "/32"), - ) - status = sai_thrift_create_route_entry( - self.client, - self.route_entry, - next_hop_id=self.next_hop, - ) - self.assertEqual(status, SAI_STATUS_SUCCESS) - - def start_session(self): - self.responder = BfdResponder( - self.peers, - self.local_ip, - self.remote_ip, - self.udp_port, - self.remote_discriminator, - ) - self.responder.start() + try: + super().setUp(**self.bfd_fixture.common_config_kwargs()) + self.bfd_fixture.setup() + except Exception: + self.bfd_fixture.teardown() + raise - self.bfd_session = sai_thrift_create_bfd_session( - self.client, - type=SAI_BFD_SESSION_TYPE_ASYNC_ACTIVE, - virtual_router=self.dut.default_vrf, - local_discriminator=self.local_discriminator, - remote_discriminator=self.remote_discriminator, - udp_src_port=49152, - bfd_encapsulation_type=SAI_BFD_ENCAPSULATION_TYPE_NONE, - iphdr_version=4, - src_ip_address=sai_ipaddress(self.local_ip), - dst_ip_address=sai_ipaddress(self.remote_ip), - min_tx=BFD_INTERVAL_USEC, - min_rx=BFD_INTERVAL_USEC, - multiplier=BFD_MULTIPLIER, - hw_lookup_valid=True, - multihop=self.multihop, - cbit=False, - admin_state=True, - ) - self.assertNotEqual(self.bfd_session, SAI_NULL_OBJECT_ID) - self.assertEqual(self.status(), SAI_STATUS_SUCCESS) + def tearDown(self): + try: + if self.bfd_fixture is not None: + self.bfd_fixture.teardown() + finally: + super().tearDown() def bfd_event(self, state): + timeout = getattr( + self.bfd_fixture, + "notification_timeout", + NOTIFICATION_TIMEOUT, + ) return self.wait_for_notification( lambda event: event.notification_type == BFD_NOTIFICATION_TYPE and event.object_id == self.bfd_session and event.state == state, - timeout=BFD_NOTIFICATION_TIMEOUT, + timeout=timeout, ) def assert_bfd_state(self, state): @@ -471,68 +227,40 @@ def assert_bfd_state(self, state): ) self.assertEqual(attributes["state"], state) - def assert_vpp_bfd_state(self, expected_state_word): - """Corroborate the SAI state against VPP's own session table. - - SAI_BFD_SESSION_ATTR_STATE is written by the same code path that emits - the notification, so it cannot on its own show that VPP really moved. - """ - output = self.log_vpp_state("show", "bfd", "sessions") - self.assertIn( - self.remote_ip, - output, - "VPP has no BFD session towards {}".format(self.remote_ip), - ) - self.assertRegex( - output, - r"(?i)\b{}\b".format(expected_state_word), - "VPP did not report BFD state {}".format(expected_state_word), - ) - - def tearDown(self): - try: - if self.responder is not None: - self.responder.stop() - if self.bfd_session is not None: - sai_thrift_remove_bfd_session(self.client, self.bfd_session) - if self.route_entry is not None: - sai_thrift_remove_route_entry(self.client, self.route_entry) - if self.next_hop is not None: - sai_thrift_remove_next_hop(self.client, self.next_hop) - if self.neighbor_entry is not None: - sai_thrift_remove_neighbor_entry(self.client, self.neighbor_entry) - if self.owns_lag_rif: - sai_thrift_remove_router_interface(self.client, self.lag_rif) - finally: - super().tearDown() - - def bring_session_up(self): - self.start_session() + self.bfd_session = self.bfd_fixture.start_session() self.bfd_event(SAI_BFD_SESSION_STATE_UP) self.assert_bfd_state(SAI_BFD_SESSION_STATE_UP) - if self.platform == VPP_PLATFORM: - self.assert_vpp_bfd_state("up") + assert_external_state = getattr( + self.bfd_fixture, + "assert_external_state", + None, + ) + if assert_external_state is not None: + assert_external_state("up") def break_path(self): - """Silence the peer so VPP misses detect_mult consecutive intervals.""" - self.responder.stop() - self.responder = None + self.bfd_fixture.stop_peer() self.bfd_event(SAI_BFD_SESSION_STATE_DOWN) self.assert_bfd_state(SAI_BFD_SESSION_STATE_DOWN) - if self.platform == VPP_PLATFORM: - self.assert_vpp_bfd_state("down") + assert_external_state = getattr( + self.bfd_fixture, + "assert_external_state", + None, + ) + if assert_external_state is not None: + assert_external_state("down") class BfdSessionUpTest(BfdNotificationTestBase): - """Verify that a responder-driven BFD session emits an UP notification.""" + """Verify that a fixture-driven BFD session emits an UP notification.""" def runTest(self): self.bring_session_up() class BfdSessionDownTest(BfdNotificationTestBase): - """Verify that stopping the responder emits a BFD DOWN notification.""" + """Verify that stopping the peer emits a BFD DOWN notification.""" def runTest(self): self.bring_session_up() @@ -540,11 +268,8 @@ def runTest(self): class BfdMultihopTest(BfdNotificationTestBase): - """Verify multihop BFD uses UDP/4784 and a routed lookup.""" + """Verify a fixture-provided multihop BFD session transition.""" - remote_ip = "10.1.2.2" - gateway_ip = "10.1.1.2" - udp_port = 4784 multihop = True def runTest(self): From f0efa443f5437d434561fa2f0407e283bcd9f35d Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Wed, 26 Aug 2026 14:40:57 -0700 Subject: [PATCH 09/10] fix: use PTF ports for common BFD responder Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 333 +++++++++++++++++++++---- 1 file changed, 283 insertions(+), 50 deletions(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index 032b313d..b79ecefc 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -7,18 +7,20 @@ # THIS CODE IS PROVIDED ON AN *AS IS* BASIS, WITHOUT WARRANTIES OR # CONDITIONS OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT # LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS -# FOR A PARTICULAR PURPOSE, MERCHANTABILITY OR NON-INFRINGEMENT. +# FOR A PARTICULAR PURPOSE, MERCHANTABILITY, NON-INFRINGEMENT. -"""Opt-in SAI notification tests with a provider-backed BFD fixture.""" +"""Opt-in SAI notification tests.""" -import importlib +import os +import threading import time -from ptf.testutils import test_params_get +from ptf.testutils import dp_poll, send_packet, test_params_get from unittest import SkipTest from sai_thrift.sai_adapter import * from sai_test_base import T0TestBase +from sai_utils import sai_ipaddress, sai_ipprefix PORT_NOTIFICATION_TYPE = ( @@ -29,9 +31,18 @@ ) NOTIFICATION_TEST_PARAM = "notification_test" NOTIFICATION_TEST_VALUE = "true" -BFD_FIXTURE_PARAM = "bfd_fixture" +PLATFORM_PARAM = "platform" +VPP_PLATFORM = "vpp" NOTIFICATION_TIMEOUT = 5.0 NOTIFICATION_POLL_INTERVAL = 0.5 +BFD_NOTIFICATION_TIMEOUT = 25.0 +VPPCTL_TIMEOUT = 10 +RESPONDER_POLL_INTERVAL = 0.5 +BFD_EPHEMERAL_SRC_PORT = 49152 +BFD_STATE_INIT = 2 +BFD_STATE_UP = 3 +BFD_INTERVAL_USEC = 1000000 +BFD_MULTIPLIER = 3 class NotificationTestBase(T0TestBase): @@ -168,55 +179,288 @@ def runTest(self): self.set_admin_state(self.initial_admin_state) -class BfdNotificationTestBase(NotificationTestBase): - """Run generic BFD notification checks against a platform-supplied fixture. +class BfdResponder: + """Respond to BFD packets received on PTF dataplane ports.""" + + def __init__( + self, + test_obj, + port_ids, + local_ip, + remote_ip, + udp_port, + discriminator, + ): + self.test_obj = test_obj + self.port_ids = list(port_ids) + self.local_ip = local_ip + self.remote_ip = remote_ip + self.udp_port = udp_port + self.discriminator = discriminator + self.stop_event = threading.Event() + self.threads = [] + + def start(self): + # PTF's dataplane thread already queues packets per port, so a responder + # thread that starts late still sees the BFD packets sent before it ran. + for port_id in self.port_ids: + thread = threading.Thread( + target=self._run, + args=(port_id,), + daemon=True, + ) + self.threads.append(thread) + thread.start() + + def stop(self): + self.stop_event.set() + for thread in self.threads: + thread.join(timeout=2.0) + self.threads = [] + + def _run(self, port_id): + while not self.stop_event.is_set(): + result = dp_poll( + self.test_obj, + device_number=0, + port_number=port_id, + timeout=RESPONDER_POLL_INTERVAL, + ) + if not isinstance(result, self.test_obj.dataplane.PollSuccess): + continue + self._respond(result.port, result.packet) + + def _respond(self, port_id, packet_data): + from scapy.all import Ether, IP, UDP + from scapy.contrib.bfd import BFD + + packet = Ether(packet_data) + if not packet.haslayer(IP) or not packet.haslayer(UDP): + return + + ip = packet[IP] + udp = packet[UDP] + if ip.src != self.local_ip or udp.dport != self.udp_port: + return + + bfd = packet.getlayer(BFD) + if bfd is None: + try: + bfd = BFD(bytes(udp.payload)) + except Exception: + return + + response_state = ( + BFD_STATE_UP + if bfd.sta in (BFD_STATE_INIT, BFD_STATE_UP) + else BFD_STATE_INIT + ) + response = ( + Ether(src=packet[Ether].dst, dst=packet[Ether].src) + / IP(src=self.remote_ip, dst=self.local_ip, ttl=255) + / UDP(sport=BFD_EPHEMERAL_SRC_PORT, dport=self.udp_port) + / BFD( + version=1, + diag=0, + sta=response_state, + flags=0, + detect_mult=BFD_MULTIPLIER, + my_discriminator=self.discriminator, + your_discriminator=bfd.my_discriminator, + min_tx_interval=BFD_INTERVAL_USEC, + min_rx_interval=BFD_INTERVAL_USEC, + echo_rx_interval=0, + ) + ) + send_packet(self.test_obj, port_id, response) - The fixture module is named by the ``bfd_fixture`` test parameter and must - expose ``create_fixture(test_obj, multihop)`` returning an object with - ``common_config_kwargs()``, ``setup()``, ``start_session()`` returning the - BFD session OID, ``stop_peer()`` and ``teardown()``. It may also provide - ``notification_timeout`` and ``assert_external_state(state_word)`` to - corroborate the transition against the platform's own state. - """ +class BfdNotificationTestBase(NotificationTestBase): + """Build a BFD session and peer through the shared SAI test topology.""" + + local_ip = "10.1.1.1" + gateway_peer_group = 1 + gateway_peer_id = 2 + remote_peer_group = 1 + remote_peer_id = 2 multihop = False def setUp(self): params = test_params_get() or {} - fixture_name = params.get(BFD_FIXTURE_PARAM) - self.bfd_fixture = None - if not fixture_name: - super().setUp(skip_reason="BFD notification fixture is not configured") + self.platform = params.get(PLATFORM_PARAM) + self.bfd_session = None + self.lag_rif = None + self.owns_lag_rif = False + self.neighbor_entry = None + self.next_hop = None + self.route_entry = None + self.responder = None + + if ( + self.platform == VPP_PLATFORM + and os.environ.get("SIMULATE_SONIC") != "1" + ): + super().setUp( + skip_reason="VPP BFD notification tests require SIMULATE_SONIC=1" + ) return - fixture_module = importlib.import_module(fixture_name) - self.bfd_fixture = fixture_module.create_fixture(self, self.multihop) + super().setUp( + is_remove_default_vlan=False, + is_create_vlan=False, + is_create_fdb=False, + is_create_default_route=False, + is_create_lag=True, + is_create_vlan_itf=False, + is_create_route_for_vlan_itf=False, + is_create_route_for_lag=False, + wait_sec=1, + ) + + if not self.dut.default_vrf: + self.route_configer.get_default_virtual_router() + + gateway_peer = self.t1_list[self.gateway_peer_group][self.gateway_peer_id] + remote_peer = self.t1_list[self.remote_peer_group][self.remote_peer_id] + self.gateway_ip = gateway_peer.ipv4 + self.remote_ip = remote_peer.ipv4 + self.udp_port = 4784 if self.multihop else 3784 + + lag = self.dut.lag_list[0] + existing_rifs = set(lag.rif_list or []) + self.lag_rif = self.route_configer.create_router_interface(lag) + self.owns_lag_rif = self.lag_rif not in existing_rifs + self.peer_port_ids = self.get_dev_port_indexes(lag.member_port_indexs) + self.peer_mac = gateway_peer.mac + + self.neighbor_entry = sai_thrift_neighbor_entry_t( + rif_id=self.lag_rif, + ip_address=sai_ipaddress(self.gateway_ip), + ) + status = sai_thrift_create_neighbor_entry( + self.client, + self.neighbor_entry, + dst_mac_address=self.peer_mac, + no_host_route=False, + ) + self.assertEqual(status, SAI_STATUS_SUCCESS) + + if self.multihop: + self.next_hop = sai_thrift_create_next_hop( + self.client, + ip=sai_ipaddress(self.gateway_ip), + router_interface_id=self.lag_rif, + type=SAI_NEXT_HOP_TYPE_IP, + ) + self.assertEqual(self.status(), SAI_STATUS_SUCCESS) + self.route_entry = sai_thrift_route_entry_t( + vr_id=self.dut.default_vrf, + destination=sai_ipprefix(self.remote_ip + "/32"), + ) + status = sai_thrift_create_route_entry( + self.client, + self.route_entry, + next_hop_id=self.next_hop, + ) + self.assertEqual(status, SAI_STATUS_SUCCESS) + + def start_session(self): + self.responder = BfdResponder( + self, + self.peer_port_ids, + self.local_ip, + self.remote_ip, + self.udp_port, + 0x2001, + ) + self.responder.start() try: - super().setUp(**self.bfd_fixture.common_config_kwargs()) - self.bfd_fixture.setup() + self.bfd_session = sai_thrift_create_bfd_session( + self.client, + type=SAI_BFD_SESSION_TYPE_ASYNC_ACTIVE, + virtual_router=self.dut.default_vrf, + local_discriminator=0x1001, + remote_discriminator=0x2001, + udp_src_port=BFD_EPHEMERAL_SRC_PORT, + bfd_encapsulation_type=SAI_BFD_ENCAPSULATION_TYPE_NONE, + iphdr_version=4, + src_ip_address=sai_ipaddress(self.local_ip), + dst_ip_address=sai_ipaddress(self.remote_ip), + min_tx=BFD_INTERVAL_USEC, + min_rx=BFD_INTERVAL_USEC, + multiplier=BFD_MULTIPLIER, + hw_lookup_valid=True, + multihop=self.multihop, + cbit=False, + admin_state=True, + ) + self.assertNotEqual(self.bfd_session, SAI_NULL_OBJECT_ID) + self.assertEqual(self.status(), SAI_STATUS_SUCCESS) except Exception: - self.bfd_fixture.teardown() + self.stop_peer() raise + def stop_peer(self): + if self.responder is not None: + self.responder.stop() + self.responder = None + + @staticmethod + def vppctl(*command): + import subprocess + + result = subprocess.run( + ["vppctl"] + list(command), + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + timeout=VPPCTL_TIMEOUT, + ) + return result.stdout.decode("utf-8", "replace") + + def log_vpp_state(self, *command): + output = self.vppctl(*command) + print("vppctl {}:\n{}".format(" ".join(command), output)) + return output + + def assert_vpp_bfd_state(self, expected_state_word): + if self.platform != VPP_PLATFORM: + return + output = self.log_vpp_state("show", "bfd", "sessions") + self.assertIn( + self.remote_ip, + output, + "VPP has no BFD session towards {}".format(self.remote_ip), + ) + self.assertRegex( + output, + r"(?i)\b{}\b".format(expected_state_word), + "VPP did not report BFD state {}".format(expected_state_word), + ) + def tearDown(self): try: - if self.bfd_fixture is not None: - self.bfd_fixture.teardown() + self.stop_peer() + if self.bfd_session is not None: + sai_thrift_remove_bfd_session(self.client, self.bfd_session) + if self.route_entry is not None: + sai_thrift_remove_route_entry(self.client, self.route_entry) + if self.next_hop is not None: + sai_thrift_remove_next_hop(self.client, self.next_hop) + if self.neighbor_entry is not None: + sai_thrift_remove_neighbor_entry(self.client, self.neighbor_entry) + if self.owns_lag_rif and self.lag_rif is not None: + sai_thrift_remove_router_interface(self.client, self.lag_rif) finally: super().tearDown() def bfd_event(self, state): - timeout = getattr( - self.bfd_fixture, - "notification_timeout", - NOTIFICATION_TIMEOUT, - ) return self.wait_for_notification( lambda event: event.notification_type == BFD_NOTIFICATION_TYPE and event.object_id == self.bfd_session and event.state == state, - timeout=timeout, + timeout=BFD_NOTIFICATION_TIMEOUT, ) def assert_bfd_state(self, state): @@ -228,39 +472,27 @@ def assert_bfd_state(self, state): self.assertEqual(attributes["state"], state) def bring_session_up(self): - self.bfd_session = self.bfd_fixture.start_session() + self.start_session() self.bfd_event(SAI_BFD_SESSION_STATE_UP) self.assert_bfd_state(SAI_BFD_SESSION_STATE_UP) - assert_external_state = getattr( - self.bfd_fixture, - "assert_external_state", - None, - ) - if assert_external_state is not None: - assert_external_state("up") + self.assert_vpp_bfd_state("up") def break_path(self): - self.bfd_fixture.stop_peer() + self.stop_peer() self.bfd_event(SAI_BFD_SESSION_STATE_DOWN) self.assert_bfd_state(SAI_BFD_SESSION_STATE_DOWN) - assert_external_state = getattr( - self.bfd_fixture, - "assert_external_state", - None, - ) - if assert_external_state is not None: - assert_external_state("down") + self.assert_vpp_bfd_state("down") class BfdSessionUpTest(BfdNotificationTestBase): - """Verify that a fixture-driven BFD session emits an UP notification.""" + """Verify that a responder-driven BFD session emits an UP notification.""" def runTest(self): self.bring_session_up() class BfdSessionDownTest(BfdNotificationTestBase): - """Verify that stopping the peer emits a BFD DOWN notification.""" + """Verify that stopping the responder emits a BFD DOWN notification.""" def runTest(self): self.bring_session_up() @@ -268,8 +500,9 @@ def runTest(self): class BfdMultihopTest(BfdNotificationTestBase): - """Verify a fixture-provided multihop BFD session transition.""" + """Verify multihop BFD uses UDP/4784 and a routed lookup.""" + remote_peer_group = 2 multihop = True def runTest(self): From 3bef46fe98eb70e40d7451bbadaa40c9d4c7da81 Mon Sep 17 00:00:00 2001 From: Nicholas Ching Date: Thu, 27 Aug 2026 12:52:20 -0700 Subject: [PATCH 10/10] fix: dispatch BFD state check by platform Signed-off-by: Nicholas Ching --- test/sai_test/sai_notification_test.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/test/sai_test/sai_notification_test.py b/test/sai_test/sai_notification_test.py index b79ecefc..00a34e92 100644 --- a/test/sai_test/sai_notification_test.py +++ b/test/sai_test/sai_notification_test.py @@ -425,8 +425,6 @@ def log_vpp_state(self, *command): return output def assert_vpp_bfd_state(self, expected_state_word): - if self.platform != VPP_PLATFORM: - return output = self.log_vpp_state("show", "bfd", "sessions") self.assertIn( self.remote_ip, @@ -439,6 +437,16 @@ def assert_vpp_bfd_state(self, expected_state_word): "VPP did not report BFD state {}".format(expected_state_word), ) + # A platform opts in to dataplane corroboration by adding its own entry. + PLATFORM_BFD_STATE_CHECKS = { + VPP_PLATFORM: assert_vpp_bfd_state, + } + + def assert_platform_bfd_state(self, expected_state_word): + check = self.PLATFORM_BFD_STATE_CHECKS.get(self.platform) + if check is not None: + check(self, expected_state_word) + def tearDown(self): try: self.stop_peer() @@ -475,13 +483,13 @@ def bring_session_up(self): self.start_session() self.bfd_event(SAI_BFD_SESSION_STATE_UP) self.assert_bfd_state(SAI_BFD_SESSION_STATE_UP) - self.assert_vpp_bfd_state("up") + self.assert_platform_bfd_state("up") def break_path(self): self.stop_peer() self.bfd_event(SAI_BFD_SESSION_STATE_DOWN) self.assert_bfd_state(SAI_BFD_SESSION_STATE_DOWN) - self.assert_vpp_bfd_state("down") + self.assert_platform_bfd_state("down") class BfdSessionUpTest(BfdNotificationTestBase):