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

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

@@ -0,0 +1,90 @@
+#  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
+
+
+# A check only needs to prove the list endpoint is reachable and returns items,
+# not enumerate every one, so ``fetch_list`` reports the count up to this cap and
+# marks it ``+`` beyond, keeping the summary bounded on huge tenants.
+DEFAULT_LIST_CAP = 100
+
+
+class DashboardStep(StepName):
+    """The steps a dashboard connector can be asked to verify."""
+
+    CheckAccess = "CheckAccess"
+    GetDashboards = "GetDashboards"
+
+
+def _count(number: int, noun: str, cap: int | None = None) -> str:
+    """``3 dashboards`` / ``1 dashboard`` - pluralize the noun to match the count.
+
+    When ``cap`` is given and the count reaches it, the figure is rendered
+    ``<cap>+`` so a capped sample is not read as an exact total.
+    """
+    plural = noun if number == 1 else noun + "s"
+    shown = f"{cap}+" if cap is not None and number >= cap else str(number)
+    return f"{shown} {plural}"
+
+
+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,
+    cap: int = DEFAULT_LIST_CAP,
+) -> Evidence:
+    """Call a REST list endpoint and report how many items it returned.
+
+    Reports the command it attempted and a count summary, capped at ``cap`` (shown
+    as ``<cap>+`` beyond) so a huge tenant does not produce an unbounded figure. 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, cap)} enumerated", command=command)

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

@@ -13,26 +13,87 @@
 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(Matchers.contains("authority")).diagnose(
+        "Invalid tenant or authority",
+        fix="The authority could not be resolved. Check the Tenant ID (a valid GUID or "
+        "tenant name) and the Authority URI, e.g. https://login.microsoftonline.com/.",
+    ),
+    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 +106,53 @@ 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.
+
+    The REST client is built lazily inside the first check, never at
+    construction: the underlying MSAL client performs authority/instance
+    discovery over the network in its constructor, so building it eagerly would
+    run before the runner's gate and surface as a raw workflow error instead of a
+    classified ``CheckAccess`` failure.
+    """
+
+    errors = POWERBI_ERRORS
+
+    def __init__(self, connection: PowerBIConnectionConfig) -> None:
+        self._connection = connection
+        self._api_client: PowerBiApiClient | None = None
+
+    def _client(self) -> PowerBiApiClient:
+        """Build (once) and return the REST client. The MSAL client it wraps does
+        authority/instance discovery over the network in its constructor, so this
+        is only ever called from inside a check - never at construction."""
+        if self._api_client is None:
+            self._api_client = get_connection(self._connection).api_client
+        return self._api_client
+
+    @check(DashboardStep.CheckAccess)
+    def check_access(self) -> Evidence:
+        return verify_access(
+            lambda: self._client().get_auth_token(),
+            command="acquire OAuth token",
+        )
+
+    @check(DashboardStep.GetDashboards)
+    def get_dashboards(self) -> Evidence:
+        return fetch_list(
+            self._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(connection=self.service_connection)