Skip to content

[Snappi] Adding support for tgen_port_info in PFC tx_drop_counter and global_pause test-script. #22380

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
amitpawar12 wants to merge 3 commits into from
Closed

[Snappi] Adding support for tgen_port_info in PFC tx_drop_counter and global_pause test-script. #22380

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0c70343
snappi: issue-22363 support of tgen_port_info in pfc tx_drop_counter …
amitpawar12 Feb 12, 2026
4573c81
snappi: issue-22363 removed multidut-port-info import from pfc tx_dro…
amitpawar12 Feb 12, 2026
a406a67
snappi:issue-22363 support of tgen_port_info in PFC test_global_pause…
amitpawar12 Feb 12, 2026
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
74 changes: 24 additions & 50 deletions tests/snappi_tests/pfc/test_global_pause_with_snappi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,34 +1,36 @@
import pytest
import logging
from tests.common.helpers.assertions import pytest_require, pytest_assert # noqa: F401
from tests.common.helpers.assertions import pytest_require, pytest_assert # noqa: F401
from tests.common.fixtures.conn_graph_facts import conn_graph_facts, fanout_graph_facts, \
fanout_graph_facts_multidut # noqa: F401
fanout_graph_facts_multidut # noqa: F401
from tests.common.snappi_tests.snappi_fixtures import snappi_api_serv_ip, snappi_api_serv_port, \
snappi_api, get_snappi_ports, is_snappi_multidut, \
get_snappi_ports_single_dut, snappi_testbed_config, \
get_snappi_ports_multi_dut, snappi_dut_base_config, cleanup_config, get_snappi_ports_for_rdma # noqa: F401
from tests.common.snappi_tests.qos_fixtures import lossless_prio_list, prio_dscp_map, disable_pfcwd # noqa: F401
from tests.snappi_tests.pfc.files.helper import run_pfc_test # noqa: F401
snappi_api, snappi_dut_base_config, get_snappi_ports_for_rdma, cleanup_config, \
snappi_testbed_config, get_snappi_ports_single_dut, snappi_port_selection, \
get_snappi_ports, tgen_port_info, is_snappi_multidut, get_snappi_ports_multi_dut # noqa: F401
from tests.common.snappi_tests.qos_fixtures import lossless_prio_list, prio_dscp_map, disable_pfcwd # noqa: F401
from tests.snappi_tests.pfc.files.helper import run_pfc_test # noqa: F401
from tests.common.snappi_tests.snappi_test_params import SnappiTestParams
from tests.snappi_tests.variables import MULTIDUT_PORT_INFO, MULTIDUT_TESTBED
from tests.snappi_tests.cisco.helper import disable_voq_watchdog # noqa: F401
from tests.snappi_tests.cisco.helper import disable_voq_watchdog # noqa: F401

logger = logging.getLogger(__name__)

pytestmark = [pytest.mark.topology('multidut-tgen', 'tgen')]


@pytest.mark.parametrize("multidut_port_info", MULTIDUT_PORT_INFO[MULTIDUT_TESTBED])
def test_global_pause(snappi_api, # noqa: F811
conn_graph_facts, # noqa: F811
fanout_graph_facts_multidut, # noqa: F811
get_snappi_ports, # noqa: F811
@pytest.fixture(autouse=True, scope='module')
def number_of_tx_rx_ports():
yield (1, 1)


def test_global_pause(snappi_api, # noqa: F811
conn_graph_facts, # noqa: F811
fanout_graph_facts_multidut, # noqa: F811
duthosts,
prio_dscp_map, # noqa: F811
lossless_prio_list, # noqa: F811
tbinfo, # noqa: F811
multidut_port_info,
disable_pfcwd): # noqa: F811
prio_dscp_map, # noqa: F811
lossless_prio_list, # noqa: F811
tbinfo, # noqa: F811
tgen_port_info, # noqa: F811
disable_pfcwd): # noqa: F811
"""
Test if IEEE 802.3X pause (a.k.a., global pause) will impact any priority

Expand All @@ -41,39 +43,13 @@ def test_global_pause(snappi_api, # noqa: F811
lossless_prio_list (pytest fixture): list of all the lossless priorities
prio_dscp_map (pytest fixture): priority vs. DSCP map (key = priority).
tbinfo (pytest fixture): fixture provides information about testbed
get_snappi_ports (pytest fixture): gets snappi ports and connected DUT port info and returns as a list
tgen_port_info (pytest fixture): fixtures returns port_config_list, snappi_ports and testbed_config
Returns:
N/A
"""
snappi_port_list = get_snappi_ports

tbname = tbinfo.get('conf-name', None)

tx_port_count = 1
rx_port_count = 1
testbed_config, port_config_list, snappi_ports = tgen_port_info

