26037842892

Committed 18 May 2026 01:52PM UTC coverage: 93.031%. Remained the same

Build # 26037842892

Build Type

Pull #49

github

Committed by

web-flow

Commit Message

Merge 2db5a9d56 into 62129d258

Pull Request Pull Request #49: feat: Add InterfaceType.VPORT to supported interface types for Virtualization feature

Coverage Stats

1 of 1 new or added line in 1 file covered. (100.0%)

7 existing lines in 1 file now uncovered.

8583 of 9226 relevant lines covered (93.03%)

3.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.96

/mfd_network_adapter/network_interface/feature/virtualization/linux.py

# Copyright (C) 2025 Intel Corporation
# SPDX-License-Identifier: MIT
"""Module for Virtualization feature for Linux."""

import logging
import re
from typing import List

from mfd_common_libs import add_logging_level, log_levels
from mfd_const.network import DESIGNED_NUMBER_VFS_BY_SPEED, Speed
from mfd_typing import MACAddress, DeviceID, SubDeviceID, PCIAddress
from mfd_typing.network_interface import InterfaceType

from mfd_network_adapter.data_structures import State
from mfd_network_adapter.exceptions import (
    VirtualFunctionNotFoundException,
    NetworkAdapterConfigurationException,
    NetworkInterfaceNotSupported,
)
from .base import BaseFeatureVirtualization
from .data_structures import MethodType
from ...data_structures import VlanProto, VFDetail, LinkState
from ...exceptions import VirtualizationFeatureException, VirtualizationWrongInterfaceException, DeviceSetupException

logger = logging.getLogger(__name__)
add_logging_level(level_name="MODULE_DEBUG", level_value=log_levels.MODULE_DEBUG)


