Add traffic tests for static and directly connected routes in IPFWD test suite (sonic-net#17058)

dt-nexthop · gshemesh2 · commit bd86c6f157d4 · 2025-12-21T11:09:27.000+02:00
What is the motivation for this PR?
Additional coverage in forwarding test

How did you do it?
Added extra cases to the existing script to handle the cases where traffic is sent to directly connected route as well as to a prefix that is reachable through a static route.

How did you verify/test it?
Run on T1 topology running master

Signed-off-by: Guy Shemesh &lt;gshemesh@nvidia.com&gt;
diff --git a/tests/ipfwd/test_dip_sip.py b/tests/ipfwd/test_dip_sip.py
@@ -7,6 +7,8 @@
 
 DEFAULT_HLIM_TTL = 64
 WAIT_EXPECTED_PACKET_TIMEOUT = 5
+STATIC_ROUTE = '201.1.1.1/32'
+STATIC_ROUTE_IPV6 = '2001:db8:1::1/128'
 
 logger = logging.getLogger(__name__)
 
@@ -23,67 +25,309 @@ def lldp_setup(duthosts, enum_rand_one_per_hwsku_frontend_hostname, patch_lldpct
     unpatch_lldpctl(localhost, duthost)
 
 
-def run_test_ipv6(ptfadapter, facts):
-    logger.info("Running test with ipv6 packets")
+@pytest.fixture(scope="function", autouse=True)
+def setup_static_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts, request):
+    """
+    Fixture to set up and tear down static routes for IPv4 and IPv6.
 
-    pkt = testutils.simple_udpv6_packet(
-        eth_dst=facts['src_router_mac'],
-        eth_src=facts['src_host_mac'],
-        ipv6_src=facts['dst_host_ipv6'],
-        ipv6_dst=facts['dst_host_ipv6'],
-        ipv6_hlim=DEFAULT_HLIM_TTL
+    This fixture performs the following actions:
+    1. Adds IPv4 and IPv6 static routes to the DUT.
+    2. Verifies that the routes are added correctly.
+    3. Yields control back to the test.
+    4. Removes the static routes after the test is complete.
 
-    )
-    logger.info("\nSend Packet:\neth_dst: {}, eth_src: {}, ipv6 ip: {}".format(
-        facts['src_router_mac'], facts['src_host_mac'], facts['dst_host_ipv6'])
-    )
+    Args:
+        duthosts: Fixture providing access to DUT hosts.
+        enum_rand_one_per_hwsku_frontend_hostname: Fixture selecting a random frontend DUT.
+        gather_facts: Fixture providing network facts.
+        request: Pytest request object.
 
-    testutils.send(ptfadapter, facts['src_port_ids'][0], pkt)
+    Raises:
+        pytest.fail: If any step in adding or verifying routes fails.
 
-    exp_pkt = testutils.simple_udpv6_packet(
-        eth_dst=facts['dst_host_mac'],
-        eth_src=facts['dst_router_mac'],
-        ipv6_src=facts['dst_host_ipv6'],
-        ipv6_dst=facts['dst_host_ipv6'],
-        ipv6_hlim=DEFAULT_HLIM_TTL-1
+    Yields:
+        None
+    """
+    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+
+    # Configure IPv4 static route
+    try:
+        result = duthost.command("ip route add {} via {}".format(STATIC_ROUTE, gather_facts['dst_host_ipv4']))
+        if result['rc'] != 0:
+            raise Exception("Failed to add IPv4 static route: {}".format(result['stderr']))
+    except Exception as e:
+        logger.error("Error occurred while adding IPv4 static route: %s", str(e))
+        pytest.fail("IPv4 static route addition failed")
+
+    # Configure IPv6 static route
+    try:
+        result = duthost.command("ip -6 route add {} via {}".format(STATIC_ROUTE_IPV6, gather_facts['dst_host_ipv6']))
+        if result['rc'] != 0:
+            raise Exception("Failed to add IPv6 static route: {}".format(result['stderr']))
+    except Exception as e:
+        logger.error("Error occurred while adding IPv6 static route: %s", str(e))
+        pytest.fail("IPv6 static route addition failed")
+
+    # Verify IPv4 route is in the routing table
+
+    try:
+        result = duthost.command("ip route show {}".format(STATIC_ROUTE))
+        assert result['rc'] == 0, "Failed to show IPv4 static route: {}".format(result['stderr'])
+        assert "via " + gather_facts['dst_host_ipv4'] in result["stdout"], "IPv4 static route verification failed"
+    except Exception as e:
+        logger.error("Error occurred while verifying IPv4 static route: %s", str(e))
+        pytest.fail("IPv4 static route verification failed")
+
+    # # Verify IPv6 route is in the routing table
+    try:
+        result = duthost.command("ip -6 route show {}".format(STATIC_ROUTE_IPV6))
+        assert result['rc'] == 0, "Failed to show IPv6 static route: {}".format(result['stderr'])
+        assert "via " + gather_facts['dst_host_ipv6'] in result["stdout"], "IPv6 static route verification failed"
+    except Exception as e:
+        logger.error("Error occurred while verifying IPv6 static route: %s", str(e))
+        pytest.fail("IPv6 static route verification failed")
+
+    # Continue with the test
+    yield
+
+    # Use either individual functions
+    delete_ipv4_static_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts)
+    delete_ipv6_static_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts)
+
+    # Or use the combined function
+    # delete_static_routes(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts)
+
+
+@pytest.fixture(autouse=True)
+def setup_teardown(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts):
+    yield
+    # Teardown - delete the static routes
+    delete_ipv4_static_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts)
+    delete_ipv6_static_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts)
+
+
+def delete_ipv4_static_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts):
+    """
+    Function to delete IPv4 static route from the DUT.
+
+    This function performs the following actions:
+    1. Gets the correct DUT host instance
+    2. Deletes IPv4 static route from the DUT
+    3. Verifies that the route is removed correctly
+
+    Args:
+        duthosts: Fixture providing access to DUT hosts
+        enum_rand_one_per_hwsku_frontend_hostname: Fixture selecting a random frontend DUT
+        gather_facts: Fixture providing network facts
+
+    Raises:
+        pytest.fail: If any step in removing or verifying route fails
+    """
+    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+
+    # Check if IPv4 route exists before deleting
+    try:
+        check_result = duthost.command("ip route show {}".format(STATIC_ROUTE))
+        if check_result['rc'] == 0 and check_result['stdout'].strip():
+            # Route exists, delete it
+            result = duthost.command("ip route del {}".format(STATIC_ROUTE))
+            if result['rc'] != 0:
+                raise Exception("Failed to delete IPv4 static route: {}".format(result['stderr']))
+            logger.info("Successfully deleted IPv4 static route: {}".format(STATIC_ROUTE))
+        else:
+            logger.info("IPv4 static route {} not present, skipping deletion".format(STATIC_ROUTE))
+    except Exception as e:
+        logger.error("Error occurred while handling IPv4 static route: %s", str(e))
+        pytest.fail("IPv4 static route operation failed")
+
+    # Verify IPv4 route is removed from the routing table
+    try:
+        result = duthost.command("ip route show {}".format(STATIC_ROUTE))
+        assert "No such process" in result['stderr'] or result['stdout'].strip() == "", \
+            "IPv4 static route still exists in routing table"
+    except Exception as e:
+        logger.error("Error occurred while verifying IPv4 static route removal: %s", str(e))
+        pytest.fail("IPv4 static route removal verification failed")
+
+
+def delete_ipv6_static_route(duthosts, enum_rand_one_per_hwsku_frontend_hostname, gather_facts):
+    """
+    Function to delete IPv6 static route from the DUT.
+
+    This function performs the following actions:
+    1. Gets the correct DUT host instance
+    2. Deletes IPv6 static route from the DUT
+    3. Verifies that the route is removed correctly
+
+    Args:
+        duthosts: Fixture providing access to DUT hosts
+        enum_rand_one_per_hwsku_frontend_hostname: Fixture selecting a random frontend DUT
+        gather_facts: Fixture providing network facts
+
+    Raises:
+        pytest.fail: If any step in removing or verifying route fails
+    """
+    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+
+    # Check if IPv6 route exists before deleting
+    try:
+        check_result = duthost.command("ip -6 route show {}".format(STATIC_ROUTE_IPV6))
+        if check_result['rc'] == 0 and check_result['stdout'].strip():
+            # Route exists, delete it
+            result = duthost.command("ip -6 route del {}".format(STATIC_ROUTE_IPV6))
+            if result['rc'] != 0:
+                raise Exception("Failed to delete IPv6 static route: {}".format(result['stderr']))
+            logger.info("Successfully deleted IPv6 static route: {}".format(STATIC_ROUTE_IPV6))
+        else:
+            logger.info("IPv6 static route {} not present, skipping deletion".format(STATIC_ROUTE_IPV6))
+    except Exception as e:
+        logger.error("Error occurred while handling IPv6 static route: %s", str(e))
+        pytest.fail("IPv6 static route operation failed")
+
+    # Verify IPv6 route is removed from the routing table
+    try:
+        result = duthost.command("ip -6 route show {}".format(STATIC_ROUTE_IPV6))
+        assert "No such process" in result['stderr'] or result['stdout'].strip() == "", \
+            "IPv6 static route still exists in routing table"
+    except Exception as e:
+        logger.error("Error occurred while verifying IPv6 static route removal: %s", str(e))
+        pytest.fail("IPv6 static route removal verification failed")
+
+
+ipv4_test_cases = [
+    pytest.param(
+        'Same SIP and DIP',
+        lambda facts: facts['dst_host_ipv4'],
+        lambda facts: facts['dst_host_ipv4'],
+        id='ipv4_same_sip_dip'
+    ),
+    pytest.param(
+        'Different subnet SIP/DIP - Directly connected route',
+        lambda facts: facts['src_host_ipv4'],
+        lambda facts: facts['dst_host_ipv4'],
+        id='ipv4_different_sip_dip_connectedroute'
+    ),
+    pytest.param(
+        'Different subnet SIP/DIP - Destination not directly connected',
+        lambda facts: facts['src_host_ipv4'],
+        lambda facts: STATIC_ROUTE.split('/')[0],
+        id='ipv4_different_sip_dip_staticrouteprefix'
     )
-    logger.info("\nExpect Packet:\neth_dst: {}, eth_src: {}, ipv6 ip: {}".format(
-        facts['dst_host_mac'], facts['dst_router_mac'], facts['dst_host_ipv6'])
+]
+
+ipv6_test_cases = [
+    pytest.param(
+        'Same SIP and DIP',
+        lambda facts: facts['dst_host_ipv6'],
+        lambda facts: facts['dst_host_ipv6'],
+        id='ipv6_same_sip_dip'
+    ),
+    pytest.param(
+        'Different subnet SIP/DIP - Directly connected route',
+        lambda facts: facts['src_host_ipv6'],
+        lambda facts: facts['dst_host_ipv6'],
+        id='ipv6_different_sip_dip_connectedroute'
+    ),
+    pytest.param(
+        'Different subnet SIP/DIP - Destination not directly connected',
+        lambda facts: facts['src_host_ipv6'],
+        lambda facts: STATIC_ROUTE_IPV6.split('/')[0],
+        id='ipv6_different_sip_dip_staticrouteprefix'
     )
+]
 
-    testutils.verify_packet_any_port(ptfadapter, exp_pkt, facts['dst_port_ids'], timeout=WAIT_EXPECTED_PACKET_TIMEOUT)
 
+@pytest.mark.parametrize('test_name, get_src_ip, get_dst_ip', ipv4_test_cases)
+def test_ipv4_forwarding(tbinfo, ptfadapter, gather_facts, enum_rand_one_frontend_asic_index,
+                         test_name, get_src_ip, get_dst_ip):
+    """Test IPv4 forwarding with various source/destination IP combinations"""
+    ptfadapter.reinit()
+    logger.info("Testing case: {}".format(test_name))
+
+    ip_src = get_src_ip(gather_facts)
+    ip_dst = get_dst_ip(gather_facts)
 
-def run_test_ipv4(ptfadapter, facts):
-    logger.info("Running test with ipv4 packets")
     pkt = testutils.simple_udp_packet(
-        eth_dst=facts['src_router_mac'],
-        eth_src=facts['src_host_mac'],
-        ip_src=facts['dst_host_ipv4'],
-        ip_dst=facts['dst_host_ipv4'],
+        eth_dst=gather_facts['src_router_mac'],
+        eth_src=gather_facts['src_host_mac'],
+        ip_src=ip_src,
+        ip_dst=ip_dst,
         ip_ttl=DEFAULT_HLIM_TTL
     )
-    logger.info("\nSend Packet:\neth_dst: {}, eth_src: {}, ipv4 ip: {}".format(
-        facts['src_router_mac'], facts['src_host_mac'], facts['dst_host_ipv4'])
+    logger.info("\nSend Packet:\neth_dst: {}, eth_src: {}, ip_src: {}, ip_dst: {}".format(
+        gather_facts['src_router_mac'], gather_facts['src_host_mac'],
+        ip_src, ip_dst)
     )
 
-    testutils.send(ptfadapter, facts['src_port_ids'][0], pkt)
+    testutils.send(ptfadapter, gather_facts['src_port_ids'][0], pkt)
 
     exp_pkt = testutils.simple_udp_packet(
-        eth_dst=facts['dst_host_mac'],
-        eth_src=facts['dst_router_mac'],
-        ip_src=facts['dst_host_ipv4'],
-        ip_dst=facts['dst_host_ipv4'],
+        eth_dst=gather_facts['dst_host_mac'],
+        eth_src=gather_facts['dst_router_mac'],
+        ip_src=ip_src,
+        ip_dst=ip_dst,
         ip_ttl=DEFAULT_HLIM_TTL-1
     )
-    logger.info("\nExpect Packet:\neth_dst: {}, eth_src: {}, ipv4 ip: {}".format(
-        facts['dst_host_mac'], facts['dst_router_mac'], facts['dst_host_ipv4'])
+    logger.info(
+        "\nExpect Packet:\neth_dst: {}, eth_src: {}, ip_src: {}, ip_dst: {}".format(
+            gather_facts['dst_host_mac'], gather_facts['dst_router_mac'],
+            ip_src, ip_dst)
     )
 
-    testutils.verify_packet_any_port(ptfadapter, exp_pkt, facts['dst_port_ids'], timeout=WAIT_EXPECTED_PACKET_TIMEOUT)
+    try:
+        testutils.verify_packet_any_port(ptfadapter, exp_pkt,
+                                         gather_facts['dst_port_ids'],
+                                         timeout=WAIT_EXPECTED_PACKET_TIMEOUT)
+    except AssertionError as e:
+        logger.error("Expected packet was not received")
+        pytest.fail("Test case failed: {} - {}".format(test_name, str(e)))
+
+    logger.info("Test case passed: {}\n".format(test_name))
 
 
-def test_dip_sip(tbinfo, ptfadapter, gather_facts, enum_rand_one_frontend_asic_index):
+@pytest.mark.parametrize('test_name, get_src_ip, get_dst_ip', ipv6_test_cases)
+def test_ipv6_forwarding(tbinfo, ptfadapter, gather_facts, enum_rand_one_frontend_asic_index,
+                         test_name, get_src_ip, get_dst_ip):
+    """Test IPv6 forwarding with various source/destination IP combinations"""
     ptfadapter.reinit()
-    run_test_ipv4(ptfadapter, gather_facts)
-    run_test_ipv6(ptfadapter, gather_facts)
+
+    logger.info("Testing case: {}".format(test_name))
+
+    ipv6_src = get_src_ip(gather_facts)
+    ipv6_dst = get_dst_ip(gather_facts)
+
+    pkt = testutils.simple_udpv6_packet(
+        eth_dst=gather_facts['src_router_mac'],
+        eth_src=gather_facts['src_host_mac'],
+        ipv6_src=ipv6_src,
+        ipv6_dst=ipv6_dst,
+        ipv6_hlim=DEFAULT_HLIM_TTL
+    )
+    logger.info("\nSend Packet:\neth_dst: {}, eth_src: {}, ipv6_src: {}, ipv6_dst: {}".format(
+        gather_facts['src_router_mac'], gather_facts['src_host_mac'],
+        ipv6_src, ipv6_dst)
+    )
+
+    testutils.send(ptfadapter, gather_facts['src_port_ids'][0], pkt)
+
+    exp_pkt = testutils.simple_udpv6_packet(
+        eth_dst=gather_facts['dst_host_mac'],
+        eth_src=gather_facts['dst_router_mac'],
+        ipv6_src=ipv6_src,
+        ipv6_dst=ipv6_dst,
+        ipv6_hlim=DEFAULT_HLIM_TTL-1
+    )
+    logger.info(
+        "\nExpect Packet:\neth_dst: {}, eth_src: {}, ipv6_src: {}, ipv6_dst: {}".format(
+            gather_facts['dst_host_mac'], gather_facts['dst_router_mac'],
+            ipv6_src, ipv6_dst)
+    )
+
+    try:
+        testutils.verify_packet_any_port(ptfadapter, exp_pkt,
+                                         gather_facts['dst_port_ids'],
+                                         timeout=WAIT_EXPECTED_PACKET_TIMEOUT)
+    except AssertionError as e:
+        logger.error("Expected packet was not received")
+        pytest.fail("Test case failed: {} - {}".format(test_name, str(e)))
+    logger.info("Test case passed: {}\n".format(test_name))
