diff --git a/examples/linux_example.py b/examples/linux_example.py
index fbb6158..2b6f975 100644
--- a/examples/linux_example.py
+++ b/examples/linux_example.py
@@ -10,7 +10,14 @@
 
 user_priority_mapping = [(0, 1), (3, 2), (5, 4), (6, 7)]
 bandwidth_per_traffic_class = [25, 25, 25, 25]
-print(dcb_obj.set_ets(user_priority_mapping, bandwidth_per_traffic_class, interface_name="eth4", mode="ieee"))
+print(
+    dcb_obj.set_ets(
+        user_priority_mapping,
+        bandwidth_per_traffic_class,
+        interface_name="eth4",
+        mode="ieee",
+    )
+)
 pfc_per_priority = [False, True, True, False, False, False, False, False]
 print(dcb_obj.set_pfc(interface_name="eth4", pfc_per_priority=pfc_per_priority))
 print(dcb_obj.restart_lldpad())
@@ -31,3 +38,27 @@
 print(dcb_obj.get_ets(interface_name="eth4", tool_name="dcbtool"))
 print(dcb_obj.get_app(interface_name="eth4", tool_name="dcbtool"))
 print(dcb_obj.get_pfc(interface_name="eth4", tool_name="dcbtool"))
+
+# iproute2 "dcb" tool (dcbnl netlink) - set configuration
+print(
+    dcb_obj.set_ets(
+        user_priority_mapping,
+        bandwidth_per_traffic_class,
+        interface_name="eth4",
+        tool_name="dcb",
+    )
+)
+print(dcb_obj.set_pfc(interface_name="eth4", pfc_per_priority=pfc_per_priority, tool_name="dcb"))
+
+# APP table management (dcb tool only) - e.g. DSCP 26 -> priority 3 (RoCEv2)
+print(dcb_obj.set_app(interface_name="eth4", proto_prio_map={26: 3}))
+print(dcb_obj.set_app(interface_name="eth4", proto_prio_map={"0x8906": 3}, selector="ethtype-prio"))
+print(dcb_obj.replace_app(interface_name="eth4", proto_prio_map={26: 5}))
+print(dcb_obj.del_app(interface_name="eth4", proto_prio_map={26: 5}))
+print(dcb_obj.flush_app(interface_name="eth4"))
+
+# iproute2 "dcb" tool - read configuration
+print(dcb_obj.get_dcb(interface_name="eth4", tool_name="dcb"))
+print(dcb_obj.get_ets(interface_name="eth4", tool_name="dcb"))
+print(dcb_obj.get_app(interface_name="eth4", tool_name="dcb"))
+print(dcb_obj.get_pfc(interface_name="eth4", tool_name="dcb"))
diff --git a/mfd_dcb/dcb_wrapper.py b/mfd_dcb/dcb_wrapper.py
new file mode 100644
index 0000000..3b9821b
--- /dev/null
+++ b/mfd_dcb/dcb_wrapper.py
@@ -0,0 +1,268 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: MIT
+
+"""Module for the iproute2 dcb tool."""
+
+import re
+
+from typing import TYPE_CHECKING, Any
+from mfd_const.qos import LOCAL_ETS, LOCAL_PFC, LOCAL_APP
+from .exceptions import DcbExecutionError, DcbExecutionProcessError
+
+if TYPE_CHECKING:
+    from mfd_connect import RPyCConnection
+
+APP_SELECTORS = (
+    "default-prio",
+    "ethtype-prio",
+    "stream-port-prio",
+    "dgram-port-prio",
+    "port-prio",
+    "dscp-prio",
+    "pcp-prio",
+)
+
+
+class DcbCli:
+    """DcbCli class provides DCB configuration read-out using the iproute2 dcb tool (dcbnl netlink)."""
+
+    def __init__(self, interface_name: str, connection: "RPyCConnection"):
+        """Initialize class parameters.
+
+        :param interface_name: tested interface
+        :param connection: connection
+        """
+        self.interface_name = interface_name
+        self._connection = connection
+
+    def get_dcb(self, numeric: bool = True) -> dict[str, dict[str, Any]]:
+        """Get complete DCB configuration for ETS, PFC and APP TLVs.
+
+        :param numeric: when True (default), APP protocol values are read numerically (``dcb -N app show``)
+        :return: Dictionary of QoS configuration
+        """
+        qos = {}
+        qos.update(self.get_ets())
+        qos.update(self.get_pfc())
+        qos.update(self.get_app(numeric=numeric))
+        return qos
+
+    def get_ets(self) -> dict[str, dict[str, Any]]:
+        """Get local ETS configuration of the interface via ``dcb ets show``.
+
+        :return: Dictionary of local ETS configuration
+        """
+        output = self._run("ets")
+        tc_tsa = self._extract_pairs(output, "tc-tsa")
+        tc_bw = self._extract_pairs(output, "tc-bw")
+        prio_tc = self._extract_pairs(output, "prio-tc")
+
+        ets_config: dict[str, dict[str, Any]] = {}
+        for tc, tsa in tc_tsa.items():
+            ets_config[str(tc)] = {
+                "TSA": tsa.upper(),
+                "Bandwidth": int(tc_bw.get(tc, 0)),
+                "Priorities": [],
+            }
+        for prio, tc in prio_tc.items():
+            ets_config.setdefault(
+                str(tc),
+                {"TSA": "", "Bandwidth": int(tc_bw.get(tc, 0)), "Priorities": []},
+            )
+            ets_config[str(tc)]["Priorities"].append(prio)
+        return {LOCAL_ETS: ets_config}
+
+    def get_pfc(self) -> dict[str, list[bool]]:
+        """Get local PFC configuration of the interface via ``dcb pfc show``.
+
+        :return: Dictionary of local PFC configuration
+        """
+        output = self._run("pfc")
+        prio_pfc = self._extract_pairs(output, "prio-pfc")
+        pfc_config = [False] * 8
+        for prio, state in prio_pfc.items():
+            if 0 <= prio < 8:
+                pfc_config[prio] = state == "on"
+        return {LOCAL_PFC: pfc_config}
+
+    def get_app(self, numeric: bool = True) -> dict[str, dict[str, Any]]:
+        """Get local APP table of the interface via ``dcb app show``.
+
+        :param numeric: when True (default), protocol values are read numerically via ``dcb -N app show``.
+        :return: Dictionary of local APP configuration
+        """
+        output = self._run("app", numeric=numeric)
+        app_config: dict[int, dict[str, Any]] = {}
+        for no, (selector, protocol, priority) in enumerate(self._extract_app_entries(output)):
+            app_config[no] = {
+                "selector": selector.replace("-prio", ""),
+                "priority": int(priority),
+                "protocol": protocol,
+            }
+        return {LOCAL_APP: app_config}
+
+    def set_ets(
+        self,
+        user_priority_mapping: list[tuple],
+        bandwidth_per_traffic_class: list[int],
+        mode: str = "ieee",
+    ) -> None:
+        """Set ETS configuration via ``dcb ets set``.
+
+        :param user_priority_mapping: User Priority mapping to Traffic Classes, list of tuples, each tuple is a next
+        Traffic Class with given User Priorities as integer values, e.g. [(0, 1), (2, 3), (5, 4), (6, 7)]
+        :param bandwidth_per_traffic_class: list of bandwidth percentage values, one per Traffic Class
+        :param mode: accepted for a common setter API across wrappers; not used by the iproute2 dcb tool
+        :raises DcbExecutionError: when the command writes to stderr
+        """
+        tc_tsa = " ".join(f"{tc}:ets" for tc in range(len(user_priority_mapping)))
+        tc_bw = " ".join(f"{tc}:{bw}" for tc, bw in enumerate(bandwidth_per_traffic_class))
+        prio_tc = " ".join(f"{prio}:{tc}" for tc, priorities in enumerate(user_priority_mapping) for prio in priorities)
+        command = f"dcb ets set dev {self.interface_name} willing off tc-tsa {tc_tsa} tc-bw {tc_bw} prio-tc {prio_tc}"
+        self._execute_dcb_command(command)
+
+    def set_pfc(self, pfc_per_priority: list[bool], mode: str = "ieee") -> None:
+        """Set Priority Flow Control via ``dcb pfc set``.
+
+        :param pfc_per_priority: list of 8 Bool values for each priority, True is for PFC ON
+        :param mode: accepted for a common setter API across wrappers; not used by the iproute2 dcb tool
+        :raises DcbExecutionError: when the command writes to stderr
+        """
+        prio_pfc = " ".join(f"{prio}:{'on' if pfc else 'off'}" for prio, pfc in enumerate(pfc_per_priority))
+        command = f"dcb pfc set dev {self.interface_name} prio-pfc {prio_pfc}"
+        self._execute_dcb_command(command)
+
+    def set_app(self, proto_prio_map: dict | list[tuple], *, selector: str = "dscp-prio") -> None:
+        """Add APP (protocol to priority) entries via ``dcb app add``.
+
+        :param proto_prio_map: mapping of protocol to priority, e.g. {26: 3} (DSCP 26 -> priority 3, RoCEv2)
+        :param selector: APP selector, e.g. "dscp-prio" (default), "ethtype-prio", "port-prio"
+        :raises DcbExecutionError: when the mapping is invalid or the command writes to stderr
+        """
+        proto_prio = self._format_app_map(proto_prio_map)
+        self._execute_dcb_command(f"dcb app add dev {self.interface_name} {selector} {proto_prio}")
+
+    def del_app(self, proto_prio_map: dict | list[tuple], *, selector: str = "dscp-prio") -> None:
+        """Delete specific APP (protocol to priority) entries via ``dcb app del``.
+
+        :param proto_prio_map: mapping of protocol to priority to remove, e.g. {26: 3}
+        :param selector: APP selector, e.g. "dscp-prio" (default), "ethtype-prio", "port-prio"
+        :raises DcbExecutionError: when the mapping is invalid or the command writes to stderr
+        """
+        proto_prio = self._format_app_map(proto_prio_map)
+        self._execute_dcb_command(f"dcb app del dev {self.interface_name} {selector} {proto_prio}")
+
+    def replace_app(self, proto_prio_map: dict | list[tuple], *, selector: str = "dscp-prio") -> None:
+        """Replace APP (protocol to priority) entries via ``dcb app replace``.
+
+        :param proto_prio_map: mapping of protocol to the new priority, e.g. {26: 5} (DSCP 26 -> priority 5)
+        :param selector: APP selector, e.g. "dscp-prio" (default), "ethtype-prio", "port-prio"
+        :raises DcbExecutionError: when the mapping is invalid or the command writes to stderr
+        """
+        proto_prio = self._format_app_map(proto_prio_map)
+        self._execute_dcb_command(f"dcb app replace dev {self.interface_name} {selector} {proto_prio}")
+
+    def flush_app(self, selector: str | None = "dscp-prio") -> None:
+        """Flush the APP table via ``dcb app flush``.
+
+        :param selector: APP selector to flush; when None, all APP entries are flushed
+        :raises DcbExecutionError: when the command writes to stderr
+        """
+        command = f"dcb app flush dev {self.interface_name}"
+        if selector:
+            command += f" {selector}"
+        self._execute_dcb_command(command)
+
+    @staticmethod
+    def _format_app_map(proto_prio_map: dict | list[tuple]) -> str:
+        """Format a protocol-to-priority mapping into the ``dcb app`` key argument.
+
+        :param proto_prio_map: mapping of protocol to priority, e.g. {26: 3, 30: 3}, [(26, 3)] or {"0x8906": 3}
+        :return: space separated "PROTO:PRIO" pairs, e.g. "26:3 30:3"
+        :raises DcbExecutionError: when the mapping is empty or malformed
+        """
+        pairs = proto_prio_map.items() if isinstance(proto_prio_map, dict) else proto_prio_map
+        try:
+            formatted = " ".join(f"{proto}:{int(prio)}" for proto, prio in pairs)
+        except (TypeError, ValueError) as err:
+            raise DcbExecutionError(f"Invalid proto_prio_map: {proto_prio_map}") from err
+        if not formatted:
+            raise DcbExecutionError("proto_prio_map must contain at least one protocol:priority entry")
+        return formatted
+
+    def _execute_dcb_command(self, cmd_list: list[str] | str) -> None:
+        """Execute iproute2 dcb tool command(s) and validate the return code and stderr.
+
+        :param cmd_list: command or list of commands to execute
+        :raises DcbExecutionProcessError: when a command returns a non-zero return code
+        :raises DcbExecutionError: when a command writes to stderr
+        """
+        cmd_list = cmd_list if isinstance(cmd_list, list) else [cmd_list]
+        for cmd in cmd_list:
+            output = self._connection.execute_command(
+                cmd,
+                custom_exception=DcbExecutionProcessError,
+                expected_return_codes=[0],
+            )
+            if output.stderr:
+                raise DcbExecutionError(f"Error while executing '{cmd}': {output.stderr}")
+
+    def _run(self, attribute: str, numeric: bool = False) -> str:
+        """Run ``dcb <attribute> show`` for the interface and return stdout.
+
+        :param attribute: dcb object to read - pfc, ets or app
+        :param numeric: when True, pass the global ``-N`` option so values are printed numerically
+        :return: command stdout
+        :raises DcbExecutionError: when the command reports an error on stderr
+        """
+        numeric_opt = "-N " if numeric else ""
+        command = f"dcb {numeric_opt}{attribute} show dev {self.interface_name}"
+        output = self._connection.execute_command(
+            command,
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+        if output.stderr:
+            raise DcbExecutionError(f"Error while executing dcb {command}: {output.stderr}")
+        return output.stdout
+
+    @staticmethod
+    def _extract_app_entries(output: str) -> list[tuple[str, str, str]]:
+        """Extract APP table entries as an ordered sequence of ``(selector, protocol, priority)``.
+
+        :param output: ``dcb app show`` stdout
+        :return: ordered list of ``(selector, protocol, priority)`` tuples
+        """
+        entries: list[tuple[str, str, str]] = []
+        current_selector = None
+        for token in output.split():
+            if token in APP_SELECTORS:
+                current_selector = token
+                continue
+            if not current_selector:
+                continue
+            if current_selector == "default-prio":
+                entries.append((current_selector, "0", token))
+            elif ":" in token:
+                protocol, _, priority = token.rpartition(":")
+                entries.append((current_selector, protocol, priority))
+        return entries
+
+    @staticmethod
+    def _extract_pairs(output: str, keyword: str) -> dict[Any, str]:
+        """Extract the ``KEY:VALUE`` pairs that follow a dcb attribute keyword.
+
+        :param output: dcb command stdout
+        :param keyword: attribute keyword to read the pairs for, e.g. "prio-pfc", "tc-tsa", "dscp-prio"
+        :return: mapping of parsed key to value string
+        """
+        match = re.search(rf"(?<![-\w]){re.escape(keyword)}\s+((?:\S+:\S+\s*)+)", output)
+        pairs: dict[Any, str] = {}
+        if not match:
+            return pairs
+        for token in match.group(1).split():
+            if ":" not in token:
+                break
+            key, _, value = token.partition(":")
+            pairs[int(key) if key.isdigit() else key] = value
+        return pairs
diff --git a/mfd_dcb/dcbnl_wrapper.py b/mfd_dcb/dcbnl_wrapper.py
index f77321b..a6d57d0 100644
--- a/mfd_dcb/dcbnl_wrapper.py
+++ b/mfd_dcb/dcbnl_wrapper.py
@@ -5,7 +5,7 @@
 
 import re
 
-from typing import Dict, TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any
 from mfd_const.qos import LOCAL_ETS, LOCAL_PFC, LOCAL_APP, DCB_TOOL_PATH_LNX
 from .exceptions import DcbExecutionError, DcbExecutionProcessError
 
@@ -25,7 +25,7 @@ def __init__(self, interface_name: str, connection: "RPyCConnection"):
         self.interface_name = interface_name
         self._connection = connection
 
-    def get_dcb(self) -> Dict[str, Dict[str, Any]]:
+    def get_dcb(self) -> dict[str, dict[str, Any]]:
         """Get complete DCB configuration for ETS, PFC and APP TLVs, local and remote.
 
         :return: Dictionary of QoS configuration
@@ -36,7 +36,7 @@ def get_dcb(self) -> Dict[str, Dict[str, Any]]:
         qos.update(self.get_app(from_cache=True))
         return qos
 
-    def get_ets(self, from_cache: bool = False) -> Dict[str, Dict[str, Any]]:
+    def get_ets(self, from_cache: bool = False) -> dict[str, dict[str, Any]]:
         """Get local ETS TLV configuration of the interface_name.
 
         :param from_cache: use current configuration output
@@ -46,7 +46,7 @@ def get_ets(self, from_cache: bool = False) -> Dict[str, Dict[str, Any]]:
             self._read_qos_configuration()
         return self._parse_ets(LOCAL_ETS)
 
-    def get_pfc(self, from_cache: bool = False) -> Dict[str, bool]:
+    def get_pfc(self, from_cache: bool = False) -> dict[str, bool]:
         """Get local PFC TLV configuration of the interface_name.
 
         :param from_cache: use current configuration output
@@ -56,7 +56,7 @@ def get_pfc(self, from_cache: bool = False) -> Dict[str, bool]:
             self._read_qos_configuration()
         return self._parse_pfc(LOCAL_PFC)
 
-    def get_app(self, from_cache: bool = False) -> Dict[str, Dict[str, Any]]:
+    def get_app(self, from_cache: bool = False) -> dict[str, dict[str, Any]]:
         """Get local APP TLV configuration of the interface_name.
 
         :param from_cache: use current configuration output
@@ -81,7 +81,7 @@ def _read_qos_configuration(self) -> None:
             raise DcbExecutionError(f"Error while executing dcbnl {command}: {output.stderr}")
         self._qos_output = output.stdout
 
-    def _parse_ets(self, key: str) -> Dict[str, Dict[str, int]]:
+    def _parse_ets(self, key: str) -> dict[str, dict[str, int]]:
         """Get local or remote QoS traffic classes as dictionary.
 
         :param key: key in configuration dictionary for ETS
@@ -108,7 +108,7 @@ def _parse_ets(self, key: str) -> Dict[str, Dict[str, int]]:
                 continue
         return {key: ets_config}
 
-    def _parse_pfc(self, key: str) -> Dict[str, bool]:
+    def _parse_pfc(self, key: str) -> dict[str, bool]:
         """Get operational or remote QoS flow control as dictionary.
 
         :param key: key in configuration dictionary for PFC
@@ -122,7 +122,7 @@ def _parse_pfc(self, key: str) -> Dict[str, bool]:
                 pfc_config[i] = pfc_bin & (pow(2, i)) > 0
         return {key: pfc_config}
 
-    def _parse_app(self, key: str) -> Dict[str, Dict[str, str]]:
+    def _parse_app(self, key: str) -> dict[str, dict[str, str]]:
         """Get operational or remote QoS apps table.
 
         :param key: key in configuration dictionary for PFC
diff --git a/mfd_dcb/dcbtool_wrapper.py b/mfd_dcb/dcbtool_wrapper.py
index 966be4a..b5f96a9 100644
--- a/mfd_dcb/dcbtool_wrapper.py
+++ b/mfd_dcb/dcbtool_wrapper.py
@@ -3,17 +3,21 @@
 
 """Module for dcbtool."""
 
+import logging
 import re
-from typing import List, Dict, Any, TYPE_CHECKING
+from typing import Any, TYPE_CHECKING
+from mfd_common_libs import log_levels
 from mfd_const.qos import LOCAL_ETS, LOCAL_PFC
 from .exceptions import DcbExecutionError, DcbExecutionProcessError
 
 if TYPE_CHECKING:
     from mfd_connect import RPyCConnection
 
+logger = logging.getLogger(__name__)
+
 
 class DcbTool:
-    """DcbTool class provide check DCB configuration using dcbtool tool."""
+    """DcbTool class provide check and set DCB configuration using dcbtool/lldptool tools."""
 
     def __init__(self, interface_name: str, connection: "RPyCConnection"):
         """Initialize class parameters.
@@ -24,7 +28,120 @@ def __init__(self, interface_name: str, connection: "RPyCConnection"):
         self.interface_name = interface_name
         self._connection = connection
 
-    def get_dcb(self) -> Dict[str, Dict[str, Any]]:
+    @staticmethod
+    def _execute_qos_command_and_check_output(connection: "RPyCConnection", cmd_list: list[str] | str) -> str:
+        """Execute QoS command(s) and check output for known problems, trying to recover from them.
+
+        :param connection: connection used to execute the commands
+        :param cmd_list: list of commands (or a single command) to execute
+        :returns: stdout of the last executed command
+        :raises DcbExecutionError: for non zero return code other than known errors
+        """
+        lldpad_fail_recovery_cmd_list = [
+            "systemctl stop lldpad",
+            "systemctl stop lldpad.socket",
+            "systemctl reset-failed lldpad.service",
+            "systemctl start lldpad.socket",
+            "systemctl start lldpad",
+        ]
+        problems_list = [
+            "lldpad.socket failed",
+            "Connection refused",
+            "command timed out",
+            "Device not capable",
+        ]
+        rerun_cmd_after_problem_list = ["Connection refused", "command timed out"]
+        cmd_list = cmd_list if isinstance(cmd_list, list) else [cmd_list]
+        for cmd in cmd_list:
+            cli_out = connection.execute_command(cmd, expected_return_codes={})
+            out = cli_out.stdout
+            if any(problem in out for problem in problems_list):
+                if "lldpad.socket failed" in out:
+                    raise DcbExecutionError("lldpad.socket failed Exception")
+                elif "Connection refused" in out:
+                    logger.log(
+                        level=log_levels.MODULE_DEBUG,
+                        msg="Connection refused, stop and start lldp.",
+                    )
+                    print("Connection refused, stop and start lldp.")
+                elif "Device not capable" in out:
+                    raise DcbExecutionError("Please check/update driver")
+                for recovery_cmd in lldpad_fail_recovery_cmd_list:
+                    connection.execute_command(
+                        recovery_cmd,
+                        custom_exception=DcbExecutionError,
+                        expected_return_codes={0, 1},
+                    )
+                for problem in rerun_cmd_after_problem_list:
+                    if problem in out:
+                        logger.log(
+                            level=log_levels.MODULE_DEBUG,
+                            msg=f"Rerun command {cmd} after recovery action, when {problem} occurred.",
+                        )
+                        out += connection.execute_command(
+                            cmd,
+                            custom_exception=DcbExecutionError,
+                            expected_return_codes={0},
+                        ).stdout
+            if cli_out.return_code != 0:
+                raise DcbExecutionError(f"Exception from Unknown Error: {out}")
+        return out
+
+    def set_ets(
+        self,
+        user_priority_mapping: list[tuple],
+        bandwidth_per_traffic_class: list[int],
+        mode: str = "ieee",
+    ) -> None:
+        """Set ETS via lldptool (ieee) / dcbtool (cee).
+
+        :param user_priority_mapping: User Priority mapping to Traffic Classes
+        :param bandwidth_per_traffic_class: bandwidth percentage per Traffic Class
+        :param mode: DCBX mode version - ieee|cee
+        :raises DcbExecutionError: when a wrong mode is specified or the command return code is unexpected
+        """
+        if mode.lower() == "ieee":
+            up2tc = ""
+            tcbw = ",".join(map(str, bandwidth_per_traffic_class))
+            tsa = ",".join([f"{i}:ets" for i in range(len(user_priority_mapping))])
+            command_list = [
+                f"lldptool -Ti {self.interface_name} -V ETS-CFG enableTx=yes willing=no "
+                f"up2tc={up2tc} tcbw={tcbw} tsa={tsa}",
+                f"lldptool -Ti {self.interface_name} -V ETS-REC enableTx=yes up2tc={up2tc} tcbw={tcbw} tsa={tsa}",
+            ]
+        elif mode.lower() == "cee":
+            pgid_list = [0] * 8
+            pgpct_list = [0] * 8
+            for tc, priorities in enumerate(user_priority_mapping):
+                for prio in priorities:
+                    pgid_list[prio] = tc
+                pgpct_list[tc] = bandwidth_per_traffic_class[tc]
+            pgid = "".join(map(str, pgid_list))
+            pgpct = ",".join(map(str, pgpct_list))
+            command_list = [f"dcbtool sc {self.interface_name} pg w:0 e:1 a:1 pgid:{pgid} pgpct:{pgpct}"]
+        else:
+            raise DcbExecutionError(f"Incorrect mode: {mode}, expected modes: ieee or cee")
+        self._execute_qos_command_and_check_output(self._connection, command_list)
+
+    def set_pfc(self, pfc_per_priority: list[bool], mode: str = "ieee") -> None:
+        """Set PFC via lldptool (ieee) / dcbtool (cee).
+
+        :param pfc_per_priority: list of 8 Bool values for each priority, True is for PFC ON
+        :param mode: DCBX mode version - ieee|cee
+        :raises DcbExecutionError: when a wrong mode is specified or the command return code is unexpected
+        """
+        if mode.lower() == "ieee":
+            enabled_list = [priority for priority, pfc in enumerate(pfc_per_priority) if pfc]
+            enabled = ",".join(map(str, enabled_list))
+            command = f"lldptool -Ti {self.interface_name} -V PFC enabled={enabled} enableTx=yes willing=no"
+        elif mode.lower() == "cee":
+            pfcup = "".join(str(int(priority)) for priority in pfc_per_priority)
+            command = f"dcbtool sc {self.interface_name} pfc w:0 e:1 a:1 pfcup:{pfcup}"
+        else:
+            raise DcbExecutionError(f"Incorrect mode: {mode}, expected modes: ieee or cee")
+        self._execute_qos_command_and_check_output(self._connection, command)
+
+    def get_dcb(self) -> dict[str, dict[str, Any]]:
         """Get complete DCB configuration for ETS, PFC and APP TLVs, local and remote.
 
         :return: Dictionary of QoS configuration
@@ -34,7 +151,7 @@ def get_dcb(self) -> Dict[str, Dict[str, Any]]:
         qos.update(self.get_pfc())
         return qos
 
-    def get_ets(self, from_cache: bool = False) -> Dict[str, Dict[str, Any]]:
+    def get_ets(self, from_cache: bool = False) -> dict[str, dict[str, Any]]:
         """Get local ETS TLV configuration of the interface_name.
 
         :param from_cache: use current configuration output
@@ -44,7 +161,7 @@ def get_ets(self, from_cache: bool = False) -> Dict[str, Dict[str, Any]]:
             self._read_qos_configuration(check_type="pg")
         return self._parse_qos(LOCAL_ETS)
 
-    def get_pfc(self, from_cache: bool = False) -> Dict[str, bool]:
+    def get_pfc(self, from_cache: bool = False) -> dict[str, bool]:
         """Get local PFC TLV configuration of the interface_name.
 
         :param from_cache: use current configuration output
@@ -66,7 +183,7 @@ def _read_qos_configuration(self, check_type: str) -> None:
             raise DcbExecutionError(f"Error while executing dcbnl {command}: {output.stderr}")
         self._qos_output = output.stdout
 
-    def _parse_qos(self, key: str) -> Dict[str, Dict[str, Any]]:
+    def _parse_qos(self, key: str) -> dict[str, dict[str, Any]]:
         """Get local or remote QoS traffic classes as dictionary.
 
         :param key: key in configuration dictionary for ETS & PFC
@@ -115,7 +232,7 @@ def _parse_lines(
                 raise DcbExecutionError("No match found for PFC")
 
     @staticmethod
-    def _parse_line(line: str, pattern: str) -> List[int]:
+    def _parse_line(line: str, pattern: str) -> list[int]:
         """To parse line and return dict with data.
 
         :param line: line to search
@@ -142,7 +259,11 @@ def _set_bw_config(bw_list: list, ets_config: dict) -> None:
         for up, bw in enumerate(bw_list):
             if bw > 0:
                 current_tc = str(up)
-                ets_config[current_tc] = {"TSA": "ETS", "Bandwidth": bw, "Priorities": []}
+                ets_config[current_tc] = {
+                    "TSA": "ETS",
+                    "Bandwidth": bw,
+                    "Priorities": [],
+                }
 
     @staticmethod
     def _set_up_config(tc_list: list, ets_config: dict) -> None:
diff --git a/mfd_dcb/linux.py b/mfd_dcb/linux.py
index 44c7b43..2ab98ac 100644
--- a/mfd_dcb/linux.py
+++ b/mfd_dcb/linux.py
@@ -7,7 +7,7 @@
 
 import logging
 import re
-from typing import List, Union, Dict, TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any
 import time
 
 from mfd_common_libs import add_logging_level, log_levels, os_supported
@@ -19,6 +19,7 @@
 from .mfd_dcb import Dcb
 import mfd_dcb.dcbnl_wrapper
 import mfd_dcb.dcbtool_wrapper
+import mfd_dcb.dcb_wrapper
 
 
 if TYPE_CHECKING:
@@ -32,6 +33,9 @@
 class LinuxDcb(Dcb):
     """Class to handle Dcb in Linux."""
 
+    _ALL_TOOL_NAMES = ("dcb", "dcbnl", "dcbtool")
+    _SETTER_TOOL_NAMES = ("dcb", "dcbtool")
+
     @os_supported(OSName.LINUX)
     def __init__(self, *, connection: "RPyCConnection") -> None:
         """Initialize Linux DCB."""
@@ -39,7 +43,7 @@ def __init__(self, *, connection: "RPyCConnection") -> None:
         # create object for mfd Ethtool
         self.ethtool_obj = Ethtool(connection=connection)
 
-    def _execute_qos_command_and_check_output(self, cmd_list: Union[List[str], str]) -> str:
+    def _execute_qos_command_and_check_output(self, cmd_list: list[str] | str) -> str:
         """
         Execute command and check output for unexpected problems and try to fix them.
 
@@ -47,58 +51,38 @@ def _execute_qos_command_and_check_output(self, cmd_list: Union[List[str], str])
         :returns stdout
         :raises DcbExecutionError: for non zero return code other than known errors.
         """
-        lldpad_fail_recovery_cmd_list = [
-            "systemctl stop lldpad",
-            "systemctl stop lldpad.socket",
-            "systemctl reset-failed lldpad.service",
-            "systemctl start lldpad.socket",
-            "systemctl start lldpad",
-        ]
-        problems_list = [
-            "lldpad.socket failed",
-            "Connection refused",
-            "command timed out",
-            "Device not capable",
-        ]
-        rerun_cmd_after_problem_list = ["Connection refused", "command timed out"]
-        cmd_list = cmd_list if isinstance(cmd_list, list) else [cmd_list]
-        for cmd in cmd_list:
-            cli_out = self._connection.execute_command(cmd, expected_return_codes={})
-            out = cli_out.stdout
-            if any(problem in out for problem in problems_list):
-                if "lldpad.socket failed" in out:
-                    raise DcbExecutionError("lldpad.socket failed Exception")
-                elif "Connection refused" in out:
-                    logger.log(
-                        level=log_levels.MODULE_DEBUG,
-                        msg="Connection refused, stop and start lldp.",
-                    )
-                    print("Connection refused, stop and start lldp.")
-                elif "Device not capable" in out:
-                    raise DcbExecutionError("Please check/update driver")
-                for recovery_cmd in lldpad_fail_recovery_cmd_list:
-                    self._connection.execute_command(
-                        recovery_cmd, custom_exception=DcbExecutionError, expected_return_codes={0, 1}
-                    )
-                for problem in rerun_cmd_after_problem_list:
-                    if problem in out:
-                        logger.log(
-                            level=log_levels.MODULE_DEBUG,
-                            msg=f"Rerun command {cmd} after recovery action, when {problem} occurred.",
-                        )
-                        out += self._connection.execute_command(
-                            cmd, custom_exception=DcbExecutionError, expected_return_codes={0}
-                        ).stdout
-            if cli_out.return_code != 0:
-                raise DcbExecutionError(f"Exception from Unknown Error: {out}")
-        return out
+        return mfd_dcb.dcbtool_wrapper.DcbTool._execute_qos_command_and_check_output(self._connection, cmd_list)
+
+    def _get_tool(
+        self,
+        interface_name: str,
+        tool_name: str,
+        is_setter: bool = False,
+    ) -> object:
+        """Produce particular tool based on tool name.
+
+        :param tool_name: name of tool - dcbnl / dcbtool / dcb.
+        :param interface_name: Name of the Network Interface under Test.
+        :param is_setter: when True, restrict allowed tools to the setter-capable ones (dcbtool, dcb).
+        :raises DcbExecutionError: when tool_name is not a supported tool
+        :return: instance of tool type
+        """
+        allowed_tool_names = self._SETTER_TOOL_NAMES if is_setter else self._ALL_TOOL_NAMES
+        if tool_name not in allowed_tool_names:
+            raise DcbExecutionError(f"Unsupported tool_name '{tool_name}', expected one of {allowed_tool_names}")
+        if tool_name == "dcbtool":
+            return mfd_dcb.dcbtool_wrapper.DcbTool(interface_name, self._connection)
+        elif tool_name == "dcb":
+            return mfd_dcb.dcb_wrapper.DcbCli(interface_name, self._connection)
+        return mfd_dcb.dcbnl_wrapper.Dcbnl(interface_name, self._connection)
 
     def set_ets(
         self,
-        user_priority_mapping: List[(int)],
-        bandwidth_per_traffic_class: List[int],
+        user_priority_mapping: list[(int)],
+        bandwidth_per_traffic_class: list[int],
         interface_name: str,
         mode: str = "ieee",
+        tool_name: str = "dcbtool",
     ) -> None:
         """Set ETS configuration based on User Priority mapping to Traffic Classes and given bandwidth.
 
@@ -108,73 +92,123 @@ def set_ets(
         :param bandwidth_per_traffic_class: List of bandwidth percentage values, each integer value is a bandwidth for
         next Traffic Class
         bandwidth_per_traffic_class = [25, 25, 25, 25]
-        :param mode: DCBX mode version - ieee|cee
         :param interface_name: Network adapter
+        :param mode: DCBX mode version - ieee|cee (used by the lldptool (ieee) / dcbtool (cee) path)
+        :param tool_name: name of tool to use. Default "dcbtool" uses the lldptool (ieee) / dcbtool (cee)
+        path selected by mode; "dcb" applies configuration via the iproute2 dcb tool (dcbnl netlink).
         :raises DcbExecutionError: when user_priority_mapping is less or equal to 1 or it is not the same length as
-        bandwidth_per_traffic_class or wrong mode is specified
+        bandwidth_per_traffic_class, when tool_name is unsupported or wrong mode is specified
         :raises DcbExecutionError: when return code of bash command is unexpected
         """
         if not isinstance((user_priority_mapping and bandwidth_per_traffic_class), list):
             raise DcbExecutionError("user_priority_mapping and bandwidth_per_traffic_class must be list")
         if len(user_priority_mapping) != len(bandwidth_per_traffic_class):
             raise DcbExecutionError(
-                "Number of Traffic Classes for User Priority mapping and bandwidth percentage lists must " "be the same"
+                "Number of Traffic Classes for User Priority mapping and bandwidth percentage lists must be the same"
             )
-        if mode.lower() == "ieee":
-            up2tc_list = []
-            up2tc = ",".join(up2tc_list)
-            tcbw = ",".join(map(str, bandwidth_per_traffic_class))
-            tsa = [",".join([f"{i}:ets" for i in range(len(user_priority_mapping))])][0]
-            command_list = [
-                f"lldptool -Ti {interface_name} -V ETS-CFG enableTx=yes willing=no up2tc={up2tc} tcbw={tcbw} tsa={tsa}",
-                f"lldptool -Ti {interface_name} -V ETS-REC enableTx=yes up2tc={up2tc} tcbw={tcbw} tsa={tsa}",
-            ]
-        elif mode.lower() == "cee":
-            pgid_list = [0] * 8
-            pgpct_list = [0] * 8
-            for tc, priorities in enumerate(user_priority_mapping):
-                for prio in priorities:
-                    pgid_list[prio] = tc
-                pgpct_list[tc] = bandwidth_per_traffic_class[tc]
-            pgid = "".join(map(str, pgid_list))
-            pgpct = ",".join(map(str, pgpct_list))
-            command_list = [f"dcbtool sc {interface_name} pg w:0 e:1 a:1 pgid:{pgid} pgpct:{pgpct}"]
-        else:
-            raise DcbExecutionError(f"Incorrect mode: {mode}, expected modes: ieee or cee")
-        self._execute_qos_command_and_check_output(command_list)
+        self._get_tool(interface_name, tool_name, is_setter=True).set_ets(
+            user_priority_mapping, bandwidth_per_traffic_class, mode=mode
+        )
 
     def set_pfc(
         self,
         interface_name: str,
-        pfc_per_priority: List[bool],
+        pfc_per_priority: list[bool],
         mode: str = "ieee",
+        tool_name: str = "dcbtool",
     ) -> None:
         """Turn ON or OFF Priority Flow Control for each User Priority.
 
         :param interface_name: Network adapter
         :param pfc_per_priority: List of 8 Bool values for each priority, True is for PFC ON
-        :param mode: DCBX mode version - ieee|cee
-        :raises DcbExecutionError: when pfc_per_priority list length is not equal to 8 or wrong mode is specified
+        :param mode: DCBX mode version - ieee|cee (used by the lldptool (ieee) / dcbtool (cee) path)
+        :param tool_name: name of tool to use. Default "dcbtool" uses the lldptool (ieee) / dcbtool (cee)
+        path selected by mode; "dcb" applies configuration via the iproute2 dcb tool (dcbnl netlink).
+        :raises DcbExecutionError: when pfc_per_priority list length is not equal to 8, when tool_name is
+        unsupported or wrong mode is specified
         :raises DcbExecutionError: when return code of bash command is unexpected
         """
         if len(pfc_per_priority) != 8:
             raise DcbExecutionError(
-                f"Unexpected length for pfc_per_priority. " f"Expected: 8, actual length: {len(pfc_per_priority)}"
+                f"Unexpected length for pfc_per_priority. Expected: 8, actual length: {len(pfc_per_priority)}"
             )
-        if mode.lower() == "ieee":
-            enabled_list = []
-            for priority, pfc in enumerate(pfc_per_priority):
-                if pfc:
-                    enabled_list.append(priority)
-            enabled = ",".join(map(str, enabled_list))
-            command = f"lldptool -Ti {interface_name} -V PFC enabled={enabled} enableTx=yes willing=no"
-        elif mode.lower() == "cee":
-            pfcup_list = [int(priority) for priority in pfc_per_priority]
-            pfcup = "".join(map(str, pfcup_list))
-            command = f"dcbtool sc {interface_name} pfc w:0 e:1 a:1 pfcup:{pfcup}"
-        else:
-            raise DcbExecutionError(f"Incorrect mode: {mode}, expected modes: ieee or cee")
-        self._execute_qos_command_and_check_output(command)
+        self._get_tool(interface_name, tool_name, is_setter=True).set_pfc(pfc_per_priority, mode=mode)
+
+    def set_app(
+        self,
+        interface_name: str,
+        proto_prio_map: dict | list[tuple],
+        *,
+        selector: str = "dscp-prio",
+        tool_name: str = "dcb",
+    ) -> None:
+        """Add APP (protocol to priority) entries via the iproute2 dcb tool.
+
+        :param interface_name: Network adapter
+        :param proto_prio_map: mapping of protocol to priority, e.g. {26: 3} (DSCP 26 -> priority 3, RoCEv2)
+        :param selector: APP selector, e.g. "dscp-prio" (default), "ethtype-prio", "port-prio"
+        :param tool_name: name of tool to use. Only "dcb" is supported (APP table setting is dcb-tool only).
+        :raises DcbExecutionError: when tool_name is not "dcb", the mapping is invalid or the command fails
+        """
+        if tool_name != "dcb":
+            raise DcbExecutionError('set_app is only supported with tool_name="dcb"')
+        self._get_tool(interface_name, tool_name).set_app(proto_prio_map, selector=selector)
+
+    def del_app(
+        self,
+        interface_name: str,
+        proto_prio_map: dict | list[tuple],
+        *,
+        selector: str = "dscp-prio",
+        tool_name: str = "dcb",
+    ) -> None:
+        """Delete specific APP (protocol to priority) entries via the iproute2 dcb tool.
+
+        :param interface_name: Network adapter
+        :param proto_prio_map: mapping of protocol to priority to remove, e.g. {26: 3}
+        :param selector: APP selector, e.g. "dscp-prio" (default), "ethtype-prio", "port-prio"
+        :param tool_name: name of tool to use. Only "dcb" is supported.
+        :raises DcbExecutionError: when tool_name is not "dcb", the mapping is invalid or the command fails
+        """
+        if tool_name != "dcb":
+            raise DcbExecutionError('del_app is only supported with tool_name="dcb"')
+        self._get_tool(interface_name, tool_name).del_app(proto_prio_map, selector=selector)
+
+    def replace_app(
+        self,
+        interface_name: str,
+        proto_prio_map: dict | list[tuple],
+        *,
+        selector: str = "dscp-prio",
+        tool_name: str = "dcb",
+    ) -> None:
+        """Replace APP (protocol to priority) entries via the iproute2 dcb tool.
+
+        :param interface_name: Network adapter
+        :param proto_prio_map: mapping of protocol to the new priority, e.g. {26: 5} (DSCP 26 -> priority 5)
+        :param selector: APP selector, e.g. "dscp-prio" (default), "ethtype-prio", "port-prio"
+        :param tool_name: name of tool to use. Only "dcb" is supported.
+        :raises DcbExecutionError: when tool_name is not "dcb", the mapping is invalid or the command fails
+        """
+        if tool_name != "dcb":
+            raise DcbExecutionError('replace_app is only supported with tool_name="dcb"')
+        self._get_tool(interface_name, tool_name).replace_app(proto_prio_map, selector=selector)
+
+    def flush_app(
+        self,
+        interface_name: str,
+        selector: str | None = "dscp-prio",
+        *,
+        tool_name: str = "dcb",
+    ) -> None:
+        """Flush the APP table via the iproute2 dcb tool.
+
+        :param tool_name: name of tool to use. Only "dcb" is supported.
+        :raises DcbExecutionError: when tool_name is not "dcb" or the command fails
+        """
+        if tool_name != "dcb":
+            raise DcbExecutionError('flush_app is only supported with tool_name="dcb"')
+        self._get_tool(interface_name, tool_name).flush_app(selector)
 
     def restart_lldpad(self, action: str = "default") -> None:
         """Restart/stop lldpad service and lldp agent daemon.
@@ -293,7 +327,10 @@ def set_willing(
         """
         if mode == "ieee":
             willing = "yes" if enable else "no"
-            cmd_list = [f"lldptool -Ti {interface_name} -V CEE-DCBX enableTx=no", "dcbtool sc dcbx v:cee"]
+            cmd_list = [
+                f"lldptool -Ti {interface_name} -V CEE-DCBX enableTx=no",
+                "dcbtool sc dcbx v:cee",
+            ]
             if not is_fwlldp_enabled:
                 lldptool_tlv = f"lldptool -Ti {interface_name} -V"
                 cmd_list.extend(
@@ -305,7 +342,10 @@ def set_willing(
                 )
         elif mode == "cee":
             willing = "1" if enable else "0"
-            cmd_list = [f"lldptool -Ti {interface_name} -V CEE-DCBX enableTx=yes", "dcbtool sc dcbx v:force-cee"]
+            cmd_list = [
+                f"lldptool -Ti {interface_name} -V CEE-DCBX enableTx=yes",
+                "dcbtool sc dcbx v:force-cee",
+            ]
             if not is_fwlldp_enabled:
                 cmd_list.extend(
                     [
@@ -324,7 +364,7 @@ def _get_frame_list(
         priority: int,
         is_10g_adapter: bool = False,
         is_40g_adapter: bool = False,
-    ) -> List[str]:
+    ) -> list[str]:
         """Get frame list based on detected adapter.
 
         :param priority: priority number to create proper frame
@@ -355,11 +395,11 @@ def _get_frame_list(
     def _get_pfc_counter_dict(
         self,
         out: str,
-        pfc_counters: Dict[str, str],
-        priority: List[int],
-        frame_list: List[bool],
+        pfc_counters: dict[str, str],
+        priority: list[int],
+        frame_list: list[bool],
         adapter_10G: bool = False,
-    ) -> Dict[str, Dict[str, str]]:
+    ) -> dict[str, dict[str, str]]:
         """
         pfc_counters dict.
 
@@ -376,7 +416,12 @@ def _get_pfc_counter_dict(
             match = re.search(xframeregex, out, re.MULTILINE)
             if not match:
                 raise DcbExecutionError(f"No match found for {frame_type} for _get_pfc_counter_dict")
-            frame_type_10g = [("tx_pb", "pxon"), ("tx_pb", "pxoff"), ("rx_pb", "pxon"), ("rx_pb", "pxoff")]
+            frame_type_10g = [
+                ("tx_pb", "pxon"),
+                ("tx_pb", "pxoff"),
+                ("rx_pb", "pxon"),
+                ("rx_pb", "pxoff"),
+            ]
             frame_type_non_10g = [
                 ("tx_priority", "xon"),
                 ("tx_priority", "xoff"),
@@ -403,7 +448,7 @@ def get_pfc_counters(
         interface_name: str,
         is_10g_adapter: bool = False,
         is_40g_adapter: bool = False,
-    ) -> Dict[str, Dict[str, str]]:
+    ) -> dict[str, dict[str, str]]:
         """Get PFC counters values for all User Priorities.
 
         :param interface_name: Name of the Network adapter under Test.
@@ -447,7 +492,7 @@ def get_dcb(
         self,
         interface_name: str,
         tool_name: str = "dcbnl",
-    ) -> Dict[str, Dict[str, str]]:
+    ) -> dict[str, dict[str, str]]:
         """Get complete DCB configuration for ETS, PFC and APP TLVs, local and remote.
 
         :param tool_name: name of tool, dcbnl/dcbtool to get DCB config.
@@ -461,7 +506,7 @@ def get_ets(
         self,
         interface_name: str,
         tool_name: str = "dcbnl",
-    ) -> Dict[str, Dict[str, Any]]:
+    ) -> dict[str, dict[str, Any]]:
         """
         Get local ETS TLV configuration of the interface.
 
@@ -472,27 +517,11 @@ def get_ets(
         self.tool = self._get_tool(tool_name=tool_name, interface_name=interface_name)
         return self.tool.get_ets()
 
-    def _get_tool(
-        self,
-        interface_name: str,
-        tool_name: str = "dcbnl",
-    ) -> object:
-        """Produce particular tool based on tool name.
-
-        :param tool_name: name of tool, dcbtool / dcbnl.
-        :param interface_name: Name of the Network Interface under Test.
-        :return: instance of tool type
-        """
-        if "dcbtool" in tool_name:
-            return mfd_dcb.dcbtool_wrapper.DcbTool(interface_name, self._connection)
-        else:
-            return mfd_dcb.dcbnl_wrapper.Dcbnl(interface_name, self._connection)
-
     def get_app(
         self,
         interface_name: str,
         tool_name: str = "dcbnl",
-    ) -> Dict[str, Dict[str, Any]]:
+    ) -> dict[str, dict[str, Any]]:
         """Get local APP TLV configuration of the Interface.
 
         :param tool_name: name of tool (dcbnl/dcbtool) to get local APP TLV config.
@@ -509,7 +538,7 @@ def get_pfc(
         self,
         interface_name: str,
         tool_name: str = "dcbnl",
-    ) -> Dict[str, Dict[str, Any]]:
+    ) -> dict[str, dict[str, Any]]:
         """Get local PFC TLV configuration of the adapter.
 
         :param tool_name: name of tool, dcbtool/dcbtool
@@ -557,7 +586,11 @@ def verify_dcb(
                 ets_correct = self.verify_ets(interface_name, dcb_map, ets_config=dcb_config.get("LOCAL_ETS"))
                 pfc_correct = self.verify_pfc(interface_name, dcb_map, pfc_config=dcb_config.get("LOCAL_PFC"))
                 app_correct = (
-                    self.verify_app(interface_name, dcb_map, app_config=dcb_config.get("LOCAL_APP", {}))
+                    self.verify_app(
+                        interface_name,
+                        dcb_map,
+                        app_config=dcb_config.get("LOCAL_APP", {}),
+                    )
                     if "LOCAL_APP" in dcb_map
                     else True
                 )
@@ -566,7 +599,10 @@ def verify_dcb(
                     return True
 
             except DcbExecutionError as e:
-                logger.log(level=log_levels.MODULE_DEBUG, msg=f"Could not verify DCB configuration\n{e}")
+                logger.log(
+                    level=log_levels.MODULE_DEBUG,
+                    msg=f"Could not verify DCB configuration\n{e}",
+                )
 
             if attempt < retry:
                 self._retry_logic(switch, switch_port, interval, retry - attempt)
@@ -593,12 +629,15 @@ def _retry_logic(
         if switch and switch_port:
             logger.log(
                 level=log_levels.MODULE_DEBUG,
-                msg='Map did not load correctly, "shutdown",' f'and "no shutdown" port:{switch_port}.',
+                msg=f'Map did not load correctly, "shutdown",and "no shutdown" port:{switch_port}.',
             )
             for status in [True, False]:
                 switch.shutdown(status, switch_port)
                 time.sleep(5)
-        logger.log(level=log_levels.MODULE_DEBUG, msg=f"Retrying in {interval} seconds, {retries_left} tries remaining")
+        logger.log(
+            level=log_levels.MODULE_DEBUG,
+            msg=f"Retrying in {interval} seconds, {retries_left} tries remaining",
+        )
         time.sleep(interval)
 
     def verify_ets(
@@ -729,7 +768,10 @@ def verify_app(
                         )
                         correct = False
                 else:
-                    logger.log(level=log_levels.MODULE_DEBUG, msg=f"Missing port {port} in APP TLV")
+                    logger.log(
+                        level=log_levels.MODULE_DEBUG,
+                        msg=f"Missing port {port} in APP TLV",
+                    )
                     correct = False
 
         return correct
diff --git a/tests/unit/test_mfd_dcb/test_dcb_wrapper.py b/tests/unit/test_mfd_dcb/test_dcb_wrapper.py
new file mode 100644
index 0000000..3bdd2c4
--- /dev/null
+++ b/tests/unit/test_mfd_dcb/test_dcb_wrapper.py
@@ -0,0 +1,257 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: MIT
+
+"""Tests for `mfd_dcb.dcb_wrapper` module."""
+
+from textwrap import dedent
+
+import pytest
+from mfd_connect import RPyCConnection
+from mfd_connect.base import ConnectionCompletedProcess
+from mfd_const.qos import LOCAL_ETS, LOCAL_PFC, LOCAL_APP
+
+from mfd_dcb.dcb_wrapper import DcbCli
+from mfd_dcb.exceptions import DcbExecutionError, DcbExecutionProcessError
+
+
+class TestDcbCli:
+    @pytest.fixture()
+    def wrapper(self, mocker):
+        conn = mocker.create_autospec(RPyCConnection)
+        return DcbCli("eth1", conn)
+
+    def _out(self, stdout):
+        return ConnectionCompletedProcess(args="", stdout=stdout, return_code=0, stderr="")
+
+    def test_get_pfc(self, wrapper):
+        stdout = dedent(
+            """\
+            pfc-cap 4 macsec-bypass off delay 1949
+            prio-pfc 0:off 1:off 2:off 3:on 4:off 5:off 6:off 7:off
+            """
+        )
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        assert wrapper.get_pfc() == {LOCAL_PFC: [False, False, False, True, False, False, False, False]}
+
+    def test_get_ets(self, wrapper):
+        stdout = dedent(
+            """\
+            willing off ets-cap 4 cbs off
+            tc-bw 0:40 1:40 2:10 3:10 4:0 5:0 6:0 7:0
+            pg-bw 0:40 1:40 2:10 3:10 4:0 5:0 6:0 7:0
+            tc-tsa 0:ets 1:ets 2:ets 3:ets 4:strict 5:strict 6:strict 7:strict
+            prio-tc 0:0 1:1 2:2 3:3 4:0 5:0 6:0 7:0
+            reco-tc-bw 0:40 1:40 2:10 3:10 4:0 5:0 6:0 7:0
+            reco-tc-tsa 0:ets 1:ets 2:ets 3:ets 4:strict 5:strict 6:strict 7:strict
+            reco-prio-tc 0:0 1:1 2:2 3:3 4:0 5:0 6:0 7:0
+            """
+        )
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        ets = wrapper.get_ets()[LOCAL_ETS]
+        assert ets["0"] == {
+            "TSA": "ETS",
+            "Bandwidth": 40,
+            "Priorities": [0, 4, 5, 6, 7],
+        }
+        assert ets["1"] == {"TSA": "ETS", "Bandwidth": 40, "Priorities": [1]}
+        assert ets["2"] == {"TSA": "ETS", "Bandwidth": 10, "Priorities": [2]}
+        assert ets["3"] == {"TSA": "ETS", "Bandwidth": 10, "Priorities": [3]}
+        assert ets["4"] == {"TSA": "STRICT", "Bandwidth": 0, "Priorities": []}
+
+    def test_get_ets_ignores_reco_only_fields(self, wrapper):
+        stdout = dedent(
+            """\
+            willing off ets-cap 8 cbs off
+            reco-tc-bw 0:40 1:60 2:0 3:0 4:0 5:0 6:0 7:0
+            reco-tc-tsa 0:ets 1:ets 2:strict 3:strict 4:strict 5:strict 6:strict 7:strict
+            reco-prio-tc 0:0 1:1 2:0 3:0 4:0 5:0 6:0 7:0
+            """
+        )
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        ets = wrapper.get_ets()[LOCAL_ETS]
+        assert ets == {}
+
+    def test_get_app(self, wrapper):
+        stdout = "ethtype-prio 8906:3\ndscp-prio 26:3 30:3"
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        app = wrapper.get_app()[LOCAL_APP]
+        assert app[0] == {"selector": "ethtype", "priority": 3, "protocol": "8906"}
+        assert app[1] == {"selector": "dscp", "priority": 3, "protocol": "26"}
+        assert app[2] == {"selector": "dscp", "priority": 3, "protocol": "30"}
+
+    def test_get_app_repeated_protocol(self, wrapper):
+        stdout = "dscp-prio 26:3 26:5"
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        app = wrapper.get_app()[LOCAL_APP]
+        assert app == {
+            0: {"selector": "dscp", "priority": 3, "protocol": "26"},
+            1: {"selector": "dscp", "priority": 5, "protocol": "26"},
+        }
+
+    def test_get_app_mixed_port_selectors(self, wrapper):
+        stdout = dedent(
+            """\
+            stream-port-prio 1234:4
+            dgram-port-prio 5678:5
+            port-prio 9999:6
+            """
+        )
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        app = wrapper.get_app()[LOCAL_APP]
+        assert app == {
+            0: {"selector": "stream-port", "priority": 4, "protocol": "1234"},
+            1: {"selector": "dgram-port", "priority": 5, "protocol": "5678"},
+            2: {"selector": "port", "priority": 6, "protocol": "9999"},
+        }
+
+    def test_get_app_default_prio(self, wrapper):
+        stdout = "default-prio 1 3"
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        app = wrapper.get_app()[LOCAL_APP]
+        assert app == {
+            0: {"selector": "default", "priority": 1, "protocol": "0"},
+            1: {"selector": "default", "priority": 3, "protocol": "0"},
+        }
+
+    def test_get_app_all_selectors(self, wrapper):
+        stdout = dedent(
+            """\
+            ethtype-prio 8906:3
+            default-prio 2
+            dscp-prio 26:3 30:3
+            stream-port-prio 1234:4
+            dgram-port-prio 5678:5
+            port-prio 9999:6
+            pcp-prio 1:7
+            """
+        )
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        app = wrapper.get_app()[LOCAL_APP]
+        assert app == {
+            0: {"selector": "ethtype", "priority": 3, "protocol": "8906"},
+            1: {"selector": "default", "priority": 2, "protocol": "0"},
+            2: {"selector": "dscp", "priority": 3, "protocol": "26"},
+            3: {"selector": "dscp", "priority": 3, "protocol": "30"},
+            4: {"selector": "stream-port", "priority": 4, "protocol": "1234"},
+            5: {"selector": "dgram-port", "priority": 5, "protocol": "5678"},
+            6: {"selector": "port", "priority": 6, "protocol": "9999"},
+            7: {"selector": "pcp", "priority": 7, "protocol": "1"},
+        }
+
+    def test_get_app_default_numeric(self, wrapper):
+        stdout = "dscp-prio 22:3 24:3"
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        wrapper.get_app()
+        command = wrapper._connection.execute_command.call_args.args[0]
+        assert command == "dcb -N app show dev eth1"
+
+    def test_get_app_names_opt_out(self, wrapper):
+        stdout = "dscp-prio flash:3"
+        wrapper._connection.execute_command.return_value = self._out(stdout)
+        wrapper.get_app(numeric=False)
+        command = wrapper._connection.execute_command.call_args.args[0]
+        assert command == "dcb app show dev eth1"
+
+    def test_get_dcb_merges(self, wrapper):
+        ets_out = dedent(
+            """\
+            willing off ets-cap 4 cbs off
+            tc-bw 0:70 1:30 2:0 3:0 4:0 5:0 6:0 7:0
+            pg-bw 0:70 1:30 2:0 3:0 4:0 5:0 6:0 7:0
+            tc-tsa 0:ets 1:ets 2:strict 3:strict 4:strict 5:strict 6:strict 7:strict
+            prio-tc 0:0 1:0 2:0 3:1 4:0 5:0 6:0 7:0
+            reco-tc-bw 0:70 1:30 2:0 3:0 4:0 5:0 6:0 7:0
+            reco-tc-tsa 0:ets 1:ets 2:strict 3:strict 4:strict 5:strict 6:strict 7:strict
+            reco-prio-tc 0:0 1:0 2:0 3:1 4:0 5:0 6:0 7:0
+            """
+        )
+        pfc_out = dedent(
+            """\
+            pfc-cap 4 macsec-bypass off delay 1949
+            prio-pfc 0:off 1:off 2:off 3:on 4:off 5:off 6:off 7:off
+            """
+        )
+        app_out = "dscp-prio 26:3"
+        wrapper._connection.execute_command.side_effect = [
+            self._out(ets_out),
+            self._out(pfc_out),
+            self._out(app_out),
+        ]
+        result = wrapper.get_dcb()
+        assert set(result) == {LOCAL_ETS, LOCAL_PFC, LOCAL_APP}
+
+        ets = result[LOCAL_ETS]
+        assert ets["0"] == {
+            "TSA": "ETS",
+            "Bandwidth": 70,
+            "Priorities": [0, 1, 2, 4, 5, 6, 7],
+        }
+        assert ets["1"] == {"TSA": "ETS", "Bandwidth": 30, "Priorities": [3]}
+        assert ets["2"] == {"TSA": "STRICT", "Bandwidth": 0, "Priorities": []}
+
+        assert result[LOCAL_PFC] == [
+            False,
+            False,
+            False,
+            True,
+            False,
+            False,
+            False,
+            False,
+        ]
+        assert result[LOCAL_APP] == {0: {"selector": "dscp", "priority": 3, "protocol": "26"}}
+
+    def test_stderr_raises(self, wrapper):
+        wrapper._connection.execute_command.return_value = ConnectionCompletedProcess(
+            args="", stdout="", return_code=0, stderr="dcb: command failed"
+        )
+        with pytest.raises(DcbExecutionError):
+            wrapper.get_pfc()
+
+    def test_empty_output(self, wrapper):
+        wrapper._connection.execute_command.return_value = self._out("")
+        assert wrapper.get_pfc() == {LOCAL_PFC: [False] * 8}
+        assert wrapper.get_app() == {LOCAL_APP: {}}
+
+    def test_execute_dcb_command(self, wrapper):
+        command = "dcb pfc set dev eth1 prio-pfc 0:off 1:off 2:off 3:on 4:off 5:off 6:off 7:off"
+        wrapper._connection.execute_command.return_value = ConnectionCompletedProcess(
+            args="", stdout="", return_code=0, stderr=""
+        )
+        wrapper._execute_dcb_command(command)
+        wrapper._connection.execute_command.assert_called_once_with(
+            command,
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_execute_dcb_command_stderr_raises(self, wrapper):
+        command = "dcb ets set dev eth1 tc-tsa 0:ets tc-bw 0:100 prio-tc 0:0"
+        wrapper._connection.execute_command.return_value = ConnectionCompletedProcess(
+            args="",
+            stdout="",
+            return_code=0,
+            stderr="Attribute write: Unknown error -1",
+        )
+        with pytest.raises(DcbExecutionError, match="Attribute write: Unknown error -1"):
+            wrapper._execute_dcb_command(command)
+
+    def test_execute_dcb_command_process_error_propagates(self, wrapper):
+        wrapper._connection.execute_command.side_effect = DcbExecutionProcessError(
+            returncode=1, cmd="dcb ets set dev eth1 willing off", stderr="failed"
+        )
+        with pytest.raises(DcbExecutionProcessError):
+            wrapper._execute_dcb_command("dcb ets set dev eth1 willing off")
+
+    def test_format_app_map(self, wrapper):
+        assert wrapper._format_app_map({26: 3, 30: 3}) == "26:3 30:3"
+        assert wrapper._format_app_map([(26, 3), (30, 3)]) == "26:3 30:3"
+        assert wrapper._format_app_map({"0x8906": 3}) == "0x8906:3"
+
+    def test_format_app_map_empty(self, wrapper):
+        with pytest.raises(DcbExecutionError):
+            wrapper._format_app_map({})
+
+    def test_format_app_map_malformed(self, wrapper):
+        with pytest.raises(DcbExecutionError):
+            wrapper._format_app_map([(1, 2, 3)])
diff --git a/tests/unit/test_mfd_dcb/test_linux.py b/tests/unit/test_mfd_dcb/test_linux.py
index 8cba4a7..0886b76 100644
--- a/tests/unit/test_mfd_dcb/test_linux.py
+++ b/tests/unit/test_mfd_dcb/test_linux.py
@@ -12,9 +12,14 @@
 from mfd_typing import OSName
 from mfd_dcb import Dcb
 from mfd_dcb.linux import LinuxDcb
+from mfd_dcb.dcb_wrapper import DcbCli
 from mfd_ethtool import Ethtool
 import mfd_dcb.dcbtool_wrapper
-from mfd_dcb.exceptions import DcbConnectedOSNotSupported, DcbExecutionError, DcbExecutionProcessError
+from mfd_dcb.exceptions import (
+    DcbConnectedOSNotSupported,
+    DcbExecutionError,
+    DcbExecutionProcessError,
+)
 
 
 class TestLinuxDcb:
@@ -25,9 +30,13 @@ def test_pass(self):
     def dcb(self, mocker):
         conn = mocker.create_autospec(RPyCConnection)
         conn.get_os_name.return_value = OSName.LINUX
-        mocker.patch("mfd_ethtool.Ethtool.check_if_available", mocker.create_autospec(Ethtool.check_if_available))
         mocker.patch(
-            "mfd_ethtool.Ethtool.get_version", mocker.create_autospec(Ethtool.get_version, return_value="4.15")
+            "mfd_ethtool.Ethtool.check_if_available",
+            mocker.create_autospec(Ethtool.check_if_available),
+        )
+        mocker.patch(
+            "mfd_ethtool.Ethtool.get_version",
+            mocker.create_autospec(Ethtool.get_version, return_value="4.15"),
         )
         mocker.patch(
             "mfd_ethtool.Ethtool._get_tool_exec_factory",
@@ -57,7 +66,12 @@ def test_set_ets_ieee(self, dcb):
         )
         user_priority_mapping = [(0, 1), (2, 3), (4, 5), (6, 7)]
         bandwidth_per_traffic_class = [25, 25, 25, 25]
-        dcb.set_ets(user_priority_mapping, bandwidth_per_traffic_class, interface_name="eth1", mode="ieee")
+        dcb.set_ets(
+            user_priority_mapping,
+            bandwidth_per_traffic_class,
+            interface_name="eth1",
+            mode="ieee",
+        )
 
     def test_set_ets_cee(self, dcb):
         dcb._connection.execute_command.return_value = ConnectionCompletedProcess(
@@ -65,7 +79,12 @@ def test_set_ets_cee(self, dcb):
         )
         user_priority_mapping = [(0, 1), (2, 3), (4, 5), (6, 7)]
         bandwidth_per_traffic_class = [25, 25, 25, 25]
-        dcb.set_ets(user_priority_mapping, bandwidth_per_traffic_class, interface_name="eth1", mode="cee")
+        dcb.set_ets(
+            user_priority_mapping,
+            bandwidth_per_traffic_class,
+            interface_name="eth1",
+            mode="cee",
+        )
 
     def test_execute_qos_command_and_check_output(self, dcb):
         cmd_output = dedent(
@@ -224,7 +243,15 @@ def test_set_willing_with_enable_false(self, dcb):
         dcb._connection.execute_command.return_value = ConnectionCompletedProcess(
             args="", stdout="", return_code=0, stderr=""
         )
-        assert dcb.set_willing(interface_name="eth4", enable=False, mode="ieee", is_fwlldp_enabled=False) is None
+        assert (
+            dcb.set_willing(
+                interface_name="eth4",
+                enable=False,
+                mode="ieee",
+                is_fwlldp_enabled=False,
+            )
+            is None
+        )
 
     def test_set_willing_with_cee(self, dcb):
         dcb._connection.execute_command.return_value = ConnectionCompletedProcess(
@@ -402,13 +429,19 @@ def test__get_pfc_counter_dict_invalid_input(self, dcb):
             args="", stdout="", return_code=0, stderr=""
         )
         with pytest.raises(
-            DcbExecutionError, match=re.escape("No match found for tx_pb_0_pxon for _get_pfc_counter_dict")
+            DcbExecutionError,
+            match=re.escape("No match found for tx_pb_0_pxon for _get_pfc_counter_dict"),
         ):
             dcb._get_pfc_counter_dict(
                 out="",
                 pfc_counters={0: {}},
                 priority=0,
-                frame_list=["tx_pb_0_pxon", "tx_pb_0_pxoff", "rx_pb_0_pxon", "rx_pb_0_pxoff"],
+                frame_list=[
+                    "tx_pb_0_pxon",
+                    "tx_pb_0_pxoff",
+                    "rx_pb_0_pxon",
+                    "rx_pb_0_pxoff",
+                ],
             )
 
     def test_remove_lldpad_conf(self, dcb):
@@ -515,7 +548,8 @@ def test_get_ets_dcbtool(self, dcb, mocker):
         mocker.patch(
             "mfd_dcb.linux.LinuxDcb._get_tool",
             mocker.create_autospec(
-                LinuxDcb._get_tool, return_value=mfd_dcb.dcbtool_wrapper.DcbTool("eth4", dcb._connection)
+                LinuxDcb._get_tool,
+                return_value=mfd_dcb.dcbtool_wrapper.DcbTool("eth4", dcb._connection),
             ),
         )
         dcb._connection.execute_command.return_value = ConnectionCompletedProcess(
@@ -530,7 +564,8 @@ def test_get_app_dcbtool(self, dcb, mocker):
         mocker.patch(
             "mfd_dcb.linux.LinuxDcb._get_tool",
             mocker.create_autospec(
-                LinuxDcb._get_tool, return_value=mfd_dcb.dcbtool_wrapper.DcbTool("eth4", dcb._connection)
+                LinuxDcb._get_tool,
+                return_value=mfd_dcb.dcbtool_wrapper.DcbTool("eth4", dcb._connection),
             ),
         )
         with pytest.raises(DcbExecutionError, match=re.escape("get_app not implemented in dcbtool")):
@@ -567,7 +602,8 @@ def test_get_dcb_dcbtool(self, dcb, mocker):
         mocker.patch(
             "mfd_dcb.linux.LinuxDcb._get_tool",
             mocker.create_autospec(
-                LinuxDcb._get_tool, return_value=mfd_dcb.dcbtool_wrapper.DcbTool("eth4", dcb._connection)
+                LinuxDcb._get_tool,
+                return_value=mfd_dcb.dcbtool_wrapper.DcbTool("eth4", dcb._connection),
             ),
         )
         dcb._connection.execute_command.return_value = ConnectionCompletedProcess(
@@ -598,7 +634,8 @@ def test_get_pfc_dcbtool(self, dcb, mocker):
         mocker.patch(
             "mfd_dcb.linux.LinuxDcb._get_tool",
             mocker.create_autospec(
-                LinuxDcb._get_tool, return_value=mfd_dcb.dcbtool_wrapper.DcbTool("eth4", dcb._connection)
+                LinuxDcb._get_tool,
+                return_value=mfd_dcb.dcbtool_wrapper.DcbTool("eth4", dcb._connection),
             ),
         )
         dcb._connection.execute_command.return_value = ConnectionCompletedProcess(
@@ -712,7 +749,16 @@ def test_verify_ets_negative(self, dcb, mocker):
     def test_verify_pfc(self, dcb, mocker):
         interface_name_mock = "eth0"
         dcb_map_mock = {"LOCAL_PFC": [False, False, False, True, True, False, False, False]}
-        pfc_config_mock = {0: False, 1: False, 2: False, 3: True, 4: True, 5: False, 6: False, 7: False}
+        pfc_config_mock = {
+            0: False,
+            1: False,
+            2: False,
+            3: True,
+            4: True,
+            5: False,
+            6: False,
+            7: False,
+        }
         # Call the verify_pfc method
         assert dcb.verify_pfc(interface_name_mock, dcb_map_mock, pfc_config_mock)
 
@@ -765,3 +811,146 @@ def test_verify_app_negative(self, dcb, mocker):
         dcb.get_app.return_value = {"LOCAL_APP": {"3260": {"Priority": 4, "Protocol": "TCP"}}}
         return_value = dcb.verify_app(interface_name_mock, dcb_map_mock)
         assert not return_value
+
+    def _ok(self):
+        return ConnectionCompletedProcess(args="", stdout="", return_code=0, stderr="")
+
+    def test_set_pfc_dcb_tool(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.set_pfc(
+            "eth1",
+            [False, False, False, True, False, False, False, False],
+            tool_name="dcb",
+        )
+        dcb._connection.execute_command.assert_any_call(
+            "dcb pfc set dev eth1 prio-pfc 0:off 1:off 2:off 3:on 4:off 5:off 6:off 7:off",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_set_pfc_lldptool_unchanged(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.set_pfc("eth1", [False, False, False, True, False, False, False, False])
+        dcb._connection.execute_command.assert_any_call(
+            "lldptool -Ti eth1 -V PFC enabled=3 enableTx=yes willing=no",
+            expected_return_codes={},
+        )
+
+    def test_set_ets_dcb_tool(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.set_ets([(0, 1, 2), (3,)], [30, 70], interface_name="eth1", tool_name="dcb")
+        dcb._connection.execute_command.assert_any_call(
+            "dcb ets set dev eth1 willing off tc-tsa 0:ets 1:ets tc-bw 0:30 1:70 prio-tc 0:0 1:0 2:0 3:1",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_set_ets_unsupported_tool_name(self, dcb):
+        with pytest.raises(DcbExecutionError):
+            dcb.set_ets([(0,)], [100], interface_name="eth1", tool_name="dcbb")
+
+    def test_set_pfc_unsupported_tool_name(self, dcb):
+        with pytest.raises(DcbExecutionError):
+            dcb.set_pfc("eth1", [False] * 8, tool_name="dcbnl")
+
+    def test_get_ets_unsupported_tool_name(self, dcb):
+        with pytest.raises(DcbExecutionError):
+            dcb.get_ets(interface_name="eth1", tool_name="dcbnll")
+
+    def test_set_app_dcb_tool(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.set_app("eth1", {26: 3})
+        dcb._connection.execute_command.assert_called_once_with(
+            "dcb app add dev eth1 dscp-prio 26:3",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_set_app_dcb_tool_multiple(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.set_app("eth1", [(26, 3), (30, 3)])
+        dcb._connection.execute_command.assert_any_call(
+            "dcb app add dev eth1 dscp-prio 26:3 30:3",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_set_app_dcb_tool_custom_selector(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.set_app("eth1", {"0x8906": 3}, selector="ethtype-prio")
+        dcb._connection.execute_command.assert_called_once_with(
+            "dcb app add dev eth1 ethtype-prio 0x8906:3",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_del_app_dcb_tool(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.del_app("eth1", {26: 3})
+        dcb._connection.execute_command.assert_any_call(
+            "dcb app del dev eth1 dscp-prio 26:3",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_del_app_dcb_tool_custom_selector(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.del_app("eth1", {3260: 3}, selector="port-prio")
+        dcb._connection.execute_command.assert_called_once_with(
+            "dcb app del dev eth1 port-prio 3260:3",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_replace_app_dcb_tool(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.replace_app("eth1", {26: 5})
+        dcb._connection.execute_command.assert_any_call(
+            "dcb app replace dev eth1 dscp-prio 26:5",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_replace_app_dcb_tool_custom_selector(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.replace_app("eth1", {"0x8906": 5}, selector="ethtype-prio")
+        dcb._connection.execute_command.assert_called_once_with(
+            "dcb app replace dev eth1 ethtype-prio 0x8906:5",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_flush_app_dcb_tool(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.flush_app("eth1")
+        dcb._connection.execute_command.assert_any_call(
+            "dcb app flush dev eth1 dscp-prio",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_flush_app_all_selectors(self, dcb):
+        dcb._connection.execute_command.return_value = self._ok()
+        dcb.flush_app("eth1", selector=None)
+        dcb._connection.execute_command.assert_any_call(
+            "dcb app flush dev eth1",
+            custom_exception=DcbExecutionProcessError,
+            expected_return_codes=[0],
+        )
+
+    def test_app_methods_require_dcb_tool(self, dcb):
+        with pytest.raises(DcbExecutionError):
+            dcb.set_app("eth1", {26: 3}, tool_name="dcbnl")
+        with pytest.raises(DcbExecutionError):
+            dcb.replace_app("eth1", {26: 3}, tool_name="dcbnl")
+        with pytest.raises(DcbExecutionError):
+            dcb.del_app("eth1", {26: 3}, tool_name="dcbnl")
+        with pytest.raises(DcbExecutionError):
+            dcb.flush_app("eth1", tool_name="dcbnl")
+
+    def test_get_tool_returns_dcb_cli(self, dcb):
+        assert isinstance(dcb._get_tool("eth1", tool_name="dcb"), DcbCli)
+
+    def test_set_app_empty_map(self, dcb):
+        with pytest.raises(DcbExecutionError):
+            dcb.set_app("eth1", {})