diff --git a/tests/ipfwd/test_dip_sip.py b/tests/ipfwd/test_dip_sip.py index b5a508361af..4670f505e0f 100644 --- 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))