Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -799,13 +799,6 @@ drop_packets:
conditions:
- "topo_type in ['m0', 'mx']"

drop_packets/test_configurable_drop_counters.py::test_neighbor_link_down:
xfail:
reason: "xfail for IPv6-only topologies, issue with python max size"
conditions:
- "https://github.com/sonic-net/sonic-mgmt/issues/19920 and '-v6-' in topo_name"


drop_packets/test_drop_counters.py:
xfail:
reason: "xfail for scale topology, PTF is not stable at the scale testbed"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,14 @@
#Hence, it is not dropped by default in Cisco-8000. For dropping link local address, it should be done through security/DATA ACL
drop_packets/test_configurable_drop_counters.py::test_dip_link_local:
skip:
reason: "MGFX topos doesn't support drop packets / Cisco 8000 platform and some mlx platforms does not drop DIP link local packets / KVM do not support drop reason in testcase."
reason: "MGFX topos doesn't support drop packets / Cisco 8000 platform and some mlx platforms does not drop DIP link local packets / KVM do not support drop reason in testcase / IPv6 DIP_LINK_LOCAL drop reason not supported on broadcom."
conditions_logical_operator: or
conditions:
- "'Mellanox' in hwsku"
- asic_type=='cisco-8000'
- "topo_type in ['m0', 'mx']"
- "asic_type in ['vs']"
- "asic_type in ['broadcom'] and '-v6-' in topo_name"

drop_packets/test_configurable_drop_counters.py::test_neighbor_link_down:
skip:
Expand All @@ -23,13 +24,14 @@ drop_packets/test_configurable_drop_counters.py::test_neighbor_link_down:

drop_packets/test_configurable_drop_counters.py::test_sip_link_local:
skip:
reason: "MGFX topos doesn't support drop packets / Cisco 8000 platform and some MLX platforms does not drop SIP link local packets / KVM do not support drop reason in testcase."
reason: "MGFX topos doesn't support drop packets / Cisco 8000 platform and some MLX platforms does not drop SIP link local packets / KVM do not support drop reason in testcase / IPv6 DIP_LINK_LOCAL drop reason not supported on broadcom."
conditions_logical_operator: or
conditions:
- asic_type=="cisco-8000"
- "'Mellanox' in hwsku"
- "topo_type in ['m0', 'mx']"
- "asic_type in ['vs']"
- "asic_type in ['broadcom'] and '-v6-' in topo_name"

#######################################
##### test_drop_counters.py #####
Expand Down
76 changes: 54 additions & 22 deletions tests/drop_packets/test_configurable_drop_counters.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,13 @@

import pytest
import ptf.testutils as testutils
from netaddr import IPNetwork, EUI
import ipaddress
from netaddr import EUI

from . import configurable_drop_counters as cdc
from tests.common.helpers.assertions import pytest_assert
from tests.common.utilities import wait_until
from tests.common.utilities import is_ipv6_only_topology
from tests.common.platform.device_utils import fanout_switch_port_lookup
from tests.common.dualtor.mux_simulator_control import toggle_all_simulator_ports_to_rand_selected_tor_m # noqa F401
from tests.common.fixtures.ptfhost_utils import copy_arp_responder_py # noqa F401
Expand All @@ -42,7 +44,9 @@
VLAN_BASE_MAC_PATTERN = "72060001{:04}"

MOCK_DEST_IP = "2.2.2.2"
LINK_LOCAL_IP = "169.254.0.1"
MOCK_DEST_IP_V6 = "2001:db8::2"
LINK_LOCAL_IP = "169.254.0.1" # 169.254.0.0/16
LINK_LOCAL_IP_V6 = "fe81::1" # fe80::/10