class LinuxVirtualization(BaseFeatureVirtualization):
    """Linux class for Virtualization feature."""

    def _raise_error_if_not_supported_type(self) -> None:
        """
        Raise error in case current interface is not PF/BTS/VPORT.

        :raises VirtualizationWrongInterfaceException: if method is called on non PF/BTS/VPORT interface
        """
        current_type = self._interface().interface_type
        if current_type not in (InterfaceType.PF, InterfaceType.BTS, InterfaceType.VPORT):
            raise VirtualizationWrongInterfaceException(
                f"Current interface type is {current_type} but should be InterfaceType.PF or InterfaceType.BTS"
            )

    def _get_vfs_details(self) -> List[VFDetail]:
        """
        Get VF details of PF interface.

        :raises VirtualizationWrongInterfaceException: if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException: in case of command failure (rc != 0)
        :return: List of VFDetail objects
        """
        self._raise_error_if_not_supported_type()

        command = f"ip link show dev {self._interface().name}"
        pattern = (
            r"vf\s*(?P<vf_id>\d+)\s*(link/ether|MAC)\s*(?P<mac_address>[0-9a-fA-F]{2}(?::[0-9a-fA-F]{2}){5})\s*.*?,"
            r"\s*spoof checking\s*(?P<spoofchk>\w+),\s*link-state\s*(?P<link_state>\w+),\s*trust\s*("
            r"?P<trust>\w+)"
        )
        link_state_map = {"enable": LinkState.ENABLE, "disable": LinkState.DISABLE, "auto": LinkState.AUTO}

        output = self._connection.execute_command(
            command=command, custom_exception=VirtualizationFeatureException
        ).stdout
        vf_details = []
        info_match = re.finditer(pattern, output)
        for match in info_match:
            vf_details.append(
                VFDetail(
                    id=int(match.group("vf_id")),
                    mac_address=MACAddress(match.group("mac_address")),
                    spoofchk=State.ENABLED if match.group("spoofchk") == "on" else State.DISABLED,
                    link_state=link_state_map.get(match.group("link_state")),
                    trust=State.ENABLED if match.group("trust") == "on" else State.DISABLED,
                )
            )
        return vf_details

    def _get_max_vfs_by_name(self) -> int:
        """
        Get maximal number of VFs per interface based on name.

        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException in case of error
        :return: number of VFs
        """
        self._raise_error_if_not_supported_type()
        command = f"cat /sys/class/net/{self._interface().name}/device/sriov_totalvfs"
        result = self._connection.execute_command(command=command, custom_exception=VirtualizationFeatureException)

        return int(result.stdout)

    def _get_max_vfs_by_pci_address(self) -> int:
        """
        Get maximal number of VFs per interface based on PCI Address.

        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException in case of error
        :return: number of VFs
        """
        self._raise_error_if_not_supported_type()
        command = f"cat /sys/bus/pci/devices/{self._interface().pci_address}/sriov_totalvfs"
        result = self._connection.execute_command(command=command, custom_exception=VirtualizationFeatureException)

        return int(result.stdout)

    def _get_current_vfs_by_name(self) -> int:
        """
        Get number of current VFs per interface based on name.

        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException in case of error
        :return: number of VFs
        """
        self._raise_error_if_not_supported_type()
        command = f"cat /sys/class/net/{self._interface().name}/device/sriov_numvfs"
        result = self._connection.execute_command(command=command, custom_exception=VirtualizationFeatureException)

        return int(result.stdout)

    def _get_current_vfs_by_pci_address(self) -> int:
        """
        Get number of current VFs per interface based on PCI Address.

        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException in case of error
        :return: number of VFs
        """
        self._raise_error_if_not_supported_type()
        command = f"cat /sys/bus/pci/devices/{self._interface().pci_address}/sriov_numvfs"
        result = self._connection.execute_command(command=command, custom_exception=VirtualizationFeatureException)

        return int(result.stdout)

    def get_max_vfs(self) -> int:
        """Get maximal number of VFs per interface."""
        return self._get_max_vfs_by_name() if self._interface().name else self._get_max_vfs_by_pci_address()

    def get_current_vfs(self) -> int:
        """Get number of current VFs per interface."""
        return self._get_current_vfs_by_name() if self._interface().name else self._get_current_vfs_by_pci_address()

    def get_designed_number_vfs(self) -> tuple[int, int]:
        """
        Return designed max number of VFs, total and per PF.

        :return: max VFs for NIC, max VFS per PF
        """
        speed = self._interface().speed
        if speed in (Speed.G1, Speed.G10, Speed.G40, Speed.G200):
            return DESIGNED_NUMBER_VFS_BY_SPEED[speed], int(
                DESIGNED_NUMBER_VFS_BY_SPEED[speed] / self._interface().get_number_of_ports()
            )
        elif speed is Speed.G100:
            device_id = self._interface().pci_device.device_id
            sub_device_id = self._interface().pci_device.sub_device_id
            if device_id == DeviceID(0x1592) and sub_device_id == SubDeviceID(0x000E):
                designed_max_vfs = 512
                designed_max_vfs_per_pf = 256
                logger.log(
                    level=log_levels.MODULE_DEBUG,
                    msg=(
                        "Recognized Chapman Beach (2 chips on NIC)."
                        f"Return designed max VFS: {designed_max_vfs} and per PF: {designed_max_vfs_per_pf}"
                    ),
                )
                return designed_max_vfs, designed_max_vfs_per_pf
            else:
                return DESIGNED_NUMBER_VFS_BY_SPEED[Speed.G100], int(
                    DESIGNED_NUMBER_VFS_BY_SPEED[Speed.G100] / self._interface().get_number_of_ports()
                )
        else:
            raise DeviceSetupException(
                "Cannot recognize NIC. \n"
                f"Device string: {self._interface().get_device_string()}.\n"
                f"Device ID: {self._interface().pci_device.device_id}.\n"
                f"Sub Device ID: {self._interface().pci_device.sub_device_id}.\n"
            )

    def set_sriov(self, sriov_enabled: bool, no_restart: bool = False) -> None:
        """
        Set network interface SRIOV.

        :param sriov_enabled: adapter SRIOV status value to be set.
        :param no_restart: whether to restart adapter after changing its settings.
        """
        raise NotImplementedError

    def set_vmq(self, vmq_enabled: bool, no_restart: bool = False) -> None:
        """
        Set network interface VMQ.

        :param vmq_enabled: adapter VMQ status value to be set.
        :param no_restart: whether to restart adapter after changing its settings.
        """
        raise NotImplementedError

    def set_max_tx_rate(self, vf_id: int, value: int) -> None:
        """
        Set max_tx_rate VF-d parameter status.

        :param: vf_id: VF (virtual function) ID
        :param value: max_tx_rate value (Mbits)
        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises VirtualizationFeatureException if command execution fails
        """
        self._raise_error_if_not_supported_type()

        cmd = f"ip link set dev {self._interface().name} vf {vf_id} max_tx_rate {value}"
        logger.log(level=log_levels.MODULE_DEBUG, msg=f"Set max tx rate using : {cmd}")
        self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)

    def set_min_tx_rate(self, vf_id: int, value: int) -> None:
        """
        Set min_tx_rate VF-d parameter status.

        :param vf_id: VF (virtual function) ID
        :param value: min_tx_rate value (Mbits)
        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises VirtualizationFeatureException if command execution fails
        """
        self._raise_error_if_not_supported_type()
        cmd = f"ip link set dev {self._interface().name} vf {vf_id} min_tx_rate {value}"
        logger.log(level=log_levels.MODULE_DEBUG, msg=f"Set min tx rate using : {cmd}")
        self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)

    def set_trust(self, vf_id: int, state: State) -> None:
        """
        Change 'trust' setting on VF Interface.

        :param vf_id: Virtual Function ID
        :param state: State to be set: Enabled or Disabled (On/Off)
        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException in case of command error
        """
        self._raise_error_if_not_supported_type()
        value = "on" if state == State.ENABLED else "off"
        cmd = f"ip link set {self._interface().name} vf {vf_id} trust {value}"
        logger.log(level=log_levels.MODULE_DEBUG, msg=f"Set trust on VF ID: {vf_id} to '{value}'")
        self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)

    def set_spoofchk(self, vf_id: int, state: State) -> None:
        """
        Set a VF to spoofchk on/off.

        Turning off spoofchk allows the VF to change its MAC address.
        :param vf_id: Virtual Function ID
        :param state: State to be set: Enabled or Disabled (On/Off)
        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException in case of error
        """
        self._raise_error_if_not_supported_type()
        value = "on" if state == State.ENABLED else "off"
        cmd = f"ip link set {self._interface().name} vf {vf_id} spoofchk {value}"
        logger.log(level=log_levels.MODULE_DEBUG, msg=f"Set spoofchk on VF ID: {vf_id} to '{value}'")
        self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)

    def get_trust(self, vf_id: int) -> State:
        """
        Get trust setting per VF.

        :param vf_id: Virtual Function ID
        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :return: State of trust setting
        """
        for vf in self._get_vfs_details():
            if vf_id == vf.id:
                return vf.trust

    def get_spoofchk(self, vf_id: int) -> State:
        """
        Get spoofhck setting per VF.

        :param vf_id: Virtual Function ID
        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :return: State of spoofhck setting
        """
        for vf in self._get_vfs_details():
            if vf_id == vf.id:
                return vf.spoofchk

    def get_mac_address(self, vf_id: int) -> MACAddress:
        """
        Get mac_address per VF.

        :param vf_id: Virtual Function ID
        :raises VirtualizationWrongInterfaceException: if method is called on non PF/BTS/VPORT interface
        :return: Mac Address of VF
        """
        for vf in self._get_vfs_details():
            if vf_id == vf.id:
                return vf.mac_address

    def get_link_state(self, vf_id: int) -> LinkState:
        """
        Get link-state per VF.

        :param vf_id: Virtual Function ID
        :raises VirtualizationWrongInterfaceException: if method is called on non PF/BTS/VPORT interface
        :return: Link State setting of VF
        """
        for vf in self._get_vfs_details():
            if vf_id == vf.id:
                return vf.link_state

    def set_vlan_for_vf(self, vf_id: int, vlan_id: int, proto: VlanProto = None) -> None:
        """
        Set port VLAN for a VF interface.

        :param vf_id: Virtual Function ID
        :param vlan_id: VLAN to set the VF to
        :param proto: Specify 802.1ad or 802.1Q type of VLAN
        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException in case of error
        """
        self._raise_error_if_not_supported_type()
        proto_suffix = f" proto {proto.value}" if proto else ""
        cmd = f"ip link set {self._interface().name} vf {vf_id} vlan {vlan_id}{proto_suffix}"
        self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)

    def set_link_for_vf(self, vf_id: int, link_state: LinkState) -> None:
        """
        Set link for a VF interface.

        :param vf_id: Virtual Function ID
        :param link_state: requested link state (one of auto, enabled, disabled)
        :raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
        :raises: VirtualizationFeatureException in case of error
        """
        self._raise_error_if_not_supported_type()
        cmd = f"ip link set {self._interface().name} vf {vf_id} state {link_state.value}"
        self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)

    def set_mac_for_vf(self, vf_id: int, mac: MACAddress) -> None:
        """
        Set MAC address for VF interface.

        :param vf_id: Virtual Function ID
        :param mac: MAC to set on VF interface
        """
        logger.warning(
            "This API is deprecated - `interface.virtualization.set_mac_for_vf()`. "
            "Use `owner.mac.set_mac_for_vf() instead."
        )
        self._raise_error_if_not_supported_type()
        cmd = f"ip link set {self._interface().name} vf {vf_id} mac {mac}"
        self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)

    def get_vf_id_by_pci(self, vf_pci_address: PCIAddress) -> int:
        """
        Get ID of VF with the given PCI address on specific PF PCI address using /sys/bus/pci/devices/pci_address.

        :param vf_pci_address: VF interface PCI address.
        :return: ID of the VF.
        """
        result = self._connection.execute_command(
            f"ls /sys/bus/pci/devices/{self._interface().pci_address}/virtfn* -la",
            shell=True,
            expected_return_codes={0, 1},
        )
        if result.return_code != 0:
            raise VirtualFunctionNotFoundException(
                f"Failed to list VFs for PF PCI Address {self._interface().pci_address}: {result.stderr.strip()}"
            )
        vf_number_regex = rf"^.*devices/{self._interface().pci_address}/virtfn(?P<vf_number>\d+).*->.*{vf_pci_address}$"
        match = re.search(vf_number_regex, result.stdout, re.M)
        if match:
            return int(match.group("vf_number"))
        else:
            raise VirtualFunctionNotFoundException(f"0 matched VFs for PF PCI Address {self._interface().pci_address}")

    def get_msix_vectors_count(self, method: MethodType = MethodType.DEVLINK) -> int:
        """
        Get number of MSI-X vectors for the given interface.

        :param method: Method to use for setting MSI-X vectors count. Options are "devlink" or "sysfs".
        :return: Number of MSI-X vectors available for the interface.
        """
        interface = self._interface()
        if interface.interface_type is not InterfaceType.PF:
            raise NetworkInterfaceNotSupported("Getting MSI-X vector count is only supported on PF interface.")
        logger.log(level=log_levels.MFD_DEBUG, msg=f"Getting MSI-X vectors count for interface {interface.name}")
        if method == MethodType.DEVLINK:
            out = self._connection.execute_command(f"devlink resource show pci/{interface.pci_address}").stdout
            match = re.search(r"name msix_vf size (\d+) ", out)
            if match:
                logger.log(
                    level=log_levels.MFD_INFO,
                    msg=f"MSI-X vectors count for interface {interface.name}: {match.group(1)}",
                )
                return int(match.group(1))
            else:
                raise NetworkAdapterConfigurationException(
                    f"Could not find MSI-X vectors count for interface {interface.name}"
                )

        if method == MethodType.SYSFS:
            out = self._connection.execute_command(
                f"cat /sys/bus/pci/devices/{self._interface().pci_address}/sriov_vf_msix_count"
            ).stdout
            if out:
                logger.log(level=log_levels.MFD_INFO, msg=f"MSI-X vectors count for interface {interface.name}: {out}")
                return int(out)
            else:
                raise NetworkAdapterConfigurationException(
                    f"Could not find MSI-X vectors count for interface {interface.name}"
                )

        raise ValueError(f"Unknown method {method} for getting MSI-X vectors count")

    def set_msix_vectors_count(self, count: int, method: MethodType = MethodType.DEVLINK) -> None:
        """
        Set number of MSI-X vectors for the given interface.

        :param count: Number of MSI-X vectors to set
        :param method: Method to use for setting MSI-X vectors count. Options are "devlink" or "sysfs".
        """
        interface = self._interface()
        if interface.interface_type is not InterfaceType.PF:
            raise NetworkInterfaceNotSupported(
                "Setting MSI-X vector count on VF is only supported through PF interface."
            )
        logger.log(
            level=log_levels.MFD_DEBUG, msg=f"Setting MSI-X vectors count to {count} for interface {interface.name}"
        )
        if method == MethodType.DEVLINK:
            command = f"devlink resource set pci/{interface.pci_address} path /msix/msix_vf/ size {count}"
        elif method == MethodType.SYSFS:
            command = f"echo {count} > /sys/bus/pci/devices/{interface.pci_address}/sriov_vf_msix_count"
        else:
            raise ValueError(f"Unknown method {method} for setting MSI-X vectors count")
        self._connection.execute_command(command, custom_exception=NetworkAdapterConfigurationException)
        logger.log(level=log_levels.MFD_INFO, msg=f"MSI-X vectors count set to {count} for interface {interface.name}")

