[Test gap]Add new testcase test_everflow_packet_format, verify that mirrored packets do not contain VLAN tags or unexpected fields. (sonic-net#17500)

What is the motivation for this PR?
[Test gap]Add new testcase test_everflow_packet_format, verify that mirrored packets do not contain VLAN tags or unexpected fields.

How did you do it?

Config everflow
Send packet
Validate packet
Conditioanl mark is same as the previous testcase test_everflow_per_interface

How did you verify/test it?
Run the new testcase test_everflow_packet_format on physical testbed and check the result.

Signed-off-by: Chun'ang Li <chunangli@microsoft.com>

Author: lerry-lee

tests/common/plugins/conditional_mark/tests_mark_conditions.yaml

@@ -849,6 +849,36 @@ everflow/test_everflow_per_interface.py:
       - "platform in ['x86_64-8800_lc_48h_o-r0', 'x86_64-8800_lc_48h-r0']"
       - "(is_multi_asic==True) and https://github.com/sonic-net/sonic-buildimage/issues/11776"
 
+everflow/test_everflow_per_interface.py::test_everflow_packet_format[ipv6-default]:
+  skip:
+    reason: "Skip everflow packet integrity IPv6 test on unsupported platforms"
+    conditions_logical_operator: or
+    conditions:
+      - "asic_type in ['cisco-8000', 'marvell', 'mellanox'] or (asic_subtype in ['broadcom-dnx'] and https://github.com/sonic-net/sonic-swss/issues/2204)"
+      - "'dualtor' in topo_name"
+      - "platform in ['x86_64-8800_lc_48h_o-r0', 'x86_64-8800_lc_48h-r0']"
+      - "(is_multi_asic==True) and https://github.com/sonic-net/sonic-buildimage/issues/11776"
+
+everflow/test_everflow_per_interface.py::test_everflow_packet_format[ipv6-m0_l3_scenario]:
+  skip:
+    reason: "Skip m0 everflow packet integrity IPv6 test on unsupported platforms"
+    conditions_logical_operator: or
+    conditions:
+      - "asic_type in ['marvell']"
+      - "'dualtor' in topo_name"
+      - "platform in ['x86_64-8800_lc_48h_o-r0', 'x86_64-8800_lc_48h-r0']"
+      - "(is_multi_asic==True) and https://github.com/sonic-net/sonic-buildimage/issues/11776"
+
+everflow/test_everflow_per_interface.py::test_everflow_packet_format[ipv6-m0_vlan_scenario]:
+  skip:
+    reason: "Skip m0 everflow packet integrity IPv6 test on unsupported platforms"
+    conditions_logical_operator: or
+    conditions:
+      - "asic_type in ['marvell']"
+      - "'dualtor' in topo_name"
+      - "platform in ['x86_64-8800_lc_48h_o-r0', 'x86_64-8800_lc_48h-r0']"
+      - "(is_multi_asic==True) and https://github.com/sonic-net/sonic-buildimage/issues/11776"
+
 everflow/test_everflow_per_interface.py::test_everflow_per_interface[ipv6-default]:
   skip:
     reason: "Skip everflow per interface IPv6 test on unsupported platforms"

tests/conftest.py

@@ -1056,6 +1056,7 @@ def pytest_runtest_makereport(item, call):
 # DummyTestUtils would always return True for all verify function in ptf.testutils.
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_call(item):
+    # See tests/common/plugins/conditional_mark/tests_mark_conditions_skip_traffic_test.yaml
     if "skip_traffic_test" in item.keywords:
         logger.info("Got skip_traffic_test marker, will skip traffic test")
         with DummyTestUtils():

tests/everflow/test_everflow_per_interface.py

@@ -4,14 +4,18 @@
 import pytest
 import os
 import ptf.testutils as testutils
+from scapy.layers.l2 import Ether
+from scapy.contrib.mpls import MPLS
+from scapy.layers.l2 import Dot1Q
+from scapy.layers.vxlan import VXLAN
 from . import everflow_test_utilities as everflow_utils
 
-from .everflow_test_utilities import BaseEverflowTest, erspan_ip_ver    # noqa: F401
+from .everflow_test_utilities import BaseEverflowTest, erspan_ip_ver  # noqa: F401
 from .everflow_test_utilities import TEMPLATE_DIR, EVERFLOW_RULE_CREATE_TEMPLATE, \
-                                    DUT_RUN_DIR, EVERFLOW_RULE_CREATE_FILE, UP_STREAM
+    DUT_RUN_DIR, EVERFLOW_RULE_CREATE_FILE, UP_STREAM
 from tests.common.helpers.assertions import pytest_require
 
-from .everflow_test_utilities import setup_info, EVERFLOW_DSCP_RULES, STABILITY_BUFFER    # noqa: F401
+from .everflow_test_utilities import setup_info, EVERFLOW_DSCP_RULES, STABILITY_BUFFER  # noqa: F401
 from tests.common.dualtor.mux_simulator_control import toggle_all_simulator_ports_to_rand_selected_tor  # noqa: F401
 
 pytestmark = [
@@ -79,7 +83,7 @@ def build_acl_rule_vars(candidate_ports, ip_ver):
 
 
 @pytest.fixture(scope='module')
-def apply_mirror_session(setup_info, erspan_ip_ver):       # noqa F811
+def apply_mirror_session(setup_info, erspan_ip_ver):  # noqa F811
     mirror_session_info = BaseEverflowTest.mirror_session_info(
         EVERFLOW_SESSION_NAME, setup_info[UP_STREAM]['everflow_dut'].facts["asic_type"])
     logger.info("Applying mirror session to DUT")
@@ -93,7 +97,7 @@ def apply_mirror_session(setup_info, erspan_ip_ver):       # noqa F811
 
 
 @pytest.fixture(scope='module')
-def setup_mirror_session_dest_ip_route(tbinfo, setup_info, apply_mirror_session, erspan_ip_ver):       # noqa F811
+def setup_mirror_session_dest_ip_route(tbinfo, setup_info, apply_mirror_session, erspan_ip_ver):  # noqa F811
     """
     Setup the route for mirror session destination ip and update monitor port list.
     Remove the route as part of cleanup.
@@ -124,7 +128,7 @@ def ip_ver(request):
 
 
 @pytest.fixture(scope='module')
-def apply_acl_rule(setup_info, tbinfo, setup_mirror_session_dest_ip_route, ip_ver):     # noqa F811
+def apply_acl_rule(setup_info, tbinfo, setup_mirror_session_dest_ip_route, ip_ver):  # noqa F811
     """
     Apply ACL rule for matching input_ports
     """
@@ -148,7 +152,7 @@ def apply_acl_rule(setup_info, tbinfo, setup_mirror_session_dest_ip_route, ip_ve
                                                    dest=os.path.join(DUT_RUN_DIR, EVERFLOW_RULE_CREATE_FILE))
     logger.info("Applying acl rule config to DUT")
     command = "acl-loader update full {} --table_name {} --session_name {}" \
-              .format(os.path.join(DUT_RUN_DIR, EVERFLOW_RULE_CREATE_FILE), table_name, EVERFLOW_SESSION_NAME)
+        .format(os.path.join(DUT_RUN_DIR, EVERFLOW_RULE_CREATE_FILE), table_name, EVERFLOW_SESSION_NAME)
     setup_info[UP_STREAM]['everflow_dut'].shell(cmd=command)
     ret = {
         "candidate_ports": candidate_ports,
@@ -192,7 +196,7 @@ def send_and_verify_packet(ptfadapter, packet, expected_packet, tx_port, rx_port
         testutils.verify_no_packet_any(ptfadapter, pkt=expected_packet, ports=rx_ports)
 
 
-def test_everflow_per_interface(ptfadapter, setup_info, apply_acl_rule, tbinfo,                           # noqa F811
+def test_everflow_per_interface(ptfadapter, setup_info, apply_acl_rule, tbinfo,  # noqa F811
                                 toggle_all_simulator_ports_to_rand_selected_tor, ip_ver, erspan_ip_ver):  # noqa F811
     """Verify packet ingress from candidate ports are captured by EVERFLOW, while packets
     ingress from unselected ports are not captured
@@ -213,3 +217,56 @@ def test_everflow_per_interface(ptfadapter, setup_info, apply_acl_rule, tbinfo,
     for port, ptf_idx in list(everflow_config['unselected_ports'].items()):
         logger.info("Verifying packet ingress from {} is not mirrored".format(port))
         send_and_verify_packet(ptfadapter, packet, exp_packet, ptf_idx, uplink_ports, False)
+
+
+def test_everflow_packet_format(ptfadapter, setup_info, apply_acl_rule, tbinfo,  # noqa F811
+                                toggle_all_simulator_ports_to_rand_selected_tor, ip_ver, erspan_ip_ver):  # noqa F811
+    """Verify that mirrored packets do not contain VLAN tags or unexpected fields."""
+    everflow_config = apply_acl_rule
+    packet, exp_packet = generate_testing_packet(ptfadapter, setup_info[UP_STREAM]['everflow_dut'],
+                                                 everflow_config['mirror_session_info'],
+                                                 setup_info[UP_STREAM]['ingress_router_mac'], setup_info, ip_ver,
+                                                 erspan_ip_ver)
+    uplink_ports = everflow_config["monitor_port_ptf_ids"]
+
+    # Send test packet
+    candidate_port, ptf_idx = list(everflow_config['candidate_ports'].items())[0]
+    logger.info(f"Sending test packet from candidate port {candidate_port}")
+    ptfadapter.dataplane.flush()
+    testutils.send(ptfadapter, pkt=packet, port_id=ptf_idx)
+
+    # Capture mirrored packet
+    logger.info("Capturing mirrored packet to verify format")
+    res = testutils.verify_packet_any_port(ptfadapter,
+                                           pkt=exp_packet,
+                                           ports=uplink_ports,
+                                           timeout=5)
+
+    # Skip traffic test if the return value is true.
+    # See tests.conftest.pytest_runtest_call and tests.common.plugins.ptfadapter.dummy_testutils.wrapped
+    if res is True:
+        logger.info("Skipped. Ptf.testutils is set to DummyTestUtils to skip traffic test.")
+        return
+
+    port_idx, raw_captured_packet = res
+    # Ensure packet is not empty
+    assert raw_captured_packet, "Captured packet is empty or None"
+
+    captured_packet = Ether(raw_captured_packet)
+
+    # Debugging: Print packet summary if assertions fail
+    packet_summary = captured_packet.summary()
+
+    # Ensure no VLAN tag
+    assert not captured_packet.haslayer(Dot1Q), \
+        f"Mirrored packet should not contain VLAN tag: {packet_summary}"
+
+    # Check for unexpected MPLS headers
+    assert not captured_packet.haslayer(MPLS), \
+        f"Mirrored packet contains unexpected MPLS label: {packet_summary}"
+
+    # Check for unexpected VXLAN encapsulation
+    assert not captured_packet.haslayer(VXLAN), \
+        f"Mirrored packet should not have VXLAN encapsulation: {packet_summary}"
+
+    logger.info(f"Mirrored packet format verified: {packet_summary}")