# For dualtor
Expand Down Expand Up @@ -131,7 +135,10 @@ def verifyFdbArp(duthost, dst_ip, dst_mac, dst_intf):
Check if the ARP and FDB entry is present
"""
logging.info("Verify if the ARP and FDB entry is present for {}".format(dst_ip))
result = duthost.command("show arp {}".format(dst_ip))
if is_ipv4_address(dst_ip):
result = duthost.command("show arp {}".format(dst_ip))
else:
result = duthost.command("show ndp {}".format(dst_ip))
pytest_assert("Total number of entries 1" in result['stdout'],
"ARP entry for {} missing in ASIC".format(dst_ip))
result = duthost.shell("ip neigh show {}".format(dst_ip))
Expand Down Expand Up @@ -170,7 +177,10 @@ def test_neighbor_link_down(testbed_params, setup_counters, duthosts, rand_one_d
if port != mock_server["server_dst_port"]])
logging.info("Selected port %s to send traffic", rx_port)

src_ip = MOCK_DEST_IP
if is_ipv6_only_topology(tbinfo):
src_ip = MOCK_DEST_IP_V6
else:
src_ip = MOCK_DEST_IP
pkt = generate_dropped_packet(rx_port, src_ip, mock_server["server_dst_addr"])

try:
Expand Down Expand Up @@ -198,7 +208,7 @@ def test_neighbor_link_down(testbed_params, setup_counters, duthosts, rand_one_d


@pytest.mark.parametrize("drop_reason", ["DIP_LINK_LOCAL"])
def test_dip_link_local(testbed_params, setup_counters, duthosts, rand_one_dut_hostname,
def test_dip_link_local(testbed_params, setup_counters, duthosts, tbinfo, rand_one_dut_hostname,
toggle_all_simulator_ports_to_rand_selected_tor_m, # noqa F811
setup_standby_ports_on_rand_unselected_tor, # noqa F811
send_dropped_traffic, drop_reason, add_default_route_to_dut, generate_dropped_packet):
Expand All @@ -214,8 +224,13 @@ def test_dip_link_local(testbed_params, setup_counters, duthosts, rand_one_dut_h
rx_port = random.choice(list(testbed_params["physical_port_map"].keys()))
logging.info("Selected port %s to send traffic", rx_port)

src_ip = "10.10.10.10"
pkt = generate_dropped_packet(rx_port, src_ip, LINK_LOCAL_IP)
if is_ipv6_only_topology(tbinfo):
src_ip = "2001:db8::10:10"
dst_ip = LINK_LOCAL_IP_V6
else:
src_ip = "10.10.10.10"
dst_ip = LINK_LOCAL_IP
pkt = generate_dropped_packet(rx_port, src_ip, dst_ip)

try:
send_dropped_traffic(counter_type, pkt, rx_port)
Expand All @@ -225,7 +240,7 @@ def test_dip_link_local(testbed_params, setup_counters, duthosts, rand_one_dut_h


@pytest.mark.parametrize("drop_reason", ["SIP_LINK_LOCAL"])
def test_sip_link_local(testbed_params, setup_counters, duthosts, rand_one_dut_hostname,
def test_sip_link_local(testbed_params, setup_counters, duthosts, tbinfo, rand_one_dut_hostname,
toggle_all_simulator_ports_to_rand_selected_tor_m, # noqa F811
setup_standby_ports_on_rand_unselected_tor, # noqa F811
send_dropped_traffic, drop_reason, add_default_route_to_dut, generate_dropped_packet):
Expand All @@ -241,8 +256,13 @@ def test_sip_link_local(testbed_params, setup_counters, duthosts, rand_one_dut_h
rx_port = random.choice(list(testbed_params["physical_port_map"].keys()))
logging.info("Selected port %s to send traffic", rx_port)

dst_ip = "10.10.10.10"
pkt = generate_dropped_packet(rx_port, LINK_LOCAL_IP, dst_ip)
if is_ipv6_only_topology(tbinfo):
src_ip = LINK_LOCAL_IP_V6
dst_ip = "2001:db8::10:10"
else:
src_ip = LINK_LOCAL_IP
dst_ip = "10.10.10.10"
pkt = generate_dropped_packet(rx_port, src_ip, dst_ip)

try:
send_dropped_traffic(counter_type, pkt, rx_port)
Expand Down Expand Up @@ -486,24 +506,34 @@ def mock_server(fanouthosts, testbed_params, arp_responder, ptfadapter, duthosts


@pytest.fixture
def generate_dropped_packet(duthosts, rand_one_dut_hostname, testbed_params, vlan_mac):
def generate_dropped_packet(duthosts, tbinfo, rand_one_dut_hostname, testbed_params, vlan_mac):

def _get_simple_ip_packet(rx_port, src_ip, dst_ip):
dst_mac = vlan_mac if rx_port in testbed_params["vlan_ports"] \
else duthost.get_dut_iface_mac(testbed_params["physical_port_map"][rx_port])
src_mac = "DE:AD:BE:EF:12:34"
# send tagged packet for t0-backend whose vlan mode is tagged
enable_vlan = rx_port in testbed_params["vlan_ports"] and testbed_params["vlan_interface"]["type"] == "tagged"
packet_params = dict(
eth_src=src_mac,
eth_dst=dst_mac,
ip_src=src_ip,
ip_dst=dst_ip
)
if is_ipv6_only_topology(tbinfo):
packet_params = dict(
eth_src=src_mac,
eth_dst=dst_mac,
ipv6_src=src_ip,
ipv6_dst=dst_ip
)
pkt_func = testutils.simple_ipv6ip_packet
else:
packet_params = dict(
eth_src=src_mac,
eth_dst=dst_mac,
ip_src=src_ip,
ip_dst=dst_ip
)
pkt_func = testutils.simple_ip_packet
if enable_vlan:
packet_params["dl_vlan_enable"] = enable_vlan
packet_params["vlan_vid"] = int(testbed_params["vlan_interface"]["attachto"].lstrip("Vlan"))
pkt = testutils.simple_ip_packet(**packet_params)
pkt = pkt_func(**packet_params)

logging.info("Generated simple IP packet (SMAC=%s, DMAC=%s, SIP=%s, DIP=%s)",
src_mac, dst_mac, src_ip, dst_ip)
Expand All @@ -522,13 +552,15 @@ def _generate_vlan_servers(vlan_network, vlan_ports):
# - MACs are generated sequentially as offsets from VLAN_BASE_MAC_PATTERN
# - IP addresses are randomly selected from the given VLAN network
# - "Hosts" (IP/MAC pairs) are distributed evenly amongst the ports in the VLAN
addr_list = list(IPNetwork(vlan_network))
network = ipaddress.ip_network(vlan_network, strict=False)
# Limit range size to avoid len error in random.sample with large IPv6 networks in Python < 3.9
max_range = min(network.num_addresses - 1, VLAN_HOSTS * 1000)
offsets = random.sample(range(1, max_range + 1), VLAN_HOSTS)

for counter, i in enumerate(range(2, VLAN_HOSTS + 2)):
mac = VLAN_BASE_MAC_PATTERN.format(counter)
port = vlan_ports[i % len(vlan_ports)]
addr = random.choice(addr_list)
# Ensure that we won't get a duplicate ip address
addr_list.remove(addr)
addr = network.network_address + offsets[counter]

vlan_host_map[port][str(addr)] = mac

Expand Down
Loading