diff --git a/ingestion/src/metadata/data_quality/source/test_suite.py b/ingestion/src/metadata/data_quality/source/test_suite.py
index b4822f8e53d..2790d83fb48 100644
--- a/ingestion/src/metadata/data_quality/source/test_suite.py
+++ b/ingestion/src/metadata/data_quality/source/test_suite.py
@@ -89,7 +89,10 @@ class TestSuiteSource(Source):
                 params={"testSuiteId": test_suite.id.root},
             )
             test_cases = cast(List[TestCase], test_cases)  # satisfy type checker
-
+            if self.source_config.testCases is not None:
+                test_cases = [
+                    t for t in test_cases if t.name in self.source_config.testCases
+                ]
             return test_cases
 
         return None
diff --git a/ingestion/tests/unit/data_quality/source/__init__.py b/ingestion/tests/unit/data_quality/source/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/ingestion/tests/unit/data_quality/source/test_test_suite.py b/ingestion/tests/unit/data_quality/source/test_test_suite.py
new file mode 100644
index 00000000000..e02e5981321
--- /dev/null
+++ b/ingestion/tests/unit/data_quality/source/test_test_suite.py
@@ -0,0 +1,96 @@
+from unittest.mock import Mock
+from uuid import UUID
+
+import pytest
+
+from metadata.data_quality.source.test_suite import TestSuiteSource
+from metadata.generated.schema.entity.data.table import Table
+from metadata.generated.schema.metadataIngestion.workflow import (
+    OpenMetadataWorkflowConfig,
+)
+from metadata.generated.schema.tests.testCase import TestCase
+from metadata.generated.schema.tests.testSuite import TestSuite
+from metadata.generated.schema.type.entityReference import EntityReference
+from metadata.ingestion.ometa.ometa_api import OpenMetadata
+
+MOCK_ENTITY_REFERENCE = EntityReference(
+    id=str(UUID(int=0)), type="test_suite", name="test_suite"
+)
+
+
+@pytest.mark.parametrize(
+    "parameters,expected",
+    [
+        (
+            {
+                "type": "TestSuite",
+                "entityFullyQualifiedName": "MyTestSuite",
+            },
+            ["test_case1", "test_case2"],
+        ),
+        (
+            {
+                "type": "TestSuite",
+                "entityFullyQualifiedName": "MyTestSuite",
+                "testCases": [
+                    "test_case1",
+                ],
+            },
+            ["test_case1"],
+        ),
+    ],
+)
+def test_source_config(parameters, expected, monkeypatch):
+    workflow_config = {
+        "source": {
+            "type": "TestSuite",
+            "serviceName": "MyTestSuite",
+            "sourceConfig": {"config": parameters},
+            "serviceConnection": {
+                "config": {
+                    "type": "Mysql",
+                    "hostPort": "localhost:3306",
+                    "username": "root",
+                }
+            },
+        },
+        "workflowConfig": {
+            "openMetadataServerConfig": {
+                "hostPort": "localhost:8585",
+            }
+        },
+    }
+    monkeypatch.setattr(TestSuiteSource, "test_connection", Mock())
+
+    mock_metadata = Mock(spec=OpenMetadata)
+    mock_metadata.get_by_name.return_value = Table(
+        id=UUID(int=0),
+        name="test_table",
+        columns=[],
+        testSuite=MOCK_ENTITY_REFERENCE,
+    )
+    mock_metadata.list_all_entities.return_value = [
+        TestCase(
+            name="test_case1",
+            id=UUID(int=0),
+            testDefinition=MOCK_ENTITY_REFERENCE,
+            testSuite=MOCK_ENTITY_REFERENCE,
+            entityLink="<#E::some::link>",
+        ),
+        TestCase(
+            name="test_case2",
+            id=UUID(int=0),
+            testDefinition=MOCK_ENTITY_REFERENCE,
+            testSuite=MOCK_ENTITY_REFERENCE,
+            entityLink="<#E::some::link>",
+        ),
+    ]
+    mock_metadata.get_by_id.return_value = TestSuite(
+        name="test_suite", executable=True, id=UUID(int=0)
+    )
+
+    source = TestSuiteSource(
+        OpenMetadataWorkflowConfig.parse_obj(workflow_config), mock_metadata
+    )
+    test_cases = list(source._iter())[0].right.test_cases
+    assert [t.name.root for t in test_cases] == expected
diff --git a/openmetadata-docs/content/partials/v1.4/connectors/yaml/data-quality.md b/openmetadata-docs/content/partials/v1.4/connectors/yaml/data-quality.md
index d04d64130cc..2494c53be64 100644
--- a/openmetadata-docs/content/partials/v1.4/connectors/yaml/data-quality.md
+++ b/openmetadata-docs/content/partials/v1.4/connectors/yaml/data-quality.md
@@ -78,6 +78,7 @@ source:
     config:
       type: TestSuite
       entityFullyQualifiedName: MySQL.default.openmetadata_db.tag_usage
+#     testCases: ["run_only_this_test_case"] # Optional, if not provided all tests will be executed
 
 processor:
   type: "orm-test-runner"
diff --git a/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/testSuitePipeline.json b/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/testSuitePipeline.json
index 1927aa7b092..e112a4f229a 100644
--- a/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/testSuitePipeline.json
+++ b/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/testSuitePipeline.json
@@ -31,7 +31,14 @@
     "profileSampleType": {
       "$ref": "../entity/data/table.json#/definitions/profileSampleType",
       "title": "Profile Sample Type"
-
+    },
+    "testCases": {
+      "description": "List of test cases to be executed on the entity. If null, all test cases will be executed.",
+      "type": "array",
+      "items": {
+        "$ref": "../type/basic.json#/definitions/testCaseEntityName"
+      },
+      "default": null
     }
   },
   "required": ["type", "entityFullyQualifiedName"],