feat: cover lineage query profile and usage APIs

Author: ayush-shah

platform-api/src/openmetadata_demo_api/catalog.py

@@ -84,6 +84,23 @@
     "dqtests/TestSuiteResource.java": "child-assets",
     "dqtests/TestCaseResource.java": "enrichment",
 }
+LINEAGE_QUERY_PREFIXES = ("lineage/", "query/", "entityProfiles/", "usage/")
+PROFILE_SAMPLE_PREFIXES = ("databases/", "drives/", "searchindex/", "storages/", "topics/")
+LINEAGE_QUERY_PREREQUISITES = {
+    "createDatabaseService",
+    "createDatabase",
+    "createDBSchema",
+    "createTable",
+    "createDriveService",
+    "createDirectory",
+    "createFile",
+    "createMessagingService",
+    "createTopic",
+    "createSearchService",
+    "createSearchIndex",
+    "createStorageService",
+    "createContainer",
+}
 CORE_SERVICE_PREREQUISITES = {
     "createApiService",
     "createDashboardService",
@@ -255,6 +272,21 @@ def _is_bulk_drive_operation(operation: Mapping[str, Any]) -> bool:
     )
 
 
+def _is_lineage_query_profile_usage_operation(operation: Mapping[str, Any]) -> bool:
+    source_file = str(operation["source"]["file"])
+    text = " ".join(
+        (str(operation["operation_id"]), str(operation["java_method"]), str(operation["path"]))
+    ).lower()
+    return (
+        source_file.startswith(LINEAGE_QUERY_PREFIXES)
+        or (
+            source_file.startswith(PROFILE_SAMPLE_PREFIXES)
+            and any(term in text for term in ("sample", "profile"))
+        )
+        or operation["operation_id"] in LINEAGE_QUERY_PREREQUISITES
+    )
+
+
 def _phase(operation: Mapping[str, Any]) -> str:
     source_file = operation["source"]["file"]
     if source_file.startswith("services/"):
@@ -419,6 +451,15 @@ def scenarios() -> dict[str, Scenario]:
                 or operation["operation_id"] in DATA_QUALITY_PREREQUISITES
             ),
         ),
+        "lineage-query-profile-usage": scenario_from_operations(
+            "lineage-query-profile-usage",
+            "Lineage, queries, profiles, usage, sample data, and deterministic replay",
+            tuple(
+                operation
+                for operation in all_operations
+                if _is_lineage_query_profile_usage_operation(operation)
+            ),
+        ),
     }
 
 

platform-api/src/openmetadata_demo_api/request_fixtures.py

@@ -10,15 +10,45 @@
 from typing import Any
 
 DEMO_ID = "00000000-0000-0000-0000-000000000001"
+DEMO_ID_2 = "00000000-0000-0000-0000-000000000002"
 DEMO_ENTITY_REFERENCE = {
     "id": DEMO_ID,
     "type": "table",
     "name": "demo",
     "fullyQualifiedName": "demo.demo.demo.demo",
 }
+OPENLINEAGE_EVENT = {
+    "eventTime": "2026-01-01T00:00:00Z",
+    "producer": "https://open-metadata.org/demo",
+    "schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json",
+    "eventType": "COMPLETE",
+    "run": {"runId": DEMO_ID},
+    "job": {"namespace": "demo", "name": "daily_orders"},
+    "inputs": [],
+    "outputs": [],
+}
 
 REQUEST_FIXTURES = MappingProxyType(
     {
+        "AddLineage": {
+            "edge": {
+                "fromEntity": {
+                    **DEMO_ENTITY_REFERENCE,
+                    "name": "source",
+                    "fullyQualifiedName": "demo.demo.demo.source",
+                },
+                "toEntity": {
+                    **DEMO_ENTITY_REFERENCE,
+                    "id": DEMO_ID_2,
+                    "name": "target",
+                    "fullyQualifiedName": "demo.demo.demo.target",
+                },
+                "lineageDetails": {
+                    "source": "Manual",
+                    "description": "Dry-run lineage",
+                },
+            }
+        },
         "AIGovernanceBulkTriageRequest": {
             "action": "register",
             "reason": "Dry-run review",
@@ -85,6 +115,7 @@
                 }
             ]
         },
+        "DailyCount": {"count": 1, "date": "2026-01-01"},
         "DatabaseProfilerConfig": {"sampleDataCount": 10, "randomizedSample": True},
         "DatabaseSchemaProfilerConfig": {
             "sampleDataCount": 10,
@@ -100,6 +131,11 @@
                 "value": "Updated by the dry-run demo",
             }
         ],
+        "HydrateLineageRequest": {
+            "entities": [DEMO_ENTITY_REFERENCE],
+            "fields": "tags,owners",
+        },
+        "LineageDetails": {"source": "Manual", "description": "Dry-run lineage"},
         "CustomProperty": {
             "name": "demoProperty",
             "description": "Dry-run custom property",
@@ -137,6 +173,8 @@
                 }
             ],
         },
