feat: cover data quality APIs

Author: ayush-shah

platform-api/src/openmetadata_demo_api/catalog.py

@@ -72,6 +72,18 @@
     "kpi/KpiResource.java",
 }
 GOVERNANCE_LEAF_FILES = {"governance/WorkflowInstanceStateResource.java"}
+DATA_QUALITY_CONNECTION_FILE = "services/connections/TestConnectionDefinitionResource.java"
+DATA_QUALITY_PREREQUISITES = {
+    "createDatabaseService",
+    "createDatabase",
+    "createDBSchema",
+    "createTable",
+}
+DATA_QUALITY_PHASES = {
+    "dqtests/TestDefinitionResource.java": "parent-assets",
+    "dqtests/TestSuiteResource.java": "child-assets",
+    "dqtests/TestCaseResource.java": "enrichment",
+}
 CORE_SERVICE_PREREQUISITES = {
     "createApiService",
     "createDashboardService",
@@ -249,6 +261,8 @@ def _phase(operation: Mapping[str, Any]) -> str:
         return "service-management"
     if source_file == "types/TypeResource.java":
         return "child-assets" if operation["operation_id"] == "addProperty" else "parent-assets"
+    if source_file.startswith("dqtests/"):
+        return DATA_QUALITY_PHASES.get(str(source_file), "platform")
     operation_text = " ".join(
         (
             str(operation["operation_id"]),
@@ -394,6 +408,17 @@ def scenarios() -> dict[str, Scenario]:
                 or operation["operation_id"] == "createDashboardService"
             ),
         ),
+        "data-quality": scenario_from_operations(
+            "data-quality",
+            "Data quality definitions, suites, cases, results, resolutions, and connection schemas",
+            tuple(
+                operation
+                for operation in all_operations
+                if str(operation["source"]["file"]).startswith("dqtests/")
+                or operation["source"]["file"] == DATA_QUALITY_CONNECTION_FILE
+                or operation["operation_id"] in DATA_QUALITY_PREREQUISITES
+            ),
+        ),
     }
 
 

platform-api/src/openmetadata_demo_api/request_fixtures.py

@@ -76,6 +76,15 @@
             "contractExecutionStatus": "Success",
             "result": "Dry-run validation passed",
         },
+        "DataQualityReportBatchRequest": {
+            "requests": [
+                {
+                    "key": "failed-tests",
+                    "index": "test_case_search_index",
+                    "aggregationQuery": ('{"failed":{"terms":{"field":"testCaseStatus.keyword"}}}'),
+                }
+            ]
+        },
         "DatabaseProfilerConfig": {"sampleDataCount": 10, "randomizedSample": True},
         "DatabaseSchemaProfilerConfig": {
             "sampleDataCount": 10,
@@ -245,6 +254,10 @@
             "metadata.generated.schema.entity.datacontract.dataContractResult",
             "DataContractResult",
         ),
+        "DataQualityReportBatchRequest": (
+            "metadata.generated.schema.tests.dataQualityReportBatchRequest",
+            "DataQualityReportBatchRequest",
+        ),
         "DatabaseProfilerConfig": (
             "metadata.generated.schema.entity.data.database",
             "DatabaseProfilerConfig",

platform-api/tests/test_runtime.py

@@ -875,3 +875,64 @@ def test_governance_insight_requests_respect_recorded_validation_boundaries() ->
         assert assets_by_model[model_name]["validationNote"].startswith(
             "RC1 generated-model defect:"
         )
+
+
+def test_data_quality_scenario_covers_source_routes_and_dependencies() -> None:
+    connection_file = "services/connections/TestConnectionDefinitionResource.java"
+    prerequisites = {
+        "createDatabaseService",
+        "createDatabase",
+        "createDBSchema",
+        "createTable",
+    }
+    expected = {
+        operation["operation_id"]
+        for operation in load_operations()
+        if str(operation["source"]["file"]).startswith("dqtests/")
+        or operation["source"]["file"] == connection_file
+    }
+    scenario = scenarios()["data-quality"]
+
+    assert {step.operation_id for step in scenario.operation_steps} == expected | prerequisites
+    assert len(scenario.run()) == len(expected | prerequisites)
+    assert {
+        "createTestDefinition",
+        "createBasicTestSuite",
+        "createTestCase",
+        "addTestCaseResult",
+        "createTestCaseResolutionStatus",
+        "getDataQualityReportBatch",
+        "listTestConnectionDefinitions",
+    } <= expected
+    ordered = [step.operation_id for step in scenario.ordered_steps()]
+    for parent, child in (
+        ("createTable", "createTestCase"),
+        ("createTestDefinition", "createTestCase"),
+        ("createBasicTestSuite", "createTestCase"),
+        ("createTestCase", "addTestCaseResult"),
+        ("createTestCase", "createTestCaseResolutionStatus"),
+    ):
+        assert ordered.index(parent) < ordered.index(child)
+
+
+def test_data_quality_requests_are_validated_and_placeholder_free() -> None:
+    scenario = scenarios()["data-quality"]
+
+    assert not {
+        str(value["_model"])
+        for step in scenario.operation_steps
+        for value in _nested_mappings(step.request)
+        if "_model" in value
+    }
+    models = {
+        str(asset["model"]).rsplit(".", 1)[-1]
+        for asset in load_assets()
+        if "data-quality" in asset["scenarios"]
+    }
+    assert {
+        "CreateTestDefinitionRequest",
+        "CreateTestSuiteRequest",
+        "CreateTestCaseRequest",
+        "CreateTestCaseResult",
+        "CreateTestCaseResolutionStatus",
+    } <= models