Fixes #29764: refactor(ingestion): migrate powerbi test-connection to declarative framework

ingestion/src/metadata/core/connections/test_connection/checks/dashboard.py

@@ -0,0 +1,72 @@
+#  Copyright 2025 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.
+"""
+Dashboard service step identity and shared check helpers.
+
+Dashboard connectors reach their source over a REST API (with OAuth or token
+auth), not a SQLAlchemy engine, so these helpers stay HTTP-generic: they call a
+connector-supplied callable and summarize what it returned, and on failure raise
+``CheckError`` carrying the label of the call they attempted, so a failed step
+still reports its ``Evidence`` to the backend. The runner and the rest of the
+package stay engine- and protocol-agnostic; connectors reuse these helpers from
+their ``@check`` methods.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from metadata.core.connections.test_connection.check import CheckError, StepName
+from metadata.core.connections.test_connection.records import Evidence
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Sized
+
+
+class DashboardStep(StepName):
+    """The steps a dashboard connector can be asked to verify."""
+
+    CheckAccess = "CheckAccess"
+    GetDashboards = "GetDashboards"
+
+
+def _count(number: int, noun: str) -> str:
+    """``3 dashboards`` / ``1 dashboard`` - pluralize the noun to match the count."""
+    return f"{number} {noun if number == 1 else noun + 's'}"
+
+
+def verify_access(authenticate: Callable[[], object], command: str) -> Evidence:
+    """Prove the connector can authenticate against the REST API.
+
+    Runs the connector's auth callable (e.g. an OAuth token acquisition) and, on
+    failure, re-raises as ``CheckError`` carrying the attempted command so the
+    gate step still reports what it tried.
+    """
+    try:
+        authenticate()
+    except Exception as cause:
+        raise CheckError(cause, Evidence(command=command)) from cause
+    return Evidence(summary="authenticated", command=command)
+
+
+def fetch_list(fetch: Callable[[], Sized | None], noun: str, command: str) -> Evidence:
+    """Call a REST list endpoint and report how many items it returned.
+
+    Reports the command it attempted and a count summary. On failure, re-raise as
+    ``CheckError`` carrying the command so the failed step still reports what it
+    ran.
+    """
+    try:
+        items = fetch()
+    except Exception as cause:
+        raise CheckError(cause, Evidence(command=command)) from cause
+    count = len(items) if items else 0
+    return Evidence(summary=f"{_count(count, noun)} enumerated", command=command)

ingestion/src/metadata/ingestion/source/dashboard/powerbi/connection.py

@@ -13,26 +13,82 @@
 Source connection handler
 """
 
-from typing import Optional
+from __future__ import annotations
 