+        "OpenLineageBatchRequest": {"events": [OPENLINEAGE_EVENT]},
+        "OpenLineageRunEvent": OPENLINEAGE_EVENT,
         "PipelineObservability": {
             "pipeline": {
                 **DEMO_ENTITY_REFERENCE,
@@ -220,6 +258,10 @@
 
 PYDANTIC_REQUEST_MODELS = MappingProxyType(
     {
+        "AddLineage": (
+            "metadata.generated.schema.api.lineage.addLineage",
+            "AddLineageRequest",
+        ),
         "AIGovernanceBulkTriageRequest": (
             "metadata.generated.schema.api.ai.aiGovernanceBulkTriageRequest",
             "AIGovernanceBulkTriageRequest",
@@ -258,6 +300,10 @@
             "metadata.generated.schema.tests.dataQualityReportBatchRequest",
             "DataQualityReportBatchRequest",
         ),
+        "DailyCount": (
+            "metadata.generated.schema.type.dailyCount",
+            "DailyCountOfSomeMeasurement",
+        ),
         "DatabaseProfilerConfig": (
             "metadata.generated.schema.entity.data.database",
             "DatabaseProfilerConfig",
@@ -270,6 +316,14 @@
             "metadata.generated.schema.email.emailTemplate",
             "EmailTemplate",
         ),
+        "HydrateLineageRequest": (
+            "metadata.generated.schema.api.lineage.hydrateLineageRequest",
+            "HydrateLineageRequest",
+        ),
+        "LineageDetails": (
+            "metadata.generated.schema.type.entityLineage",
+            "LineageDetails",
+        ),
         "EntityReference": (
             "metadata.generated.schema.type.entityReference",
             "EntityReference",
@@ -290,6 +344,14 @@
             "metadata.generated.schema.entity.services.ingestionPipelines.operationMetrics",
             "OperationMetricsBatch",
         ),
+        "OpenLineageBatchRequest": (
+            "metadata.generated.schema.api.lineage.openlineage.openLineageBatchRequest",
+            "OpenLineageBatchRequest",
+        ),
+        "OpenLineageRunEvent": (
+            "metadata.generated.schema.api.lineage.openlineage.openLineageRunEvent",
+            "OpenLineageRunEvent",
+        ),
         "PipelineObservability": (
             "metadata.generated.schema.type.pipelineObservability",
             "PipelineObservability",

platform-api/tests/test_runtime.py

@@ -936,3 +936,82 @@ def test_data_quality_requests_are_validated_and_placeholder_free() -> None:
         "CreateTestCaseResult",
         "CreateTestCaseResolutionStatus",
     } <= models
+
+
+def test_lineage_query_profile_usage_scenario_covers_source_and_dependent_routes() -> None:
+    prefixes = ("lineage/", "query/", "entityProfiles/", "usage/")
+    prerequisites = {
+        "createDatabaseService",
+        "createDatabase",
+        "createDBSchema",
+        "createTable",
+        "createDriveService",
+        "createDirectory",
+        "createFile",
+        "createMessagingService",
+        "createTopic",
+        "createSearchService",
+        "createSearchIndex",
+        "createStorageService",
+        "createContainer",
+    }
+
+    def selected(operation) -> bool:
+        source_file = str(operation["source"]["file"])
+        text = " ".join(
+            (str(operation["operation_id"]), str(operation["java_method"]), str(operation["path"]))
+        ).lower()
+        return source_file.startswith(prefixes) or (
+            source_file.startswith(
+                ("databases/", "drives/", "searchindex/", "storages/", "topics/")
+            )
+            and any(term in text for term in ("sample", "profile"))
+        )
+
+    expected = {operation["operation_id"] for operation in load_operations() if selected(operation)}
+    scenario = scenarios()["lineage-query-profile-usage"]
+
+    assert {step.operation_id for step in scenario.operation_steps} == expected | prerequisites
+    assert len(scenario.run()) == len(expected | prerequisites)
+    assert {
+        "addLineageEdge",
+        "addLineageEdgeByName",
+        "postOpenLineageEvent",
+        "createQuery",
+        "addProfileDataById",
+        "reportEntityUsageWithID",
+        "addSampleData",
+        "addDataProfiler",
+    } <= expected
+    ordered = [step.operation_id for step in scenario.ordered_steps()]
+    for dependent in (
+        "addLineageEdge",
+        "addLineageEdgeByName",
+        "reportEntityUsageWithID",
+        "addSampleData",
+        "addDataProfiler",
+    ):
+        assert ordered.index("createTable") < ordered.index(dependent)
+    assert scenario.run() == scenario.run()
+
+
+def test_lineage_query_profile_usage_requests_are_validated_and_placeholder_free() -> None:
+    scenario = scenarios()["lineage-query-profile-usage"]
+
+    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 "lineage-query-profile-usage" in asset["scenarios"]
+    }
+    assert {
+        "CreateEntityProfileRequest",
+        "CreateQueryRequest",
+        "CreateQueryCostRecordRequest",
+        "CreateTableProfileRequest",
+    } <= models