diff --git a/ansible/testbed-cli.sh b/ansible/testbed-cli.sh
index 73863536de2..448d2721ca2 100755
--- a/ansible/testbed-cli.sh
+++ b/ansible/testbed-cli.sh
@@ -165,6 +165,8 @@ function read_yaml
   dut=${line_arr[11]}
   duts=$(python -c "from __future__ import print_function; print(','.join(eval(\"$dut\")))")
   inv_name=${line_arr[12]}
+  # Remove the dpu duts by the keyword 'dpu' in the dut name
+  duts=$(echo $duts | sed "s/,[^,]*dpu[^,]*//g")
 }
 
 function read_file
@@ -283,7 +285,7 @@ function add_topo
 
   cache_files_path_value=$(is_cache_exist)
   if [[ -n $cache_files_path_value ]]; then
-    echo "$testbed_name" > $cache_files_path_value/$dut
+    echo "$testbed_name" > $cache_files_path_value/$duts
   fi
 
   echo Done
@@ -694,7 +696,7 @@ function deploy_topo_with_cache
   fi
 
   read_file ${testbed_name}
-  setup_name=$dut
+  setup_name=$duts
   if [[ "$setup_name" == "" ]]; then
       echo "No such testbed: $testbed_name, exiting..."
       exit
diff --git a/tests/conftest.py b/tests/conftest.py
index 5df8e812e37..4352dc1cd20 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -233,6 +233,11 @@ def pytest_addoption(parser):
     parser.addoption("--is_parallel_leader", action="store_true", default=False, help="Is the parallel leader")
     parser.addoption("--parallel_followers", action="store", default=0, type=int, help="Number of parallel followers")
 
+    ############################
+    #   SmartSwitch options    #
+    ############################
+    parser.addoption("--dpu-pattern", action="store", default="all", help="dpu host name")
+
 
 def pytest_configure(config):
     if config.getoption("enable_macsec"):
@@ -366,10 +371,9 @@ def parallel_run_context(request):
     )
 
 
-def get_specified_duts(request):
+def get_specified_device_info(request, device_pattern):
     """
-    Get a list of DUT hostnames specified with the --host-pattern CLI option
-    or -d if using `run_tests.sh`
+    Get a list of device hostnames specified with the --host-pattern or --dpu-pattern CLI option
     """
     tbname, tbinfo = get_tbinfo(request)
     testbed_duts = tbinfo['duts']
@@ -377,8 +381,11 @@ def get_specified_duts(request):
     if is_parallel_run(request):
         return [get_target_hostname(request)]
 
-    host_pattern = request.config.getoption("--host-pattern")
+    host_pattern = request.config.getoption(device_pattern)
     if host_pattern == 'all':
+        if device_pattern == '--dpu-pattern':
+            testbed_duts = [dut for dut in testbed_duts if 'dpu' in dut]
+            logger.info(f"dpu duts: {testbed_duts}")
         return testbed_duts
     else:
         specified_duts = get_duts_from_host_pattern(host_pattern)
@@ -394,6 +401,21 @@ def get_specified_duts(request):
     return duts
 
 
+def get_specified_duts(request):
+    """
+    Get a list of DUT hostnames specified with the --host-pattern CLI option
+    or -d if using `run_tests.sh`
+    """
+    return get_specified_device_info(request, "--host-pattern")
+
+
+def get_specified_dpus(request):
+    """
+    Get a list of DUT hostnames specified with the --dpu-pattern CLI option
+    """
+    return get_specified_device_info(request, "--dpu-pattern")
+
+
 def pytest_sessionstart(session):
     # reset all the sonic_custom_msg keys from cache
     # reset here because this fixture will always be very first fixture to be called
@@ -451,6 +473,48 @@ def duthost(duthosts, request):
     return duthost
 
 
+@pytest.fixture(name="dpuhosts", scope="session")
+def fixture_dpuhosts(enhance_inventory, ansible_adhoc, tbinfo, request):
+    """
+    @summary: fixture to get DPU hosts defined in testbed.
+    @param ansible_adhoc: Fixture provided by the pytest-ansible package.
+        Source of the various device objects. It is
+        mandatory argument for the class constructors.
+    @param tbinfo: fixture provides information about testbed.
+    """
+    # Before calling dpuhosts, we must enable NAT on NPU.
+    # E.g. run sonic-dpu-mgmt-traffic.sh on NPU to enable NAT
+    # sonic-dpu-mgmt-traffic.sh inbound -e --dpus all --ports 5021,5022,5023,5024
+    try:
+        host = DutHosts(ansible_adhoc, tbinfo, request, get_specified_dpus(request),
+                        target_hostname=get_target_hostname(request), is_parallel_leader=is_parallel_leader(request))
+        return host
+    except BaseException as e:
+        logger.error("Failed to initialize dpuhosts.")
+        request.config.cache.set("dpuhosts_fixture_failed", True)
+        pt_assert(False, "!!!!!!!!!!!!!!!! dpuhosts fixture failed !!!!!!!!!!!!!!!!"
+                  "Exception: {}".format(repr(e)))
+
+
+@pytest.fixture(scope="session")
+def dpuhost(dpuhosts, request):
+    '''
+    @summary: Shortcut fixture for getting DPU host. For a lengthy test case, test case module can
+              pass a request to disable sh time out mechanis on dut in order to avoid ssh timeout.
+              After test case completes, the fixture will restore ssh timeout.
+    @param duthosts: fixture to get DPU hosts
+    @param request: request parameters for duphost test fixture
+    '''
+    dpu_index = getattr(request.session, "dpu_index", 0)
+    assert dpu_index < len(dpuhosts), \
+        "DPU index '{0}' is out of bound '{1}'".format(dpu_index,
+                                                       len(dpuhosts))
+
+    duthost = dpuhosts[dpu_index]
+
+    return duthost
+
+
 @pytest.fixture(scope="session")
 def mg_facts(duthost):
     return duthost.minigraph_facts(host=duthost.hostname)['ansible_facts']