-from metadata.generated.schema.entity.automations.workflow import (
-    Workflow as AutomationWorkflow,
+from typing import TYPE_CHECKING
+
+from metadata.core.connections.test_connection import (
+    ErrorPack,
+    Evidence,
+    Matchers,
+    check,
+    when,
+)
+from metadata.core.connections.test_connection.checks.dashboard import (
+    DashboardStep,
+    fetch_list,
+    verify_access,
 )
+from metadata.core.connections.test_connection.classifier import exception_chain
+from metadata.core.connections.test_connection.network import NETWORK_ERRORS
 from metadata.generated.schema.entity.services.connections.dashboard.powerBIConnection import (
     PowerBIConnection as PowerBIConnectionConfig,
 )
-from metadata.generated.schema.entity.services.connections.testConnectionResult import (
-    TestConnectionResult,
-)
+from metadata.ingestion.api.steps import InvalidSourceException
 from metadata.ingestion.connections.connection import BaseConnection
-from metadata.ingestion.connections.test_connections import test_connection_steps
-from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.dashboard.powerbi.client import (
     PowerBiApiClient,
     PowerBiClient,
 )
 from metadata.ingestion.source.dashboard.powerbi.file_client import PowerBiFileClient
-from metadata.utils.constants import THREE_MIN
+
+if TYPE_CHECKING:
+    from metadata.core.connections.test_connection import ChecksProvider
+    from metadata.core.connections.test_connection.classifier import Matcher
+
+
+def _http_status(*codes: int) -> Matcher:
+    """Match a PowerBI REST error by HTTP status.
+
+    The REST client raises ``APIError`` carrying the response status on its
+    ``.status_code`` property; we walk the cause chain and match on it. The status
+    is the stable signal - the message body varies by tenant and endpoint."""
+    wanted = frozenset(codes)
+
+    def match(error: BaseException) -> bool:
+        result = False
+        for current in exception_chain(error):
+            code = getattr(current, "status_code", None)
+            if isinstance(code, int) and code in wanted:
+                result = True
+        return result
+
+    return match
+
+
+POWERBI_ERRORS = ErrorPack(
+    when(_http_status(401)).diagnose(
+        "Authentication failed",
+        fix="The service principal token was rejected (401). Check the Client ID, Client Secret, "
+        "and Tenant ID, and that the secret has not expired.",
+    ),
+    when(Matchers.exception(InvalidSourceException)).diagnose(
+        "Authentication failed",
+        fix="Could not acquire an OAuth token. Check the Client ID, Client Secret, and Tenant ID, "
+        "and that the app registration is allowed to request the configured scope.",
+    ),
+    when(_http_status(403)).diagnose(
+        "Insufficient permissions",
+        fix="The token authenticated but is not authorized (403). Grant the service principal the "
+        "required Power BI tenant settings / admin API access, and enable service-principal access "
+        "in the Power BI admin portal.",
+    ),
+    when(_http_status(404)).diagnose(
+        "Resource not found",
+        fix="The requested resource was not found (404). Check the API URL and that the configured "
+        "tenant/workspace exists and is visible to the service principal.",
+    ),
+).including(NETWORK_ERRORS)
 
 
 def get_connection(connection: PowerBIConnectionConfig) -> PowerBiClient:
@@ -45,28 +101,39 @@ def get_connection(connection: PowerBIConnectionConfig) -> PowerBiClient:
     return PowerBiClient(api_client=PowerBiApiClient(connection), file_client=file_client)
 
 
+class PowerBIChecks:
+    """Test-connection checks for PowerBI.
+
+    ``CheckAccess`` is the gate: it acquires an OAuth token, so a bad service
+    principal fails fast before any list endpoint is dialled. ``GetDashboards``
+    then exercises list access. No network call happens at construction; each
+    ``@check`` builds and runs its own call.
+    """
+
+    errors = POWERBI_ERRORS
+
+    def __init__(self, client: PowerBiClient) -> None:
+        self.client = client
+
+    @check(DashboardStep.CheckAccess)
+    def check_access(self) -> Evidence:
+        return verify_access(
+            self.client.api_client.get_auth_token,
+            command="acquire OAuth token",
+        )
+
+    @check(DashboardStep.GetDashboards)
+    def get_dashboards(self) -> Evidence:
+        return fetch_list(
+            self.client.api_client.fetch_dashboards,
+            noun="dashboard",
+            command="fetch dashboards",
+        )
+
+
 class PowerBIConnection(BaseConnection[PowerBIConnectionConfig, PowerBiClient]):
     def _get_client(self) -> PowerBiClient:
         return get_connection(self.service_connection)
 
-    def test_connection(
-        self,
-        metadata: OpenMetadata,
-        automation_workflow: Optional[AutomationWorkflow] = None,  # noqa: UP045
-        timeout_seconds: Optional[int] = THREE_MIN,  # noqa: UP045
-    ) -> TestConnectionResult:
-        """
-        Test connection. This can be executed either as part
-        of a metadata workflow or during an Automation Workflow
-        """
-        client = self.client
-        service_connection = self.service_connection
-        test_fn = {"GetDashboards": client.api_client.fetch_dashboards}
-
-        return test_connection_steps(
-            metadata=metadata,
-            test_fn=test_fn,
-            service_type=service_connection.type.value,  # pyright: ignore[reportOptionalMemberAccess]
-            automation_workflow=automation_workflow,
-            timeout_seconds=timeout_seconds,
-        )
+    def checks(self) -> ChecksProvider:
+        return PowerBIChecks(client=self.client)

ingestion/tests/unit/source/dashboard/powerbi/test_connection.py

@@ -8,16 +8,38 @@
 #  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.
-"""Unit tests for Power BI connection handling."""
+"""Unit tests for Power BI test-connection checks."""
 
+import socket
 from unittest.mock import MagicMock, patch
 
+import pytest
+
+from metadata.core.connections.test_connection.check import CheckError, collect_checks
+from metadata.core.connections.test_connection.checks.dashboard import DashboardStep
+from metadata.ingestion.api.steps import InvalidSourceException
 from metadata.ingestion.connections.connection import BaseConnection
-from metadata.ingestion.source.dashboard.powerbi.connection import PowerBIConnection
+from metadata.ingestion.ometa.client import APIError
+from metadata.ingestion.source.dashboard.powerbi.connection import (
+    POWERBI_ERRORS,
+    PowerBIChecks,
+    PowerBIConnection,
+)
 
 CONNECTION_MODULE = "metadata.ingestion.source.dashboard.powerbi.connection"
 
 
+def _api_error(status_code: int) -> APIError:
+    http_error = MagicMock()
+    http_error.response.status_code = status_code
+    return APIError({"message": "boom", "code": "x"}, http_error)
+
+
+def _checks() -> tuple[PowerBIChecks, MagicMock]:
+    client = MagicMock()
+    return PowerBIChecks(client=client), client
+
+
 def test_powerbi_connection_is_base_connection():
     assert issubclass(PowerBIConnection, BaseConnection)
 
@@ -31,10 +53,91 @@ def test_get_client_delegates_to_get_connection():
     mock_get.assert_called_once_with(conn.service_connection)
 
 
-def test_test_connection_runs_steps():
-    conn = PowerBIConnection(MagicMock())
-    conn._client = MagicMock()
-    with patch(f"{CONNECTION_MODULE}.test_connection_steps") as mock_step:
-        result = conn.test_connection(metadata=MagicMock())
+def test_checks_does_not_touch_the_network():
+    with patch(f"{CONNECTION_MODULE}.get_connection") as mock_get:
+        conn = PowerBIConnection(MagicMock())
+        provider = conn.checks()
+
+    assert isinstance(provider, PowerBIChecks)
+    api_client = mock_get.return_value.api_client
+    api_client.get_auth_token.assert_not_called()
+    api_client.fetch_dashboards.assert_not_called()
+
+
+def test_collect_checks_maps_every_step():
+    provider, _ = _checks()
+    collected = collect_checks(provider)
+
+    assert set(collected) == {DashboardStep.CheckAccess, DashboardStep.GetDashboards}
+
+
+def test_check_access_authenticates():
+    provider, client = _checks()
+    client.api_client.get_auth_token.return_value = ("token", "3600")
+
+    evidence = provider.check_access()
+
+    client.api_client.get_auth_token.assert_called_once_with()
+    assert evidence.summary == "authenticated"
+    assert evidence.command == "acquire OAuth token"
+
+
+def test_check_access_wraps_failure_as_check_error():
+    provider, client = _checks()
+    client.api_client.get_auth_token.side_effect = InvalidSourceException("no token")
+
+    with pytest.raises(CheckError) as exc_info:
+        provider.check_access()
+
+    assert isinstance(exc_info.value.cause, InvalidSourceException)
+    assert exc_info.value.evidence.command == "acquire OAuth token"
+
+
+def test_get_dashboards_counts_results():
+    provider, client = _checks()
+    client.api_client.fetch_dashboards.return_value = [object(), object(), object()]
+
+    evidence = provider.get_dashboards()
+
+    assert evidence.summary == "3 dashboards enumerated"
+    assert evidence.command == "fetch dashboards"
+
+
+def test_get_dashboards_empty_is_singular_aware():
+    provider, client = _checks()
+    client.api_client.fetch_dashboards.return_value = None
+
+    evidence = provider.get_dashboards()
+
+    assert evidence.summary == "0 dashboards enumerated"
+
+
+def test_get_dashboards_wraps_failure_as_check_error():
+    provider, client = _checks()
+    client.api_client.fetch_dashboards.side_effect = _api_error(403)
+
+    with pytest.raises(CheckError) as exc_info:
+        provider.get_dashboards()
+
+    assert exc_info.value.evidence.command == "fetch dashboards"
+
+
+@pytest.mark.parametrize(
+    ("error", "title"),
+    [
+        (_api_error(401), "Authentication failed"),
+        (InvalidSourceException("bad creds"), "Authentication failed"),
+        (_api_error(403), "Insufficient permissions"),
+        (_api_error(404), "Resource not found"),
+        (socket.gaierror("name resolution failed"), "Host could not be resolved"),
+    ],
+)
+def test_error_pack_classifies(error, title):
+    diagnosis = POWERBI_ERRORS.classify(error)
+
+    assert diagnosis is not None
+    assert diagnosis.title == title
+
 
-    assert result is mock_step.return_value
+def test_error_pack_ignores_unknown_error():
+    assert POWERBI_ERRORS.classify(ValueError("something else")) is None

openmetadata-service/src/main/resources/json/data/testConnections/dashboard/powerbi.json

@@ -3,6 +3,14 @@
     "displayName": "PowerBI Test Connection",
     "description": "This Test Connection validates the access against the server and basic metadata extraction of dashboards and charts.",
     "steps": [
+      {
+        "name": "CheckAccess",
+        "description": "Validate that we can acquire an OAuth token and authenticate against the PowerBI REST API with the given service principal credentials.",
+        "errorMessage": "Failed to authenticate against PowerBI, please validate the Client ID, Client Secret and Tenant ID.",
+        "mandatory": true,
+        "shortCircuit": true,
+        "category": "ConnectionGate"
+      },
       {
         "name": "GetDashboards",
         "description": "List all the dashboards available to the user",
@@ -12,5 +20,3 @@
       }
     ]
   }
-
-