1	# Copyright (C) 2025 Intel Corporation
2	# SPDX-License-Identifier: MIT
3	"""Module for Virtualization feature for Linux."""	4✔
4
5	import logging	4✔
6	import re	4✔
7	from typing import List	4✔
8
9	from mfd_common_libs import add_logging_level, log_levels	4✔
10	from mfd_const.network import DESIGNED_NUMBER_VFS_BY_SPEED, Speed	4✔
11	from mfd_typing import MACAddress, DeviceID, SubDeviceID, PCIAddress	4✔
12	from mfd_typing.network_interface import InterfaceType	4✔
13
14	from mfd_network_adapter.data_structures import State	4✔
15	from mfd_network_adapter.exceptions import (	4✔
16	VirtualFunctionNotFoundException,
17	NetworkAdapterConfigurationException,
18	NetworkInterfaceNotSupported,
19	)
20	from .base import BaseFeatureVirtualization	4✔
21	from .data_structures import MethodType	4✔
22	from ...data_structures import VlanProto, VFDetail, LinkState	4✔
23	from ...exceptions import VirtualizationFeatureException, VirtualizationWrongInterfaceException, DeviceSetupException	4✔
24
25	logger = logging.getLogger(__name__)	4✔
26	add_logging_level(level_name="MODULE_DEBUG", level_value=log_levels.MODULE_DEBUG)	4✔
27
28
29	class LinuxVirtualization(BaseFeatureVirtualization):	4✔
30	"""Linux class for Virtualization feature."""
31
32	def _raise_error_if_not_supported_type(self) -> None:	4✔
33	"""
34	Raise error in case current interface is not PF/BTS/VPORT.
35
36	:raises VirtualizationWrongInterfaceException: if method is called on non PF/BTS/VPORT interface
37	"""
38	current_type = self._interface().interface_type	4✔
39	if current_type not in (InterfaceType.PF, InterfaceType.BTS, InterfaceType.VPORT):	4✔
40	raise VirtualizationWrongInterfaceException(	4✔
41	f"Current interface type is {current_type} but should be InterfaceType.PF or InterfaceType.BTS"
42	)
43
44	def _get_vfs_details(self) -> List[VFDetail]:	4✔
45	"""
46	Get VF details of PF interface.
47
48	:raises VirtualizationWrongInterfaceException: if method is called on non PF/BTS/VPORT interface
49	:raises: VirtualizationFeatureException: in case of command failure (rc != 0)
50	:return: List of VFDetail objects
51	"""
52	self._raise_error_if_not_supported_type()	4✔
53
54	command = f"ip link show dev {self._interface().name}"	4✔
55	pattern = (	4✔
56	r"vf\s(?P<vf_id>\d+)\s(link/ether\|MAC)\s(?P<mac_address>[0-9a-fA-F]{2}(?::[0-9a-fA-F]{2}){5})\s.*?,"
57	r"\sspoof checking\s(?P<spoofchk>\w+),\slink-state\s(?P<link_state>\w+),\strust\s("
58	r"?P<trust>\w+)"
59	)
60	link_state_map = {"enable": LinkState.ENABLE, "disable": LinkState.DISABLE, "auto": LinkState.AUTO}	4✔
61
62	output = self._connection.execute_command(	4✔
63	command=command, custom_exception=VirtualizationFeatureException
64	).stdout
65	vf_details = []	4✔
66	info_match = re.finditer(pattern, output)	4✔
67	for match in info_match:	4✔
68	vf_details.append(	4✔
69	VFDetail(
70	id=int(match.group("vf_id")),
71	mac_address=MACAddress(match.group("mac_address")),
72	spoofchk=State.ENABLED if match.group("spoofchk") == "on" else State.DISABLED,
73	link_state=link_state_map.get(match.group("link_state")),
74	trust=State.ENABLED if match.group("trust") == "on" else State.DISABLED,
75	)
76	)
77	return vf_details	4✔
78
79	def _get_max_vfs_by_name(self) -> int:	4✔
80	"""
81	Get maximal number of VFs per interface based on name.
82
83	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
84	:raises: VirtualizationFeatureException in case of error
85	:return: number of VFs
86	"""
87	self._raise_error_if_not_supported_type()	4✔
88	command = f"cat /sys/class/net/{self._interface().name}/device/sriov_totalvfs"	4✔
89	result = self._connection.execute_command(command=command, custom_exception=VirtualizationFeatureException)	4✔
90
91	return int(result.stdout)	4✔
92
93	def _get_max_vfs_by_pci_address(self) -> int:	4✔
94	"""
95	Get maximal number of VFs per interface based on PCI Address.
96
97	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
98	:raises: VirtualizationFeatureException in case of error
99	:return: number of VFs
100	"""
101	self._raise_error_if_not_supported_type()	4✔
102	command = f"cat /sys/bus/pci/devices/{self._interface().pci_address}/sriov_totalvfs"	4✔
103	result = self._connection.execute_command(command=command, custom_exception=VirtualizationFeatureException)	4✔
104
105	return int(result.stdout)	4✔
106
107	def _get_current_vfs_by_name(self) -> int:	4✔
108	"""
109	Get number of current VFs per interface based on name.
110
111	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
112	:raises: VirtualizationFeatureException in case of error
113	:return: number of VFs
114	"""
115	self._raise_error_if_not_supported_type()	4✔
116	command = f"cat /sys/class/net/{self._interface().name}/device/sriov_numvfs"	4✔
117	result = self._connection.execute_command(command=command, custom_exception=VirtualizationFeatureException)	4✔
118
119	return int(result.stdout)	4✔
120
121	def _get_current_vfs_by_pci_address(self) -> int:	4✔
122	"""
123	Get number of current VFs per interface based on PCI Address.
124
125	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
126	:raises: VirtualizationFeatureException in case of error
127	:return: number of VFs
128	"""
129	self._raise_error_if_not_supported_type()	4✔
130	command = f"cat /sys/bus/pci/devices/{self._interface().pci_address}/sriov_numvfs"	4✔
131	result = self._connection.execute_command(command=command, custom_exception=VirtualizationFeatureException)	4✔
132
133	return int(result.stdout)	4✔
134
135	def get_max_vfs(self) -> int:	4✔
136	"""Get maximal number of VFs per interface."""
137	return self._get_max_vfs_by_name() if self._interface().name else self._get_max_vfs_by_pci_address()	4✔
138
139	def get_current_vfs(self) -> int:	4✔
140	"""Get number of current VFs per interface."""
141	return self._get_current_vfs_by_name() if self._interface().name else self._get_current_vfs_by_pci_address()	4✔
142
143	def get_designed_number_vfs(self) -> tuple[int, int]:	4✔
144	"""
145	Return designed max number of VFs, total and per PF.
146
147	:return: max VFs for NIC, max VFS per PF
148	"""
UNCOV 149	speed = self._interface().speed	×
150	if speed in (Speed.G1, Speed.G10, Speed.G40, Speed.G200):	×
151	return DESIGNED_NUMBER_VFS_BY_SPEED[speed], int(	×
152	DESIGNED_NUMBER_VFS_BY_SPEED[speed] / self._interface().get_number_of_ports()
153	)
UNCOV 154	elif speed is Speed.G100:	×
155	device_id = self._interface().pci_device.device_id	×
156	sub_device_id = self._interface().pci_device.sub_device_id	×
157	if device_id == DeviceID(0x1592) and sub_device_id == SubDeviceID(0x000E):	×
158	designed_max_vfs = 512	×
159	designed_max_vfs_per_pf = 256	×
160	logger.log(	×
161	level=log_levels.MODULE_DEBUG,
162	msg=(
163	"Recognized Chapman Beach (2 chips on NIC)."
164	f"Return designed max VFS: {designed_max_vfs} and per PF: {designed_max_vfs_per_pf}"
165	),
166	)
UNCOV 167	return designed_max_vfs, designed_max_vfs_per_pf	×
168	else:
UNCOV 169	return DESIGNED_NUMBER_VFS_BY_SPEED[Speed.G100], int(	×
170	DESIGNED_NUMBER_VFS_BY_SPEED[Speed.G100] / self._interface().get_number_of_ports()
171	)
172	else:
UNCOV 173	raise DeviceSetupException(	×
174	"Cannot recognize NIC. \n"
175	f"Device string: {self._interface().get_device_string()}.\n"
176	f"Device ID: {self._interface().pci_device.device_id}.\n"
177	f"Sub Device ID: {self._interface().pci_device.sub_device_id}.\n"
178	)
179
180	def set_sriov(self, sriov_enabled: bool, no_restart: bool = False) -> None:	4✔
181	"""
182	Set network interface SRIOV.
183
184	:param sriov_enabled: adapter SRIOV status value to be set.
185	:param no_restart: whether to restart adapter after changing its settings.
186	"""
UNCOV 187	raise NotImplementedError	×
188
189	def set_vmq(self, vmq_enabled: bool, no_restart: bool = False) -> None:	4✔
190	"""
191	Set network interface VMQ.
192
193	:param vmq_enabled: adapter VMQ status value to be set.
194	:param no_restart: whether to restart adapter after changing its settings.
195	"""
UNCOV 196	raise NotImplementedError	×
197
198	def set_max_tx_rate(self, vf_id: int, value: int) -> None:	4✔
199	"""
200	Set max_tx_rate VF-d parameter status.
201
202	:param: vf_id: VF (virtual function) ID
203	:param value: max_tx_rate value (Mbits)
204	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
205	:raises VirtualizationFeatureException if command execution fails
206	"""
207	self._raise_error_if_not_supported_type()	4✔
208
209	cmd = f"ip link set dev {self._interface().name} vf {vf_id} max_tx_rate {value}"	4✔
210	logger.log(level=log_levels.MODULE_DEBUG, msg=f"Set max tx rate using : {cmd}")	4✔
211	self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)	4✔
212
213	def set_min_tx_rate(self, vf_id: int, value: int) -> None:	4✔
214	"""
215	Set min_tx_rate VF-d parameter status.
216
217	:param vf_id: VF (virtual function) ID
218	:param value: min_tx_rate value (Mbits)
219	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
220	:raises VirtualizationFeatureException if command execution fails
221	"""
222	self._raise_error_if_not_supported_type()	4✔
223	cmd = f"ip link set dev {self._interface().name} vf {vf_id} min_tx_rate {value}"	4✔
224	logger.log(level=log_levels.MODULE_DEBUG, msg=f"Set min tx rate using : {cmd}")	4✔
225	self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)	4✔
226
227	def set_trust(self, vf_id: int, state: State) -> None:	4✔
228	"""
229	Change 'trust' setting on VF Interface.
230
231	:param vf_id: Virtual Function ID
232	:param state: State to be set: Enabled or Disabled (On/Off)
233	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
234	:raises: VirtualizationFeatureException in case of command error
235	"""
236	self._raise_error_if_not_supported_type()	4✔
237	value = "on" if state == State.ENABLED else "off"	4✔
238	cmd = f"ip link set {self._interface().name} vf {vf_id} trust {value}"	4✔
239	logger.log(level=log_levels.MODULE_DEBUG, msg=f"Set trust on VF ID: {vf_id} to '{value}'")	4✔
240	self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)	4✔
241
242	def set_spoofchk(self, vf_id: int, state: State) -> None:	4✔
243	"""
244	Set a VF to spoofchk on/off.
245
246	Turning off spoofchk allows the VF to change its MAC address.
247	:param vf_id: Virtual Function ID
248	:param state: State to be set: Enabled or Disabled (On/Off)
249	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
250	:raises: VirtualizationFeatureException in case of error
251	"""
252	self._raise_error_if_not_supported_type()	4✔
253	value = "on" if state == State.ENABLED else "off"	4✔
254	cmd = f"ip link set {self._interface().name} vf {vf_id} spoofchk {value}"	4✔
255	logger.log(level=log_levels.MODULE_DEBUG, msg=f"Set spoofchk on VF ID: {vf_id} to '{value}'")	4✔
256	self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)	4✔
257
258	def get_trust(self, vf_id: int) -> State:	4✔
259	"""
260	Get trust setting per VF.
261
262	:param vf_id: Virtual Function ID
263	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
264	:return: State of trust setting
265	"""
266	for vf in self._get_vfs_details():	4✔
267	if vf_id == vf.id:	4✔
268	return vf.trust	4✔
269
270	def get_spoofchk(self, vf_id: int) -> State:	4✔
271	"""
272	Get spoofhck setting per VF.
273
274	:param vf_id: Virtual Function ID
275	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
276	:return: State of spoofhck setting
277	"""
278	for vf in self._get_vfs_details():	4✔
279	if vf_id == vf.id:	4✔
280	return vf.spoofchk	4✔
281
282	def get_mac_address(self, vf_id: int) -> MACAddress:	4✔
283	"""
284	Get mac_address per VF.
285
286	:param vf_id: Virtual Function ID
287	:raises VirtualizationWrongInterfaceException: if method is called on non PF/BTS/VPORT interface
288	:return: Mac Address of VF
289	"""
290	for vf in self._get_vfs_details():	4✔
291	if vf_id == vf.id:	4✔
292	return vf.mac_address	4✔
293
294	def get_link_state(self, vf_id: int) -> LinkState:	4✔
295	"""
296	Get link-state per VF.
297
298	:param vf_id: Virtual Function ID
299	:raises VirtualizationWrongInterfaceException: if method is called on non PF/BTS/VPORT interface
300	:return: Link State setting of VF
301	"""
302	for vf in self._get_vfs_details():	4✔
303	if vf_id == vf.id:	4✔
304	return vf.link_state	4✔
305
306	def set_vlan_for_vf(self, vf_id: int, vlan_id: int, proto: VlanProto = None) -> None:	4✔
307	"""
308	Set port VLAN for a VF interface.
309
310	:param vf_id: Virtual Function ID
311	:param vlan_id: VLAN to set the VF to
312	:param proto: Specify 802.1ad or 802.1Q type of VLAN
313	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
314	:raises: VirtualizationFeatureException in case of error
315	"""
316	self._raise_error_if_not_supported_type()	4✔
317	proto_suffix = f" proto {proto.value}" if proto else ""	4✔
318	cmd = f"ip link set {self._interface().name} vf {vf_id} vlan {vlan_id}{proto_suffix}"	4✔
319	self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)	4✔
320
321	def set_link_for_vf(self, vf_id: int, link_state: LinkState) -> None:	4✔
322	"""
323	Set link for a VF interface.
324
325	:param vf_id: Virtual Function ID
326	:param link_state: requested link state (one of auto, enabled, disabled)
327	:raises VirtualizationWrongInterfaceException if method is called on non PF/BTS/VPORT interface
328	:raises: VirtualizationFeatureException in case of error
329	"""
330	self._raise_error_if_not_supported_type()	4✔
331	cmd = f"ip link set {self._interface().name} vf {vf_id} state {link_state.value}"	4✔
332	self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)	4✔
333
334	def set_mac_for_vf(self, vf_id: int, mac: MACAddress) -> None:	4✔
335	"""
336	Set MAC address for VF interface.
337
338	:param vf_id: Virtual Function ID
339	:param mac: MAC to set on VF interface
340	"""
341	logger.warning(	4✔
342	"This API is deprecated - `interface.virtualization.set_mac_for_vf()`. "
343	"Use `owner.mac.set_mac_for_vf() instead."
344	)
345	self._raise_error_if_not_supported_type()	4✔
346	cmd = f"ip link set {self._interface().name} vf {vf_id} mac {mac}"	4✔
347	self._connection.execute_command(command=cmd, custom_exception=VirtualizationFeatureException)	4✔
348
349	def get_vf_id_by_pci(self, vf_pci_address: PCIAddress) -> int:	4✔
350	"""
351	Get ID of VF with the given PCI address on specific PF PCI address using /sys/bus/pci/devices/pci_address.
352
353	:param vf_pci_address: VF interface PCI address.
354	:return: ID of the VF.
355	"""
356	result = self._connection.execute_command(	4✔
357	f"ls /sys/bus/pci/devices/{self._interface().pci_address}/virtfn* -la",
358	shell=True,
359	expected_return_codes={0, 1},
360	)
361	if result.return_code != 0:	4✔
362	raise VirtualFunctionNotFoundException(	4✔
363	f"Failed to list VFs for PF PCI Address {self._interface().pci_address}: {result.stderr.strip()}"
364	)
365	vf_number_regex = rf"^.devices/{self._interface().pci_address}/virtfn(?P<vf_number>\d+).->.*{vf_pci_address}$"	4✔
366	match = re.search(vf_number_regex, result.stdout, re.M)	4✔
367	if match:	4✔
368	return int(match.group("vf_number"))	4✔
369	else:
370	raise VirtualFunctionNotFoundException(f"0 matched VFs for PF PCI Address {self._interface().pci_address}")	4✔
371
372	def get_msix_vectors_count(self, method: MethodType = MethodType.DEVLINK) -> int:	4✔
373	"""
374	Get number of MSI-X vectors for the given interface.
375
376	:param method: Method to use for setting MSI-X vectors count. Options are "devlink" or "sysfs".
377	:return: Number of MSI-X vectors available for the interface.
378	"""
379	interface = self._interface()	4✔
380	if interface.interface_type is not InterfaceType.PF:	4✔
381	raise NetworkInterfaceNotSupported("Getting MSI-X vector count is only supported on PF interface.")	4✔
382	logger.log(level=log_levels.MFD_DEBUG, msg=f"Getting MSI-X vectors count for interface {interface.name}")	4✔
383	if method == MethodType.DEVLINK:	4✔
384	out = self._connection.execute_command(f"devlink resource show pci/{interface.pci_address}").stdout	4✔
385	match = re.search(r"name msix_vf size (\d+) ", out)	4✔
386	if match:	4✔
387	logger.log(	4✔
388	level=log_levels.MFD_INFO,
389	msg=f"MSI-X vectors count for interface {interface.name}: {match.group(1)}",
390	)
391	return int(match.group(1))	4✔
392	else:
393	raise NetworkAdapterConfigurationException(	4✔
394	f"Could not find MSI-X vectors count for interface {interface.name}"
395	)
396
397	if method == MethodType.SYSFS:	4✔
398	out = self._connection.execute_command(	4✔
399	f"cat /sys/bus/pci/devices/{self._interface().pci_address}/sriov_vf_msix_count"
400	).stdout
401	if out:	4✔
402	logger.log(level=log_levels.MFD_INFO, msg=f"MSI-X vectors count for interface {interface.name}: {out}")	4✔
403	return int(out)	4✔
404	else:
405	raise NetworkAdapterConfigurationException(	4✔
406	f"Could not find MSI-X vectors count for interface {interface.name}"
407	)
408
409	raise ValueError(f"Unknown method {method} for getting MSI-X vectors count")	4✔
410
411	def set_msix_vectors_count(self, count: int, method: MethodType = MethodType.DEVLINK) -> None:	4✔
412	"""
413	Set number of MSI-X vectors for the given interface.
414
415	:param count: Number of MSI-X vectors to set
416	:param method: Method to use for setting MSI-X vectors count. Options are "devlink" or "sysfs".
417	"""
418	interface = self._interface()	4✔
419	if interface.interface_type is not InterfaceType.PF:	4✔
420	raise NetworkInterfaceNotSupported(	4✔
421	"Setting MSI-X vector count on VF is only supported through PF interface."
422	)
423	logger.log(	4✔
424	level=log_levels.MFD_DEBUG, msg=f"Setting MSI-X vectors count to {count} for interface {interface.name}"
425	)
426	if method == MethodType.DEVLINK:	4✔
427	command = f"devlink resource set pci/{interface.pci_address} path /msix/msix_vf/ size {count}"	4✔
428	elif method == MethodType.SYSFS:	4✔
429	command = f"echo {count} > /sys/bus/pci/devices/{interface.pci_address}/sriov_vf_msix_count"	4✔
430	else:
431	raise ValueError(f"Unknown method {method} for setting MSI-X vectors count")	4✔
432	self._connection.execute_command(command, custom_exception=NetworkAdapterConfigurationException)	4✔
433	logger.log(level=log_levels.MFD_INFO, msg=f"MSI-X vectors count set to {count} for interface {interface.name}")	4✔

intel / mfd-network-adapter / 26037842892

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous