| #!/usr/bin/env python3 |
| # SPDX-License-Identifier: Linux-OpenIB |
| # Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES |
| """Unit tests for NVCX_Topo and its nested NIC / Board types. |
| |
| Tests import rdma_topo via load_rdma_topo() (which strips the bare main() |
| call) and use lightweight mock PCIDevice objects so no sysfs or dump file |
| is needed. |
| """ |
| from __future__ import annotations |
| |
| import sys |
| import types |
| |
| from pathlib import Path |
| from typing import Dict, List, Optional |
| from unittest.mock import MagicMock |
| |
| try: |
| import pytest |
| except ImportError: |
| print("Missing dependency: pytest", file=sys.stderr) |
| print("Install with: pip3 install pytest", file=sys.stderr) |
| sys.exit(1) |
| |
| HERE = Path(__file__).resolve().parent |
| RDMA_TOPO = HERE / "rdma_topo" |
| |
| |
| def _strip_trailing_main_call(src: str) -> str: |
| """rdma_topo ends with bare main(); skip it so test import does not run CLI.""" |
| lines = src.splitlines() |
| i = len(lines) - 1 |
| while i >= 0 and lines[i].strip() == "": |
| i -= 1 |
| if i < 0: |
| return src |
| if lines[i].split("#", 1)[0].strip() == "main()": |
| return "\n".join(lines[:i]) + ("\n" if i else "") |
| return src |
| |
| |
| def load_rdma_topo(): |
| raw = RDMA_TOPO.read_text(encoding="utf-8") |
| code_s = _strip_trailing_main_call(raw) |
| mod = types.ModuleType("rdma_topo") |
| mod.__file__ = str(RDMA_TOPO) |
| mod.__name__ = "rdma_topo" |
| mod.__package__ = "" |
| sys.modules["rdma_topo"] = mod |
| exec(compile(code_s, str(RDMA_TOPO), "exec"), mod.__dict__) |
| return mod |
| |
| |
| load_rdma_topo() |
| |
| from rdma_topo import PCIBDF, NVCX_Topo, TopoUnexpectedError |
| |
| # --------------------------------------------------------------------------- |
| # Helpers |
| # --------------------------------------------------------------------------- |
| |
| |
| def make_dev( |
| bdf_str: str, |
| vpd_sn: Optional[str] = None, |
| parent=None, |
| has_ats: bool = False, |
| subsystems: Optional[Dict] = None, |
| ) -> MagicMock: |
| """Return a mock PCIDevice for use with NVCX_Topo code.""" |
| dev = MagicMock(name=f"PCIDevice({bdf_str})") |
| seg, bus, rest = bdf_str.split(":") |
| d, func = rest.split(".") |
| dev.bdf = PCIBDF(seg, bus, d, func) |
| dev.vpd_sn = vpd_sn |
| dev.parent = parent |
| dev.has_ats = has_ats |
| dev.get_subsystems.return_value = subsystems or {} |
| return dev |
| |
| |
| def make_parent(bdf_str: str) -> MagicMock: |
| """Return a mock parent PCIDevice (used as NIC.parent).""" |
| p = MagicMock(name=f"ParentDevice({bdf_str})") |
| seg, bus, rest = bdf_str.split(":") |
| d, func = rest.split(".") |
| p.bdf = PCIBDF(seg, bus, d, func) |
| return p |
| |
| |
| def make_nic( |
| pf_bdfs: List[str], |
| parent_bdf: Optional[str] = None, |
| vpd_sn: Optional[str] = None, |
| has_ats: bool = False, |
| subsystems: Optional[Dict] = None, |
| ) -> NVCX_Topo.NIC: |
| """Construct a NVCX_Topo.NIC from mock PCIDevices.""" |
| parent = make_parent(parent_bdf) if parent_bdf else None |
| devs = [ |
| make_dev( |
| bdf, vpd_sn=vpd_sn, parent=parent, has_ats=has_ats, subsystems=subsystems |
| ) |
| for bdf in pf_bdfs |
| ] |
| return NVCX_Topo.NIC(set(devs)) |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo.NIC — constructor |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_NIC_Constructor: |
| def test_single_pf_no_parent(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf=None, vpd_sn=None) |
| assert nic.parent is None |
| assert nic.vpd_sn is None |
| assert len(nic.pfs) == 1 |
| |
| def test_single_pf_with_parent_and_sn(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| assert str(nic.parent.bdf) == "0000:00:00.0" |
| assert nic.vpd_sn == "SN123" |
| |
| def test_multi_pf_same_parent_same_sn(self): |
| nic = make_nic( |
| ["0000:00:01.0", "0000:00:02.0"], |
| parent_bdf="0000:00:00.0", |
| vpd_sn="SN123", |
| ) |
| assert len(nic.pfs) == 2 |
| |
| def test_multi_pf_different_parent_raises(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev_a = make_dev("0000:00:01.0", parent=parent_a, vpd_sn="SN") |
| dev_b = make_dev("0000:00:02.0", parent=parent_b, vpd_sn="SN") |
| with pytest.raises(TopoUnexpectedError, match="same parent"): |
| NVCX_Topo.NIC({dev_a, dev_b}) |
| |
| def test_multi_pf_different_sn_raises(self): |
| parent = make_parent("0000:00:00.0") |
| dev_a = make_dev("0000:00:01.0", parent=parent, vpd_sn="SN1") |
| dev_b = make_dev("0000:00:02.0", parent=parent, vpd_sn="SN2") |
| with pytest.raises(TopoUnexpectedError, match="same VPD SN"): |
| NVCX_Topo.NIC({dev_a, dev_b}) |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo.NIC — primary_pf |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_NIC_PrimaryPf: |
| def test_multi_pf_returns_min_bdf(self): |
| parent = make_parent("0000:00:00.0") |
| dev_lo = make_dev("0000:00:01.0", parent=parent, vpd_sn="SN") |
| dev_hi = make_dev("0000:00:02.0", parent=parent, vpd_sn="SN") |
| nic = NVCX_Topo.NIC({dev_lo, dev_hi}) |
| assert nic.primary_pf is dev_lo |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo.NIC — to_dict |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_NIC_ToDict: |
| def test_has_parent_bdf(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| assert nic.to_dict()["parent_bdf"] == "0000:00:00.0" |
| |
| def test_no_parent_gives_unknown(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf=None, vpd_sn=None) |
| assert nic.to_dict()["parent_bdf"] == "UNKNOWN" |
| |
| |
| def test_ats_taken_from_min_bdf_pf(self): |
| parent = make_parent("0000:00:00.0") |
| dev_lo = make_dev("0000:00:01.0", parent=parent, vpd_sn="SN", has_ats=False) |
| dev_hi = make_dev("0000:00:02.0", parent=parent, vpd_sn="SN", has_ats=True) |
| nic = NVCX_Topo.NIC({dev_lo, dev_hi}) |
| assert nic.to_dict()["ats"] == False |
| nic = NVCX_Topo.NIC({dev_hi, dev_lo}) |
| assert nic.to_dict()["ats"] == False |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo.NIC — __str__ |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_NIC_Str: |
| def test_single_pf_starts_with_nic_parent_header(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| assert str(nic).startswith("RDMA NIC Parent=0000:00:00.0") |
| |
| def test_single_pf_no_parent_shows_unknown(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf=None, vpd_sn=None) |
| assert str(nic).startswith("RDMA NIC Parent=UNKNOWN") |
| |
| def test_single_pf_ats_no(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| assert "NIC ATS: no" in str(nic) |
| |
| def test_single_pf_ats_yes(self): |
| nic = make_nic( |
| ["0000:00:01.0"], |
| parent_bdf="0000:00:00.0", |
| vpd_sn="SN123", |
| has_ats=True, |
| ) |
| assert "NIC ATS: yes" in str(nic) |
| |
| def test_no_trailing_newline(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| assert not str(nic).endswith("\n") |
| |
| def test_single_pf_with_infiniband(self): |
| nic = make_nic( |
| ["0000:00:01.0"], |
| parent_bdf="0000:00:00.0", |
| vpd_sn="SN123", |
| subsystems={"infiniband": {"mlx5_0"}}, |
| ) |
| assert "RDMA device: mlx5_0" in str(nic) |
| |
| def test_single_pf_with_net(self): |
| nic = make_nic( |
| ["0000:00:01.0"], |
| parent_bdf="0000:00:00.0", |
| vpd_sn="SN123", |
| subsystems={"net": {"eth0"}}, |
| ) |
| assert "Net device: eth0" in str(nic) |
| |
| def test_multi_pf_shows_sorted_pci_device_list(self): |
| nic = make_nic( |
| ["0000:00:02.0", "0000:00:01.0"], |
| parent_bdf="0000:00:00.0", |
| vpd_sn="SN123", |
| ) |
| assert "NIC PCI devices: 0000:00:01.0, 0000:00:02.0" in str(nic) |
| |
| def test_multi_pf_subsystems_merged(self): |
| parent = make_parent("0000:00:00.0") |
| dev1 = make_dev( |
| "0000:00:01.0", |
| parent=parent, |
| vpd_sn="SN", |
| subsystems={"infiniband": {"mlx5_0"}, "net": {"eth0"}}, |
| ) |
| dev2 = make_dev( |
| "0000:00:02.0", |
| parent=parent, |
| vpd_sn="SN", |
| subsystems={"infiniband": {"mlx5_1"}, "net": {"eth1"}}, |
| ) |
| nic = NVCX_Topo.NIC({dev1, dev2}) |
| result = str(nic) |
| assert "RDMA devices: mlx5_0, mlx5_1" in result |
| assert "Net devices: eth0, eth1" in result |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo.Board — constructor |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_Board_Constructor: |
| def test_single_nic_sn_stored(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| board = NVCX_Topo.Board({nic}) |
| assert board.sn == "SN123" |
| assert nic in board.nics |
| |
| def test_multi_nic_same_sn(self): |
| nic_a = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| nic_b = make_nic(["0000:00:11.0"], parent_bdf="0000:00:10.0", vpd_sn="SN123") |
| board = NVCX_Topo.Board({nic_a, nic_b}) |
| assert len(board.nics) == 2 |
| assert board.sn == "SN123" |
| |
| def test_multi_nic_different_sn_raises(self): |
| nic_a = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN1") |
| nic_b = make_nic(["0000:00:11.0"], parent_bdf="0000:00:10.0", vpd_sn="SN2") |
| with pytest.raises(TopoUnexpectedError, match="same VPD SN"): |
| NVCX_Topo.Board({nic_a, nic_b}) |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo.Board — to_dict |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_Board_ToDict: |
| def test_has_board_sn(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| board = NVCX_Topo.Board({nic}) |
| assert board.to_dict()["board_sn"] == "SN123" |
| |
| def test_none_sn_gives_unknown(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf=None, vpd_sn=None) |
| board = NVCX_Topo.Board({nic}) |
| assert board.to_dict()["board_sn"] == "UNKNOWN" |
| |
| def test_nics_list_length(self): |
| nic_a = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| nic_b = make_nic(["0000:00:11.0"], parent_bdf="0000:00:10.0", vpd_sn="SN123") |
| board = NVCX_Topo.Board({nic_a, nic_b}) |
| assert len(board.to_dict()["nics"]) == 2 |
| |
| def test_nics_list_contains_nic_dicts(self): |
| nic = make_nic( |
| ["0000:00:01.0", "0000:00:02.0"], |
| parent_bdf="0000:00:00.0", |
| vpd_sn="SN123", |
| ) |
| board = NVCX_Topo.Board({nic}) |
| nic_dict = board.to_dict()["nics"][0] |
| assert nic_dict == nic.to_dict() |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo.Board — __str__ |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_Board_Str: |
| def test_single_nic_starts_with_board_header(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| board = NVCX_Topo.Board({nic}) |
| assert str(board).startswith("RDMA NIC Board=SN123") |
| |
| def test_none_sn_shows_unknown(self): |
| nic = make_nic(["0000:00:01.0"], parent_bdf=None, vpd_sn=None) |
| board = NVCX_Topo.Board({nic}) |
| assert str(board).startswith("RDMA NIC Board=UNKNOWN") |
| |
| def test_multi_nic_includes_nic_parent_headers(self): |
| nic_a = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| nic_b = make_nic(["0000:00:11.0"], parent_bdf="0000:00:10.0", vpd_sn="SN123") |
| board = NVCX_Topo.Board({nic_a, nic_b}) |
| result = str(board) |
| assert "RDMA NIC Parent=0000:00:00.0" in result |
| assert "RDMA NIC Parent=0000:00:10.0" in result |
| |
| def test_multi_nic_nic_body_indented(self): |
| nic_a = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| nic_b = make_nic(["0000:00:11.0"], parent_bdf="0000:00:10.0", vpd_sn="SN123") |
| board = NVCX_Topo.Board({nic_a, nic_b}) |
| result = str(board) |
| assert "\t\tNIC ATS: no" in result |
| |
| def test_multi_nic_no_trailing_newline(self): |
| nic_a = make_nic(["0000:00:01.0"], parent_bdf="0000:00:00.0", vpd_sn="SN123") |
| nic_b = make_nic(["0000:00:11.0"], parent_bdf="0000:00:10.0", vpd_sn="SN123") |
| board = NVCX_Topo.Board({nic_a, nic_b}) |
| assert not str(board).endswith("\n") |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo — constructor |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_Constructor: |
| def test_single_pf_with_parent_and_sn_yields_one_board(self): |
| parent = make_parent("0000:00:00.0") |
| dev = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev}) |
| assert len(topo.boards) == 1 |
| |
| def test_two_pfs_same_parent_same_sn_one_nic_one_board(self): |
| parent = make_parent("0000:00:00.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| dev2 = make_dev("0000:00:02.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev1, dev2}) |
| assert len(topo.boards) == 1 |
| assert len(topo.boards[0].nics) == 1 |
| assert len(topo.boards[0].nics[0].pfs) == 2 |
| |
| def test_two_pfs_different_parent_same_sn_two_nics_one_board(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent_a) |
| dev2 = make_dev("0000:00:11.0", vpd_sn="SN123", parent=parent_b) |
| topo = NVCX_Topo({dev1, dev2}) |
| assert len(topo.boards) == 1 |
| assert len(topo.boards[0].nics) == 2 |
| |
| def test_two_pfs_different_sn_two_boards(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN1", parent=parent_a) |
| dev2 = make_dev("0000:00:11.0", vpd_sn="SN2", parent=parent_b) |
| topo = NVCX_Topo({dev1, dev2}) |
| assert len(topo.boards) == 2 |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo — pfs / primary_pf |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_Pfs: |
| def test_pfs_returns_all_devs(self): |
| parent = make_parent("0000:00:00.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| dev2 = make_dev("0000:00:02.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev1, dev2}) |
| assert topo.pfs == {dev1, dev2} |
| |
| def test_primary_pf_is_min_bdf(self): |
| parent = make_parent("0000:00:00.0") |
| dev_lo = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| dev_hi = make_dev("0000:00:02.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev_lo, dev_hi}) |
| assert topo.primary_pf is dev_lo |
| |
| def test_primary_pf_across_boards(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev_lo = make_dev("0000:00:01.0", vpd_sn="SN1", parent=parent_a) |
| dev_hi = make_dev("0000:00:11.0", vpd_sn="SN2", parent=parent_b) |
| topo = NVCX_Topo({dev_lo, dev_hi}) |
| assert topo.primary_pf is dev_lo |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo — to_dict |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_ToDict: |
| def test_single_pf_returns_flat_dict(self): |
| parent = make_parent("0000:00:00.0") |
| dev = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev}) |
| result = topo.to_dict() |
| assert result["rdma_nic_pf_bdf"] == "0000:00:01.0" |
| assert result["rdma_nic_ats"] == False |
| |
| def test_single_board_multi_pf_nic(self): |
| parent = make_parent("0000:00:00.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| dev2 = make_dev("0000:00:02.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev1, dev2}) |
| result = topo.to_dict() |
| boards = result["rdma_nic_boards"] |
| assert len(boards) == 1 |
| board = boards[0] |
| assert board["board_sn"] == "SN123" |
| assert len(board["nics"]) == 1 |
| nic_dict = board["nics"][0] |
| assert nic_dict["parent_bdf"] == "0000:00:00.0" |
| assert "ats" in nic_dict |
| assert nic_dict["pf_bdfs"] == ["0000:00:01.0", "0000:00:02.0"] |
| |
| def test_multi_board_multi_pf_nics(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN1", parent=parent_a) |
| dev2 = make_dev("0000:00:02.0", vpd_sn="SN1", parent=parent_a) |
| dev3 = make_dev("0000:00:11.0", vpd_sn="SN2", parent=parent_b) |
| dev4 = make_dev("0000:00:12.0", vpd_sn="SN2", parent=parent_b) |
| topo = NVCX_Topo({dev1, dev2, dev3, dev4}) |
| result = topo.to_dict() |
| assert len(result["rdma_nic_boards"]) == 2 |
| |
| def test_single_board_multi_nic_multi_pf(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent_a) |
| dev2 = make_dev("0000:00:02.0", vpd_sn="SN123", parent=parent_a) |
| dev3 = make_dev("0000:00:11.0", vpd_sn="SN123", parent=parent_b) |
| dev4 = make_dev("0000:00:12.0", vpd_sn="SN123", parent=parent_b) |
| topo = NVCX_Topo({dev1, dev2, dev3, dev4}) |
| result = topo.to_dict() |
| boards = result["rdma_nic_boards"] |
| assert len(boards) == 1 |
| assert len(boards[0]["nics"]) == 2 |
| |
| def test_boards_sorted_by_sn_in_to_dict(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev_b = make_dev("0000:00:11.0", vpd_sn="SN_B", parent=parent_b) |
| dev_a = make_dev("0000:00:01.0", vpd_sn="SN_A", parent=parent_a) |
| topo = NVCX_Topo({dev_a, dev_b}) |
| boards = topo.to_dict()["rdma_nic_boards"] |
| assert boards[0]["board_sn"] == "SN_A" |
| assert boards[1]["board_sn"] == "SN_B" |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo — topo_str_key |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_TopoStrKey: |
| def test_single_pf_returns_nic_bdf(self): |
| parent = make_parent("0000:00:00.0") |
| dev = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev}) |
| assert topo.topo_str_key() == "RDMA NIC=0000:00:01.0" |
| |
| def test_multi_pf_returns_empty(self): |
| parent = make_parent("0000:00:00.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| dev2 = make_dev("0000:00:02.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev1, dev2}) |
| assert topo.topo_str_key() == "" |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo — topo_str |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_TopoStr: |
| def test_single_pf_contains_ats(self): |
| parent = make_parent("0000:00:00.0") |
| dev = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev}) |
| result = topo.topo_str() |
| assert "NIC ATS: no" in result |
| assert not result.endswith("\n") |
| |
| def test_single_board_single_nic_multi_pf(self): |
| parent = make_parent("0000:00:00.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent) |
| dev2 = make_dev("0000:00:02.0", vpd_sn="SN123", parent=parent) |
| topo = NVCX_Topo({dev1, dev2}) |
| result = topo.topo_str() |
| assert "NIC ATS: no" in result |
| assert "NIC PCI devices: 0000:00:01.0, 0000:00:02.0" in result |
| |
| def test_single_board_multi_nic_has_nic_parent_headers(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN123", parent=parent_a) |
| dev2 = make_dev("0000:00:11.0", vpd_sn="SN123", parent=parent_b) |
| topo = NVCX_Topo({dev1, dev2}) |
| result = topo.topo_str() |
| assert "RDMA NIC Parent=0000:00:00.0" in result |
| assert "RDMA NIC Parent=0000:00:10.0" in result |
| |
| def test_multi_board_has_nic_parent_headers(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN1", parent=parent_a) |
| dev2 = make_dev("0000:00:11.0", vpd_sn="SN2", parent=parent_b) |
| topo = NVCX_Topo({dev1, dev2}) |
| result = topo.topo_str() |
| assert "RDMA NIC Parent=0000:00:00.0" in result |
| assert "RDMA NIC Parent=0000:00:10.0" in result |
| |
| def test_multi_board_body_indented(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN1", parent=parent_a) |
| dev2 = make_dev("0000:00:11.0", vpd_sn="SN2", parent=parent_b) |
| topo = NVCX_Topo({dev1, dev2}) |
| result = topo.topo_str() |
| assert "\t\t\tNIC ATS: no" in result |
| |
| def test_multi_board_no_trailing_newline(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev1 = make_dev("0000:00:01.0", vpd_sn="SN1", parent=parent_a) |
| dev2 = make_dev("0000:00:11.0", vpd_sn="SN2", parent=parent_b) |
| topo = NVCX_Topo({dev1, dev2}) |
| assert not topo.topo_str().endswith("\n") |
| |
| |
| # --------------------------------------------------------------------------- |
| # NVCX_Topo — board / NIC ordering |
| # --------------------------------------------------------------------------- |
| |
| |
| class TestNVCX_Topo_Ordering: |
| def test_boards_sorted_by_sn(self): |
| parent_a = make_parent("0000:00:00.0") |
| parent_b = make_parent("0000:00:10.0") |
| dev_b = make_dev("0000:00:11.0", vpd_sn="SN_B", parent=parent_b) |
| dev_a = make_dev("0000:00:01.0", vpd_sn="SN_A", parent=parent_a) |
| topo = NVCX_Topo({dev_a, dev_b}) |
| assert topo.boards[0].sn == "SN_A" |
| assert topo.boards[1].sn == "SN_B" |
| |
| def test_board_nics_sorted_by_parent_bdf(self): |
| parent_lo = make_parent("0000:00:00.0") |
| parent_hi = make_parent("0000:00:10.0") |
| dev_hi = make_dev("0000:00:11.0", vpd_sn="SN", parent=parent_hi) |
| dev_lo = make_dev("0000:00:01.0", vpd_sn="SN", parent=parent_lo) |
| topo = NVCX_Topo({dev_lo, dev_hi}) |
| board = topo.boards[0] |
| assert str(board.nics[0].parent.bdf) == "0000:00:00.0" |
| assert str(board.nics[1].parent.bdf) == "0000:00:10.0" |
| |
| def test_board_nic_without_parent_sorts_after_nic_with_parent(self): |
| parent = make_parent("0000:00:00.0") |
| dev_parented = make_dev("0000:00:01.0", vpd_sn="SN", parent=parent) |
| dev_orphan = make_dev("0000:00:02.0", vpd_sn="SN", parent=None) |
| nic_parented = NVCX_Topo.NIC({dev_parented}) |
| nic_orphan = NVCX_Topo.NIC({dev_orphan}) |
| board = NVCX_Topo.Board({nic_parented, nic_orphan}) |
| assert board.nics[0] is nic_parented |
| assert board.nics[1] is nic_orphan |
