diff --git a/mfd_switchmanagement/vendors/dell/dell_os10/base.py b/mfd_switchmanagement/vendors/dell/dell_os10/base.py index 1695366..5e098a2 100644 --- a/mfd_switchmanagement/vendors/dell/dell_os10/base.py +++ b/mfd_switchmanagement/vendors/dell/dell_os10/base.py @@ -981,3 +981,112 @@ def configure_lldp(self, port: str, param: str) -> str: f"lldp {param}", ] ) + + def create_qos_policy_userspace(self, suffix: str = "100") -> None: + """ + Create QOS policy for userspace traffic. + + Set all traffic to traffic class 0 and qos group 3. + :param suffix: suffix to use in names + """ + trust_commands = [f"trust dot1p-map TM_{suffix}"] + priority = 8 + cos_value = ",".join([str(qos_prio) for qos_prio in range(priority)]) + + trust_commands.append(f"qos-group 3 dot1p {cos_value}") + trust_commands.append("exit") + + self._connection.send_configuration(trust_commands) + self.create_qos_map(queues=[0], tc_name=f"QMU_{suffix}", queue_type="ucast") + self.create_qos_map(queues=[0], tc_name=f"QMU_{suffix}", queue_type="mcast") + + def set_port_bw_by_tc_userspace(self, port: str, suffix: str = "100") -> None: + """ + Set the bandwidth of traffic class on a selected port. + + Same as set_port_bw_by_tc but with changed suffix. + + :param port: switch port + :param suffix: suffix for names + """ + self._validate_configure_parameters(ports=port) + prange = self._port_range(port) + + configuration = [ + f"interface {prange}{self._convert_port_name(port)}", + f"trust-map dot1p TM_{suffix}", + f"qos-map traffic-class QMU_{suffix}", + f"service-policy output type queuing PMU_{suffix}", + f"service-policy input type network-qos PMQU_{suffix}", + ] + self._connection.send_configuration(configuration) + + def configure_pfc_ndk(self, port: str) -> None: + """ + Set default ndk Priority Flow Control (PFC) settings on the interface. + + :param port: port of switch + """ + self.create_qos_policy([90, 10, 0, 0, 0, 0, 0, 0], [0, 0, 0, 1, 0, 0, 0, 0], "100") + self.set_port_bw_by_tc(port, suffix="100") + self.set_port_pfc_by_tc(port, qos_priority=None, pfc="on") + self.set_port_dcbx_version(port, mode="ieee") + + def configure_pfc_userspace(self, port: str) -> None: + """ + Set default userspace Priority Flow Control (PFC) settings on the interface. + + :param port: port of switch + """ + self.create_qos_policy_userspace(suffix="100") + self.create_qos_class_map(name="Q0U_100", priority="0", class_type="queuing") + self.create_qos_queuing_policy_map(name="PMU_100", class_bandwidth_dict={"Q0U_100": 100}) + self.create_qos_class_map(name="CMQU_100", priority="3", class_type="network-qos") + self.create_network_qos_policy_map(name="PMQU_100", class_names=["CMQU_100"], cos_values=["3"]) + self.set_port_bw_by_tc_userspace(port, suffix="100") + self.set_port_pfc_by_tc(port, qos_priority=None, pfc="on") + self.set_port_dcbx_version(port, mode="ieee") + + def delete_qos_policy_userspace(self, suffix: str = "100") -> None: + """ + Delete the userspace QoS policy. + + :param suffix: suffix used in names + """ + configuration = [ + f"no policy-map type queuing PMU_{suffix}", + f"no class-map type queuing Q0U_{suffix}", + f"no qos-map traffic-class QMU_{suffix}", + f"no trust dot1p-map TM_{suffix}", + f"no policy-map type network-qos PMQU_{suffix}", + f"no class-map type network-qos CMQU_{suffix}", + ] + self._connection.send_configuration(configuration) + + def disable_pfc(self, port: str, suffix: str = "100", pfc_type: str = "ndk") -> None: + """ + Disable Priority Flow Control (PFC) settings on the interface. + + :param port: port of switch + :param suffix: suffix used in QoS policy names (default: "100") + :param pfc_type: type of PFC configuration - "ndk" or "userspace" (default: "ndk") + """ + logger.log(level=log_levels.MODULE_DEBUG, msg=f"Disabling {pfc_type} PFC on port: {port}") + self._validate_port_and_port_channel_syntax(ethernet_port=port) + + # Remove port-specific configurations + self.delete_port_bw_by_tc(port, suffix=suffix) + self.delete_port_pfc(port) + self.clear_port_dcbx(port) + + # Delete global QoS policies based on type + if pfc_type == "userspace": + self.delete_qos_policy_userspace(suffix=suffix) + else: + self.delete_qos_policy(suffix=suffix) + + # Reset interface to defaults + commands = [ + f"default interface {port}", + ] + self._connection.send_configuration(commands) diff --git a/tests/unit/test_mfd_switchmanagement/test_vendors/test_dell/test_dell_os10/test_base.py b/tests/unit/test_mfd_switchmanagement/test_vendors/test_dell/test_dell_os10/test_base.py index d73f2d0..6e6dca5 100644 --- a/tests/unit/test_mfd_switchmanagement/test_vendors/test_dell/test_dell_os10/test_base.py +++ b/tests/unit/test_mfd_switchmanagement/test_vendors/test_dell/test_dell_os10/test_base.py @@ -1120,3 +1120,253 @@ def test_remove_port_convert_port_name_called(self, switch, mocker): # Assert convert_port_name.assert_called_once_with(port) switch._connection.send_command_list.assert_called_once() + + def test_create_qos_policy_userspace_default_suffix(self, switch, mocker): + """Test create_qos_policy_userspace with default suffix.""" + # Arrange + switch._connection = mocker.Mock() + switch.create_qos_map = mocker.Mock() + + # Act + switch.create_qos_policy_userspace() + + # Assert + expected_commands = ["trust dot1p-map TM_100", "qos-group 3 dot1p 0,1,2,3,4,5,6,7", "exit"] + switch._connection.send_configuration.assert_called_once_with(expected_commands) + assert switch.create_qos_map.call_count == 2 + switch.create_qos_map.assert_any_call(queues=[0], tc_name="QMU_100", queue_type="ucast") + switch.create_qos_map.assert_any_call(queues=[0], tc_name="QMU_100", queue_type="mcast") + + def test_create_qos_policy_userspace_custom_suffix(self, switch, mocker): + """Test create_qos_policy_userspace with custom suffix.""" + # Arrange + switch._connection = mocker.Mock() + switch.create_qos_map = mocker.Mock() + suffix = "200" + + # Act + switch.create_qos_policy_userspace(suffix=suffix) + + # Assert + expected_commands = ["trust dot1p-map TM_200", "qos-group 3 dot1p 0,1,2,3,4,5,6,7", "exit"] + switch._connection.send_configuration.assert_called_once_with(expected_commands) + switch.create_qos_map.assert_any_call(queues=[0], tc_name="QMU_200", queue_type="ucast") + switch.create_qos_map.assert_any_call(queues=[0], tc_name="QMU_200", queue_type="mcast") + + def test_set_port_bw_by_tc_userspace_default_suffix(self, switch, mocker): + """Test set_port_bw_by_tc_userspace with default suffix.""" + # Arrange + port = "eth1/1/1" + switch._connection = mocker.Mock() + switch._validate_configure_parameters = mocker.Mock() + switch._port_range = mocker.Mock(return_value="") + switch._convert_port_name = mocker.Mock(return_value="ethernet1/1/1") + + # Act + switch.set_port_bw_by_tc_userspace(port) + + # Assert + switch._validate_configure_parameters.assert_called_once_with(ports=port) + expected_commands = [ + "interface ethernet1/1/1", + "trust-map dot1p TM_100", + "qos-map traffic-class QMU_100", + "service-policy output type queuing PMU_100", + "service-policy input type network-qos PMQU_100", + ] + switch._connection.send_configuration.assert_called_once_with(expected_commands) + + def test_set_port_bw_by_tc_userspace_custom_suffix(self, switch, mocker): + """Test set_port_bw_by_tc_userspace with custom suffix.""" + # Arrange + port = "eth1/1/2" + suffix = "250" + switch._connection = mocker.Mock() + switch._validate_configure_parameters = mocker.Mock() + switch._port_range = mocker.Mock(return_value="range ") + switch._convert_port_name = mocker.Mock(return_value="ethernet1/1/2-1/1/3") + + # Act + switch.set_port_bw_by_tc_userspace(port, suffix=suffix) + + # Assert + switch._validate_configure_parameters.assert_called_once_with(ports=port) + expected_commands = [ + "interface range ethernet1/1/2-1/1/3", + "trust-map dot1p TM_250", + "qos-map traffic-class QMU_250", + "service-policy output type queuing PMU_250", + "service-policy input type network-qos PMQU_250", + ] + switch._connection.send_configuration.assert_called_once_with(expected_commands) + + def test_set_port_bw_by_tc_userspace_invalid_port(self, switch, mocker): + """Test set_port_bw_by_tc_userspace with invalid port.""" + # Arrange + port = "invalid_port" + switch._validate_configure_parameters = mocker.Mock(side_effect=ValueError("Invalid port")) + switch._connection = mocker.Mock() + + # Act & Assert + with raises(ValueError, match="Invalid port"): + switch.set_port_bw_by_tc_userspace(port) + switch._connection.send_configuration.assert_not_called() + + def test_configure_pfc_ndk(self, switch, mocker): + """Test configure_pfc_ndk configuration.""" + # Arrange + port = "eth1/1/1" + switch.create_qos_policy = mocker.Mock() + switch.set_port_bw_by_tc = mocker.Mock() + switch.set_port_pfc_by_tc = mocker.Mock() + switch.set_port_dcbx_version = mocker.Mock() + + # Act + switch.configure_pfc_ndk(port) + + # Assert + switch.create_qos_policy.assert_called_once_with([90, 10, 0, 0, 0, 0, 0, 0], [0, 0, 0, 1, 0, 0, 0, 0], "100") + switch.set_port_bw_by_tc.assert_called_once_with(port, suffix="100") + switch.set_port_pfc_by_tc.assert_called_once_with(port, qos_priority=None, pfc="on") + switch.set_port_dcbx_version.assert_called_once_with(port, mode="ieee") + + def test_configure_pfc_userspace(self, switch, mocker): + """Test configure_pfc_userspace configuration.""" + # Arrange + port = "eth1/1/2" + switch.create_qos_policy_userspace = mocker.Mock() + switch.create_qos_class_map = mocker.Mock() + switch.create_qos_queuing_policy_map = mocker.Mock() + switch.create_network_qos_policy_map = mocker.Mock() + switch.set_port_bw_by_tc_userspace = mocker.Mock() + switch.set_port_pfc_by_tc = mocker.Mock() + switch.set_port_dcbx_version = mocker.Mock() + + # Act + switch.configure_pfc_userspace(port) + + # Assert + switch.create_qos_policy_userspace.assert_called_once_with(suffix="100") + switch.create_qos_class_map.assert_any_call(name="Q0U_100", priority="0", class_type="queuing") + switch.create_qos_queuing_policy_map.assert_called_once_with( + name="PMU_100", class_bandwidth_dict={"Q0U_100": 100} + ) + switch.create_qos_class_map.assert_any_call(name="CMQU_100", priority="3", class_type="network-qos") + switch.create_network_qos_policy_map.assert_called_once_with( + name="PMQU_100", class_names=["CMQU_100"], cos_values=["3"] + ) + switch.set_port_bw_by_tc_userspace.assert_called_once_with(port, suffix="100") + switch.set_port_pfc_by_tc.assert_called_once_with(port, qos_priority=None, pfc="on") + switch.set_port_dcbx_version.assert_called_once_with(port, mode="ieee") + + def test_delete_qos_policy_userspace_default_suffix(self, switch, mocker): + """Test delete_qos_policy_userspace with default suffix.""" + # Arrange + switch._connection = mocker.Mock() + + # Act + switch.delete_qos_policy_userspace() + + # Assert + expected_commands = [ + "no policy-map type queuing PMU_100", + "no class-map type queuing Q0U_100", + "no qos-map traffic-class QMU_100", + "no trust dot1p-map TM_100", + "no policy-map type network-qos PMQU_100", + "no class-map type network-qos CMQU_100", + ] + switch._connection.send_configuration.assert_called_once_with(expected_commands) + + def test_delete_qos_policy_userspace_custom_suffix(self, switch, mocker): + """Test delete_qos_policy_userspace with custom suffix.""" + # Arrange + switch._connection = mocker.Mock() + suffix = "300" + + # Act + switch.delete_qos_policy_userspace(suffix=suffix) + + # Assert + expected_commands = [ + "no policy-map type queuing PMU_300", + "no class-map type queuing Q0U_300", + "no qos-map traffic-class QMU_300", + "no trust dot1p-map TM_300", + "no policy-map type network-qos PMQU_300", + "no class-map type network-qos CMQU_300", + ] + switch._connection.send_configuration.assert_called_once_with(expected_commands) + + def test_disable_pfc_ndk_type(self, switch, mocker): + """Test disable_pfc with NDK type (default).""" + # Arrange + port = "eth1/1/1" + switch._validate_port_and_port_channel_syntax = mocker.Mock() + switch.delete_port_bw_by_tc = mocker.Mock() + switch.delete_port_pfc = mocker.Mock() + switch.clear_port_dcbx = mocker.Mock() + switch.delete_qos_policy = mocker.Mock() + switch._connection = mocker.Mock() + + # Act + switch.disable_pfc(port) + + # Assert + switch._validate_port_and_port_channel_syntax.assert_called_once_with(ethernet_port=port) + switch.delete_port_bw_by_tc.assert_called_once_with(port, suffix="100") + switch.delete_port_pfc.assert_called_once_with(port) + switch.clear_port_dcbx.assert_called_once_with(port) + switch.delete_qos_policy.assert_called_once_with(suffix="100") + switch._connection.send_configuration.assert_called_once_with([f"default interface {port}"]) + + def test_disable_pfc_userspace_type(self, switch, mocker): + """Test disable_pfc with userspace type.""" + # Arrange + port = "eth1/1/2" + switch._validate_port_and_port_channel_syntax = mocker.Mock() + switch.delete_port_bw_by_tc = mocker.Mock() + switch.delete_port_pfc = mocker.Mock() + switch.clear_port_dcbx = mocker.Mock() + switch.delete_qos_policy_userspace = mocker.Mock() + switch._connection = mocker.Mock() + + # Act + switch.disable_pfc(port, pfc_type="userspace") + + # Assert + switch._validate_port_and_port_channel_syntax.assert_called_once_with(ethernet_port=port) + switch.delete_port_bw_by_tc.assert_called_once_with(port, suffix="100") + switch.delete_port_pfc.assert_called_once_with(port) + switch.clear_port_dcbx.assert_called_once_with(port) + switch.delete_qos_policy_userspace.assert_called_once_with(suffix="100") + switch._connection.send_configuration.assert_called_once_with([f"default interface {port}"]) + + def test_disable_pfc_custom_suffix(self, switch, mocker): + """Test disable_pfc with custom suffix.""" + # Arrange + port = "eth1/1/3" + suffix = "200" + switch._validate_port_and_port_channel_syntax = mocker.Mock() + switch.delete_port_bw_by_tc = mocker.Mock() + switch.delete_port_pfc = mocker.Mock() + switch.clear_port_dcbx = mocker.Mock() + switch.delete_qos_policy = mocker.Mock() + switch._connection = mocker.Mock() + + # Act + switch.disable_pfc(port, suffix=suffix) + + # Assert + switch.delete_port_bw_by_tc.assert_called_once_with(port, suffix="200") + switch.delete_qos_policy.assert_called_once_with(suffix="200") + + def test_disable_pfc_invalid_port(self, switch, mocker): + """Test disable_pfc with invalid port.""" + # Arrange + port = "invalid_port" + switch._validate_port_and_port_channel_syntax = mocker.Mock(side_effect=ValueError("Invalid port format")) + + # Act & Assert + with raises(ValueError, match="Invalid port format"): + switch.disable_pfc(port)