pytest_require(len(snappi_port_list) >= tx_port_count + rx_port_count,
"Need Minimum of 2 ports defined in ansible/files/*links.csv file")

for testbed_subtype, rdma_ports in multidut_port_info.items():
pytest_require(len(rdma_ports['tx_ports']) >= tx_port_count,
'MULTIDUT_PORT_INFO doesn\'t have the required Tx ports defined for \
testbed {}, subtype {} in variables.py'.
format(MULTIDUT_TESTBED, testbed_subtype))

pytest_require(len(rdma_ports['rx_ports']) >= rx_port_count,
'MULTIDUT_PORT_INFO doesn\'t have the required Rx ports defined for \
testbed {}, subtype {} in variables.py'.
format(tbname, testbed_subtype))
logger.info('Running test for testbed subtype: {}'.format(testbed_subtype))
if is_snappi_multidut(duthosts):
snappi_ports = get_snappi_ports_for_rdma(snappi_port_list, rdma_ports,
tx_port_count, rx_port_count, MULTIDUT_TESTBED)
else:
snappi_ports = snappi_port_list
testbed_config, port_config_list, snappi_ports = snappi_dut_base_config(duthosts,
snappi_ports,
snappi_api)
logger.info("Snappi Ports : {}".format(snappi_ports))

all_prio_list = prio_dscp_map.keys()
test_prio_list = lossless_prio_list
Expand All @@ -93,5 +69,3 @@ def test_global_pause(snappi_api, # noqa: F811
prio_dscp_map=prio_dscp_map,
test_traffic_pause=False,
snappi_extra_params=snappi_extra_params)

cleanup_config(duthosts, snappi_ports)
31 changes: 15 additions & 16 deletions tests/snappi_tests/pfc/test_tx_drop_counter_with_snappi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,35 +1,34 @@
import pytest
import logging

from tests.common.helpers.assertions import pytest_require # noqa: F401
from tests.common.fixtures.conn_graph_facts import conn_graph_facts,\
fanout_graph_facts, fanout_graph_facts_multidut # noqa: F401
from tests.common.helpers.assertions import pytest_require # noqa: F401
from tests.common.fixtures.conn_graph_facts import conn_graph_facts, \
fanout_graph_facts, fanout_graph_facts_multidut # noqa: F401
from tests.common.snappi_tests.snappi_fixtures import snappi_api_serv_ip, snappi_api_serv_port, \
snappi_api, snappi_dut_base_config, get_snappi_ports_for_rdma, cleanup_config, get_snappi_ports_multi_dut, \
snappi_testbed_config, get_snappi_ports_single_dut, \
get_snappi_ports, is_snappi_multidut # noqa: F401
snappi_api, snappi_dut_base_config, get_snappi_ports_for_rdma, cleanup_config, \
snappi_testbed_config, get_snappi_ports_single_dut, snappi_port_selection, \
get_snappi_ports, tgen_port_info, is_snappi_multidut, get_snappi_ports_multi_dut # noqa: F401
from tests.common.snappi_tests.qos_fixtures import prio_dscp_map, \
lossless_prio_list, disable_pfcwd # noqa: F401
lossless_prio_list, disable_pfcwd # noqa: F401
from tests.snappi_tests.pfc.files.helper import run_tx_drop_counter
from tests.common.snappi_tests.snappi_test_params import SnappiTestParams
from tests.snappi_tests.files.helper import multidut_port_info, setup_ports_and_dut # noqa: F401
from tests.snappi_tests.cisco.helper import disable_voq_watchdog # noqa: F401
from tests.snappi_tests.cisco.helper import disable_voq_watchdog # noqa: F401

logger = logging.getLogger(__name__)
pytestmark = [pytest.mark.topology('multidut-tgen', 'tgen')]


@pytest.fixture(autouse=True)
@pytest.fixture(autouse=True, scope='module')
def number_of_tx_rx_ports():
yield (1, 1)


def test_tx_drop_counter(
snappi_api, # noqa: F811
lossless_prio_list, # noqa: F811
prio_dscp_map, # noqa: F811
setup_ports_and_dut, # noqa: F811
disable_pfcwd # noqa: F811
snappi_api, # noqa: F811
lossless_prio_list, # noqa: F811
prio_dscp_map, # noqa: F811
tgen_port_info, # noqa: F811
disable_pfcwd # noqa: F811
):
"""
Test if device under test (DUT) is incrementing
Expand All @@ -52,7 +51,7 @@ def test_tx_drop_counter(
N/A
"""

testbed_config, port_config_list, snappi_ports = setup_ports_and_dut
testbed_config, port_config_list, snappi_ports = tgen_port_info

logger.info("Snappi Ports : {}".format(snappi_ports))

Expand Down
Loading