diff --git a/ingestion/pyproject.toml b/ingestion/pyproject.toml index f30a40158e9a..2d3f2a72a4cd 100644 --- a/ingestion/pyproject.toml +++ b/ingestion/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta" # since it helps us organize and isolate version management [project] name = "openmetadata-ingestion" -version = "1.12.11.0" +version = "1.12.11.0+superset.chart.bridge.3" dynamic = ["readme", "dependencies", "optional-dependencies"] authors = [ { name = "OpenMetadata Committers" } diff --git a/ingestion/src/metadata/ingestion/lineage/sql_lineage.py b/ingestion/src/metadata/ingestion/lineage/sql_lineage.py index 011d8f7cb4c8..67607302dd98 100644 --- a/ingestion/src/metadata/ingestion/lineage/sql_lineage.py +++ b/ingestion/src/metadata/ingestion/lineage/sql_lineage.py @@ -608,6 +608,38 @@ def get_table_entities_from_query( if table_entities: return table_entities + # Cross-catalog fallback: the database resolved from the query/context may not + # match the database the source table was ingested under. This happens when the + # same physical data is exposed through different catalogs across services, so a + # view in one catalog references a table that another service ingested under a + # different database name. Searching with the view's database never matches the + # table under a different database name, so no lineage edge is created even with + # crossDatabaseServiceNames configured. Retry with database=None so the FQN search + # wildcards the database (`.*..`) and resolves the table + # across the configured cross-database services. This is scoped to `service_names`, + # which only holds the current service plus the explicitly allowed + # crossDatabaseServiceNames, so it cannot reach unrelated services. + table_entities = search_table_entities( + metadata=metadata, + service_names=service_names, + database=None, + database_schema=schema_query if schema_query else database_schema, + table=table, + ) + if table_entities: + return table_entities + + if schema_fallback: + table_entities = search_table_entities( + metadata=metadata, + service_names=service_names, + database=None, + database_schema=None, + table=table, + ) + if table_entities: + return table_entities + return None diff --git a/ingestion/src/metadata/ingestion/sink/metadata_rest.py b/ingestion/src/metadata/ingestion/sink/metadata_rest.py index 3be1d9dce23b..b4628f7e5f81 100644 --- a/ingestion/src/metadata/ingestion/sink/metadata_rest.py +++ b/ingestion/src/metadata/ingestion/sink/metadata_rest.py @@ -453,7 +453,9 @@ def _flush_query_buffer(self) -> Either[Entity]: return self._record_query_flush_result(result) - def _record_query_flush_result(self, result: Optional[BulkOperationResult]) -> Either[Entity]: # noqa: UP045 + def _record_query_flush_result( + self, result: Optional[BulkOperationResult] + ) -> Either[Entity]: # noqa: UP045 """Record a query bulk response. Already-present queries are reported as warnings (not failures) so a lineage run is not marked failed over queries that lost no metadata. Any other failure is still recorded as a failure.""" @@ -468,7 +470,10 @@ def _record_query_flush_result(self, result: Optional[BulkOperationResult]) -> E for failed in result.failedRequest or []: query_ref = failed.request or "unknown" if is_duplicate_query_conflict(failed.message): - self.status.warning("Query", f"Skipped already-present query [{query_ref}]: {failed.message}") + self.status.warning( + "Query", + f"Skipped already-present query [{query_ref}]: {failed.message}", + ) else: failure = StackTraceError( name="Query Buffer", @@ -478,7 +483,9 @@ def _record_query_flush_result(self, result: Optional[BulkOperationResult]) -> E self.status.failed(failure) first_failure = first_failure or failure - return Either(left=first_failure) if first_failure else Either(right=result) # pyright: ignore[reportCallIssue] + return ( + Either(left=first_failure) if first_failure else Either(right=result) + ) # pyright: ignore[reportCallIssue] @_run_dispatch.register def patch_entity(self, record: PatchRequest) -> Either[Entity]: @@ -566,14 +573,20 @@ def write_classification_and_tag( @_run_dispatch.register def write_lineage(self, add_lineage: AddLineageRequest) -> Either[Dict[str, Any]]: created_lineage = self.metadata.add_lineage(add_lineage, check_patch=True) - if created_lineage.get("error"): + if created_lineage and created_lineage.get("error"): return Either( left=StackTraceError( name="AddLineageRequestError", error=created_lineage["error"] ) ) - return Either(right=created_lineage["entity"]["fullyQualifiedName"]) + # The edge was written, but reading back the origin node's lineage can + # fail (e.g. 404 when its graph holds an edge to a deleted entity). + # Fall back to the request's from-entity so this is not a false error. + entity_fqn = (created_lineage or {}).get("entity", {}).get( + "fullyQualifiedName" + ) or model_str(add_lineage.edge.fromEntity.id) + return Either(right=entity_fqn) @_run_dispatch.register def write_override_lineage( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py index 1e6b87ae2f5a..bac65e0bf60f 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py @@ -219,6 +219,22 @@ def __init__( self._added_lineage: Optional[Dict] = {} + @property + def _ui_base(self) -> str: + """ + Base URL for human-facing "View in Looker" links. + + `hostPort` is bound to the Looker API host (the SDK uses it as + LOOKERSDK_BASE_URL), which is not always the URL a user should open in + the browser. When the API host differs from the UI host, set the + LOOKER_UI_BASE env var to the UI base URL; otherwise we fall back to + hostPort so behavior is unchanged. + """ + return clean_uri( + os.environ.get("LOOKER_UI_BASE") + or str(self.service_connection.hostPort) + ) + @classmethod def create( cls, @@ -1368,7 +1384,7 @@ def yield_dashboard( # like LookML assets, but rather just organised in folders. project=self.get_project_name(dashboard_details), sourceUrl=SourceUrl( - f"{clean_uri(self.service_connection.hostPort)}/dashboards/{dashboard_details.id}" + f"{self._ui_base}/dashboards/{dashboard_details.id}" ), service=self.context.get().dashboard_service, owners=self.get_owner_ref(dashboard_details=dashboard_details), @@ -1760,8 +1776,18 @@ def yield_dashboard_chart( source_url = chart.query.share_url elif getattr(chart.result_maker, "query", None) is not None: source_url = chart.result_maker.query.share_url + elif chart.merge_result_id is not None: + source_url = f"{self._ui_base}/merge?mid={chart.merge_result_id}" else: - source_url = f"{clean_uri(self.service_connection.hostPort)}/merge?mid={chart.merge_result_id}" + # No query, no result-maker query, and no merge result: + # a data-less Looker tile (text/markdown/decoration). Skip + # instead of emitting a dead {host}/merge?mid=None link. + logger.info( + f"Skipping data-less Looker tile id={chart.id} " + f"title={chart.title!r} type={chart.type}: no query, " + "no result-maker query, no merge result" + ) + continue yield Either( right=CreateChartRequest( name=EntityName(chart.id), diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py index 76f7fb97713c..a4de8346ba4a 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py @@ -117,6 +117,12 @@ def yield_dashboard( ) for chart in self.context.get().charts or [] ], + # Force-clear Dashboard.dataModels by sending an empty list. + # See comment in SupersetDBSource.yield_dashboard for why + # we represent the DataModel<->Dashboard relationship via + # the DataModel -> Chart -> Dashboard lineage chain + # instead of the structural Dashboard.dataModels field. + dataModels=[], service=FullyQualifiedEntityName(self.context.get().dashboard_service), owners=self.get_owner_ref(dashboard_details=dashboard_details), ) diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py index a0b6225c16b1..8e4e3c04b747 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py @@ -144,6 +144,13 @@ def yield_dashboard( ) for chart in self.context.get().charts or [] ], + # Force-clear Dashboard.dataModels by sending an empty list. + # The DataModel<->Dashboard relationship is represented via + # the DataModel -> Chart -> Dashboard lineage chain emitted + # in SupersetSourceMixin.yield_dashboard_lineage_details. + # Sending [] (instead of omitting the field) ensures any + # datamodel entries left over from prior runs are deleted. + dataModels=[], service=FullyQualifiedEntityName(self.context.get().dashboard_service), owners=self.get_owner_ref(dashboard_details=dashboard_details), ) diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py index 58551bb87445..ce087169f41f 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py @@ -19,6 +19,8 @@ from collate_sqllineage.core.models import Table as LineageTable from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest +from metadata.generated.schema.entity.data.chart import Chart +from metadata.generated.schema.entity.data.dashboard import Dashboard from metadata.generated.schema.entity.data.dashboardDataModel import DashboardDataModel from metadata.generated.schema.entity.data.table import Column, DataType, Table from metadata.generated.schema.entity.services.connections.dashboard.supersetConnection import ( @@ -145,8 +147,11 @@ def _get_charts_of_dashboard( raw_position_data = dashboard_details.position_json # For Superset 5.0.0+, position_json may be missing from list endpoint - # In this case, fetch individual dashboard details - if not raw_position_data and hasattr(self, "client"): + # In this case, fetch individual dashboard details. Only the API + # client supports this; in DB mode self.client is a SQLAlchemy Engine. + if not raw_position_data and hasattr( + getattr(self, "client", None), "fetch_dashboard" + ): dashboard_response = self.client.fetch_dashboard(dashboard_details.id) if dashboard_response and dashboard_response.result: raw_position_data = dashboard_response.result.position_json @@ -319,14 +324,36 @@ def _get_dashboard_data_model_entity( fqn=datamodel_fqn, ) + def yield_datamodel_dashboard_lineage( + self, + ) -> Iterable[Either[AddLineageRequest]]: + """ + Skip the base class' direct DataModel -> Dashboard lineage edge. + For Superset we bridge the chain through the Chart node so the graph + renders DataModel -> Chart -> Dashboard rather than DataModel -> + Dashboard alongside the chart list. The DataModel -> Chart edge is + emitted in yield_dashboard_lineage_details. + """ + return + yield # pragma: no cover # noqa: F841 # mark this as a generator + def yield_dashboard_lineage_details( self, dashboard_details: Union[FetchDashboard, DashboardResult], db_service_prefix: Optional[str] = None, ) -> Iterable[Either[AddLineageRequest]]: """ - Get lineage between datamodel and table + Emit lineage edges Table -> DataModel -> Chart -> Dashboard for every + chart on this dashboard. Dashboard.charts (set in yield_dashboard) + is a structural ref only — the dashboard lineage graph traverses + through explicit lineage edges, so we also emit Chart -> Dashboard + here. The base class' direct DataModel -> Dashboard edge is + suppressed by the override of yield_datamodel_dashboard_lineage so + the chart node bridges the chain in the rendered graph. """ + # Resolve the dashboard entity once per dashboard, not once per chart, + # to avoid an N+1 lookup against the metadata server. + dashboard_entity = self._get_dashboard_entity(dashboard_details) for chart_json in filter( None, [ @@ -350,6 +377,26 @@ def yield_dashboard_lineage_details( from_entity=from_entity_table, column_lineage=column_lineage, ) + + # DataModel -> Chart -> Dashboard bridge: emit BOTH edges + # so the dashboard's lineage graph renders the chart + # between the datamodel and the dashboard, instead of + # the datamodel hanging off the dashboard directly. + chart_entity = self._get_chart_entity(chart_json) + if chart_entity is not None: + dm_to_chart = self._get_add_lineage_request( + to_entity=chart_entity, + from_entity=to_entity, + ) + if dm_to_chart is not None: + yield dm_to_chart + if dashboard_entity is not None: + chart_to_dash = self._get_add_lineage_request( + to_entity=dashboard_entity, + from_entity=chart_entity, + ) + if chart_to_dash is not None: + yield chart_to_dash except Exception as exc: yield Either( left=StackTraceError( @@ -362,6 +409,54 @@ def yield_dashboard_lineage_details( ) ) + def _get_dashboard_entity(self, dashboard_details) -> Optional[Dashboard]: + """ + Look up the Dashboard entity created earlier so we can emit a + Chart -> Dashboard lineage edge. + """ + dashboard_id = getattr(dashboard_details, "id", None) + if dashboard_id is None: + return None + try: + dashboard_fqn = fqn.build( + self.metadata, + entity_type=Dashboard, + service_name=self.context.get().dashboard_service, + dashboard_name=str(dashboard_id), + ) + return self.metadata.get_by_name(entity=Dashboard, fqn=dashboard_fqn) + except Exception as exc: # pylint: disable=broad-except + logger.warning( + "Failed to resolve dashboard entity for dashboard_id=%s: %s", + dashboard_id, + exc, + ) + return None + + def _get_chart_entity(self, chart_json) -> Optional[Chart]: + """ + Look up the Chart entity created earlier in this pipeline so we can + emit a DataModel -> Chart lineage edge. + """ + chart_id = getattr(chart_json, "id", None) + if chart_id is None: + return None + try: + chart_fqn = fqn.build( + self.metadata, + entity_type=Chart, + service_name=self.context.get().dashboard_service, + chart_name=str(chart_id), + ) + return self.metadata.get_by_name(entity=Chart, fqn=chart_fqn) + except Exception as exc: # pylint: disable=broad-except + logger.warning( + "Failed to resolve chart entity for chart_id=%s: %s", + chart_id, + exc, + ) + return None + def _get_datamodel( self, datamodel: Union[SupersetDatasource, FetchChart] ) -> Optional[DashboardDataModel]: diff --git a/ingestion/tests/unit/test_sink_lineage_none_response.py b/ingestion/tests/unit/test_sink_lineage_none_response.py new file mode 100644 index 000000000000..49217008a27e --- /dev/null +++ b/ingestion/tests/unit/test_sink_lineage_none_response.py @@ -0,0 +1,69 @@ +# Copyright 2026 Collate +# Licensed under the Collate Community License, Version 1.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" +MetadataRestSink.write_lineage must not crash when add_lineage returns None, +which happens when the edge is written but reading back the origin node's +lineage fails (e.g. 404 because its graph references a deleted entity). +""" +from unittest.mock import MagicMock + +from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest +from metadata.generated.schema.type.entityLineage import EntitiesEdge +from metadata.generated.schema.type.entityReference import EntityReference +from metadata.ingestion.sink.metadata_rest import ( + MetadataRestSink, + MetadataRestSinkConfig, +) + +FROM_ID = "ab8f49d7-afba-4164-ba06-a0dc6300c5d2" +TO_ID = "615ff273-1180-4fc2-8682-b5e4a646a6fe" + +LINEAGE_REQUEST = AddLineageRequest( + edge=EntitiesEdge( + fromEntity=EntityReference(id=FROM_ID, type="dashboardDataModel"), + toEntity=EntityReference(id=TO_ID, type="chart"), + ) +) + + +def _sink_with_add_lineage_returning(value) -> MetadataRestSink: + metadata = MagicMock() + metadata.add_lineage.return_value = value + return MetadataRestSink(config=MetadataRestSinkConfig(), metadata=metadata) + + +def test_write_lineage_handles_none_response(): + sink = _sink_with_add_lineage_returning(None) + + result = sink.write_lineage(LINEAGE_REQUEST) + + assert result.left is None + assert result.right == FROM_ID + + +def test_write_lineage_error_response_is_reported(): + sink = _sink_with_add_lineage_returning({"error": "Error 400 trying to PUT"}) + + result = sink.write_lineage(LINEAGE_REQUEST) + + assert result.left is not None + assert "Error 400" in result.left.error + + +def test_write_lineage_success_returns_entity_fqn(): + sink = _sink_with_add_lineage_returning( + {"entity": {"fullyQualifiedName": "service.model_1"}} + ) + + result = sink.write_lineage(LINEAGE_REQUEST) + + assert result.left is None + assert result.right == "service.model_1" diff --git a/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py b/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py new file mode 100644 index 000000000000..3d84d89bcfef --- /dev/null +++ b/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py @@ -0,0 +1,264 @@ +# Copyright 2026 Collate +# Licensed under the Collate Community License, Version 1.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" +Regression tests for the Superset Dashboard <-> DataModel <-> Chart linking +behaviour. Verifies: + +1. yield_dashboard sends dataModels=[] on every CreateDashboardRequest, so + stale Dashboard.dataModels entries from previous runs are cleared. + +2. The Superset override of yield_datamodel_dashboard_lineage produces no + edges, suppressing the base class' direct DataModel -> Dashboard edge. + +3. yield_dashboard_lineage_details emits both DataModel -> Chart and + Chart -> Dashboard edges so the lineage graph renders the chart as the + bridge between datamodels and the dashboard. +""" + +import uuid +from unittest import TestCase +from unittest.mock import MagicMock + +from metadata.generated.schema.api.data.createDashboard import CreateDashboardRequest +from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest +from metadata.generated.schema.entity.data.chart import Chart, ChartType +from metadata.generated.schema.entity.data.dashboard import Dashboard, DashboardType +from metadata.generated.schema.entity.data.dashboardDataModel import ( + DashboardDataModel, + DataModelType, +) +from metadata.generated.schema.type.basic import ( + EntityName, + FullyQualifiedEntityName, + SourceUrl, +) +from metadata.generated.schema.type.entityReference import EntityReference +from metadata.ingestion.api.models import Either +from metadata.ingestion.source.dashboard.dashboard_service import ( + DashboardServiceSource, +) +from metadata.ingestion.source.dashboard.superset.api_source import SupersetAPISource +from metadata.ingestion.source.dashboard.superset.db_source import SupersetDBSource +from metadata.ingestion.source.dashboard.superset.mixin import SupersetSourceMixin +from metadata.ingestion.source.dashboard.superset.models import ( + DashboardResult, + FetchChart, + FetchDashboard, +) + + +def _make_dashboard_entity() -> Dashboard: + return Dashboard( + id=uuid.uuid4(), + name=EntityName("4"), + fullyQualifiedName=FullyQualifiedEntityName("superset_test.4"), + dashboardType=DashboardType.Dashboard, + service=EntityReference(id=uuid.uuid4(), type="dashboardService"), + ) + + +def _make_chart_entity(chart_id: str) -> Chart: + return Chart( + id=uuid.uuid4(), + name=EntityName(chart_id), + fullyQualifiedName=FullyQualifiedEntityName(f"superset_test.{chart_id}"), + chartType=ChartType.Table, + service=EntityReference(id=uuid.uuid4(), type="dashboardService"), + ) + + +def _make_datamodel_entity(datamodel_id: str) -> DashboardDataModel: + return DashboardDataModel( + id=uuid.uuid4(), + name=EntityName(datamodel_id), + fullyQualifiedName=FullyQualifiedEntityName( + f"superset_test.model.{datamodel_id}" + ), + dataModelType=DataModelType.SupersetDataModel, + columns=[], + service=EntityReference(id=uuid.uuid4(), type="dashboardService"), + ) + + +def _build_context(charts=None, datamodels=None) -> MagicMock: + ctx = MagicMock() + ctx.dashboard_service = "superset_test" + ctx.charts = charts or [] + ctx.dataModels = datamodels or [] + return ctx + + +class TestSupersetDashboardDataModelsCleared(TestCase): + """yield_dashboard must send dataModels=[] (not omit) so server clears + any stale Dashboard.dataModels entries persisted from earlier runs.""" + + def test_db_source_yields_empty_data_models(self): + source = SupersetDBSource.__new__(SupersetDBSource) + source.metadata = MagicMock() + source.service_connection = MagicMock(hostPort="https://superset.example.com") + source.context = MagicMock() + source.context.get.return_value = _build_context( + charts=["10", "11"], datamodels=["45", "46"] + ) + source.get_owner_ref = MagicMock(return_value=None) + + dashboard_details = FetchDashboard( + id=4, + dashboard_title="Meltano", + position_json=None, + published=True, + email=None, + json_metadata=None, + ) + + results = list(SupersetDBSource.yield_dashboard(source, dashboard_details)) + self.assertEqual(len(results), 1) + request: CreateDashboardRequest = results[0].right + # The whole point of the regression fix: dataModels MUST be an empty + # list, not None or absent — that's what tells the server to clear + # the field. + self.assertEqual(request.dataModels, []) + self.assertEqual(len(request.charts), 2) + + def test_api_source_yields_empty_data_models(self): + source = SupersetAPISource.__new__(SupersetAPISource) + source.metadata = MagicMock() + source.service_connection = MagicMock(hostPort="https://superset.example.com") + source.context = MagicMock() + source.context.get.return_value = _build_context( + charts=["10"], datamodels=["45"] + ) + source.get_owner_ref = MagicMock(return_value=None) + + dashboard_details = DashboardResult( + id=4, + dashboard_title="Meltano", + url="/dashboard/4/", + published=True, + position_json=None, + email=None, + json_metadata=None, + ) + + results = list(SupersetAPISource.yield_dashboard(source, dashboard_details)) + self.assertEqual(len(results), 1) + request: CreateDashboardRequest = results[0].right + self.assertEqual(request.dataModels, []) + self.assertEqual(len(request.charts), 1) + + +class TestSupersetSuppressesDirectDataModelDashboardEdge(TestCase): + """The Superset mixin must override yield_datamodel_dashboard_lineage to + emit zero edges — otherwise the base class produces a direct + DataModel -> Dashboard lineage edge that bypasses the chart node.""" + + def test_override_yields_no_edges(self): + source = MagicMock() + # Bind the unbound method to the mock so the override runs + result = list( + SupersetSourceMixin.yield_datamodel_dashboard_lineage(source) + ) + self.assertEqual(result, []) + + +class TestSupersetEmitsChartBridgeEdges(TestCase): + """yield_dashboard_lineage_details must emit DataModel -> Chart and + Chart -> Dashboard edges so the chart bridges the chain in the + lineage graph.""" + + def _make_source(self, chart_entity, datamodel_entity, dashboard_entity): + source = MagicMock() + # Real bound methods we want to exercise + source.yield_dashboard_lineage_details = ( + lambda *a, **kw: SupersetSourceMixin.yield_dashboard_lineage_details( + source, *a, **kw + ) + ) + source._get_chart_entity = lambda chart_json: chart_entity + source._get_dashboard_entity = lambda dashboard_details: dashboard_entity + source._get_dashboard_data_model_entity = lambda chart_json: datamodel_entity + source._get_input_tables = lambda chart_json: [] + source._enrich_raw_input_tables = lambda inputs, to_entity, prefix: [] + source._get_charts_of_dashboard = lambda dashboard_details: ["10"] + source._get_add_lineage_request = DashboardServiceSource._get_add_lineage_request + source.all_charts = { + "10": FetchChart( + id=10, + slice_name="chart-10", + datasource_id=45, + viz_type="table", + table_name="t", + table_id=1, + table_schema=None, + schema_name=None, + sql=None, + params=None, + description=None, + url=None, + ) + } + return source + + def test_chart_bridge_edges_emitted(self): + chart = _make_chart_entity("10") + datamodel = _make_datamodel_entity("45") + dashboard = _make_dashboard_entity() + source = self._make_source(chart, datamodel, dashboard) + + dashboard_details = FetchDashboard( + id=4, + dashboard_title="Meltano", + position_json=None, + published=True, + email=None, + json_metadata=None, + ) + results = [ + r for r in source.yield_dashboard_lineage_details(dashboard_details) + if r is not None and r.right is not None + ] + + edges = [r.right for r in results if isinstance(r.right, AddLineageRequest)] + self.assertEqual(len(edges), 2, f"expected 2 edges, got {edges}") + + from_to = {(e.edge.fromEntity.id.root, e.edge.toEntity.id.root) for e in edges} + self.assertIn( + (datamodel.id.root, chart.id.root), + from_to, + "DataModel -> Chart edge missing", + ) + self.assertIn( + (chart.id.root, dashboard.id.root), + from_to, + "Chart -> Dashboard edge missing", + ) + + def test_no_chart_entity_skips_bridge_edges(self): + # When the Chart entity isn't yet visible on the server, both bridge + # edges should be skipped (no crash, just no emission). + datamodel = _make_datamodel_entity("45") + source = self._make_source( + chart_entity=None, datamodel_entity=datamodel, dashboard_entity=None + ) + dashboard_details = FetchDashboard( + id=4, + dashboard_title="Meltano", + position_json=None, + published=True, + email=None, + json_metadata=None, + ) + results = [ + r for r in source.yield_dashboard_lineage_details(dashboard_details) + if r is not None and r.right is not None + ] + edges = [r.right for r in results if isinstance(r.right, AddLineageRequest)] + self.assertEqual(edges, [], "no edges should be emitted without chart entity") diff --git a/ingestion/tests/unit/topology/dashboard/test_superset_charts_fallback.py b/ingestion/tests/unit/topology/dashboard/test_superset_charts_fallback.py new file mode 100644 index 000000000000..c0ba72d9ae7d --- /dev/null +++ b/ingestion/tests/unit/topology/dashboard/test_superset_charts_fallback.py @@ -0,0 +1,75 @@ +# Copyright 2026 Collate +# Licensed under the Collate Community License, Version 1.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" +Tests for the Superset 5.0+ position_json fallback in +SupersetSourceMixin._get_charts_of_dashboard +""" +from types import SimpleNamespace +from unittest.mock import MagicMock + +from metadata.ingestion.source.dashboard.superset.mixin import SupersetSourceMixin +from metadata.ingestion.source.dashboard.superset.models import ( + DashboardResult, + FetchDashboard, + FetchedDashboard, +) + + +def _get_charts_of_dashboard(client, dashboard): + """Call the mixin method unbound: the class is abstract, and the method + only relies on self.client""" + stub_self = SimpleNamespace(client=client) + return SupersetSourceMixin._get_charts_of_dashboard(stub_self, dashboard) + + +def test_db_mode_engine_without_position_json_does_not_call_api_fallback(caplog): + """ + In DB mode self.client is a SQLAlchemy Engine, which has no + fetch_dashboard method. The API fallback must be skipped instead of + raising `'Engine' object has no attribute 'fetch_dashboard'`. + """ + engine = MagicMock(spec=["connect", "execute"]) + dashboard = FetchDashboard(id=464, position_json=None) + + result = _get_charts_of_dashboard(engine, dashboard) + + assert result == [] + assert not any( + "Failed to charts of dashboard" in record.message for record in caplog.records + ) + + +def test_api_mode_fallback_fetches_dashboard_details(): + client = MagicMock(spec=["fetch_dashboard"]) + client.fetch_dashboard.return_value = FetchedDashboard( + id=14, + result=DashboardResult( + position_json='{"CHART-abc": {"meta": {"chartId": 69}}}' + ), + ) + dashboard = FetchDashboard(id=14, position_json=None) + + result = _get_charts_of_dashboard(client, dashboard) + + client.fetch_dashboard.assert_called_once_with(14) + assert result == [69] + + +def test_position_json_present_skips_fallback(): + client = MagicMock(spec=["fetch_dashboard"]) + dashboard = FetchDashboard( + id=7, position_json='{"CHART-abc": {"meta": {"chartId": 42}}}' + ) + + result = _get_charts_of_dashboard(client, dashboard) + + client.fetch_dashboard.assert_not_called() + assert result == [42]