Merge branch 'fix/cascading-testcase-status-cleanup-27418' of https://github.com/Megh-Shah-08/OpenMetadata into fix/cascading-testcase-status-cleanup-27418

Author: Megh-Shah-08

.github/workflows/py-cli-e2e-tests.yml

@@ -182,12 +182,12 @@ jobs:
         echo "import os" >> $SITE_CUSTOMIZE_PATH
         echo "try:" >> $SITE_CUSTOMIZE_PATH
         echo "    import coverage" >> $SITE_CUSTOMIZE_PATH
-        echo "    os.environ['COVERAGE_PROCESS_START'] = 'ingestion/pyproject.toml'" >> $SITE_CUSTOMIZE_PATH
+        echo "    os.environ['COVERAGE_PROCESS_START'] = os.path.join(os.environ.get('GITHUB_WORKSPACE', os.getcwd()), 'ingestion', 'pyproject.toml')" >> $SITE_CUSTOMIZE_PATH
         echo "    coverage.process_startup()" >> $SITE_CUSTOMIZE_PATH
         echo "except ImportError:" >> $SITE_CUSTOMIZE_PATH
         echo "    pass" >> $SITE_CUSTOMIZE_PATH
-        coverage run --rcfile ingestion/pyproject.toml -a --branch -m pytest -c ingestion/pyproject.toml --junitxml=ingestion/junit/test-results-$E2E_TEST.xml --ignore=ingestion/tests/unit/source ingestion/tests/cli_e2e/test_cli_$E2E_TEST.py
-        coverage combine --data-file=.coverage.$E2E_TEST --rcfile=ingestion/pyproject.toml --keep -a .coverage*
+        coverage run --rcfile ingestion/pyproject.toml --branch -m pytest -c ingestion/pyproject.toml --junitxml=ingestion/junit/test-results-$E2E_TEST.xml --ignore=ingestion/tests/unit/source ingestion/tests/cli_e2e/test_cli_$E2E_TEST.py
+        coverage combine --data-file=.coverage.$E2E_TEST --rcfile=ingestion/pyproject.toml --keep .coverage*
         coverage report --rcfile ingestion/pyproject.toml --data-file .coverage.$E2E_TEST || true
 
     - name: Upload coverage artifact for Python unit tests
@@ -293,7 +293,7 @@ jobs:
           done
           source env/bin/activate
           cd ingestion
-          coverage combine --rcfile=pyproject.toml --keep -a .coverage*
+          coverage combine --rcfile=pyproject.toml --keep .coverage*
           coverage xml --rcfile=pyproject.toml --data-file=.coverage
         shell: bash
 

ingestion/pyproject.toml

@@ -43,6 +43,7 @@ provider_info = "airflow_provider_openmetadata:get_provider_config"
 source = ["metadata"]
 relative_files = true
 branch = true
+parallel = true
 
 # Remap installed-package paths back to the source tree so that
 # ``coverage combine`` (in a lightweight CI job without the package

ingestion/src/metadata/ingestion/source/dashboard/ssrs/client.py

@@ -11,11 +11,12 @@
 """
 SSRS REST client
 """
-import traceback
-from typing import List, Optional, Union
+from typing import Iterable, Iterator, Optional, Union
 
 import requests
+from requests.adapters import HTTPAdapter
 from requests_ntlm import HttpNtlmAuth
+from urllib3.util.retry import Retry
 
 from metadata.generated.schema.entity.services.connections.dashboard.ssrsConnection import (
     SsrsConnection,
@@ -33,8 +34,14 @@
 logger = ingestion_logger()
 
 API_VERSION = "api/v2.0"
-DEFAULT_TIMEOUT = 30
+CONNECT_TIMEOUT = 10
+READ_TIMEOUT = 120
 PAGE_SIZE = 100
+MAX_RETRIES = 2
+BACKOFF_FACTOR = 1
+RETRY_STATUS_CODES = (500, 502, 503, 504)
+REPORT_SELECT_FIELDS = "Id,Name,Path,Description,Type,Hidden,HasDataSources"
+FOLDER_SELECT_FIELDS = "Id,Name,Path"
 
 
 class SsrsClient:
@@ -53,17 +60,51 @@ def __init__(
         self.session.headers.update({"Accept": "application/json"})
         if verify_ssl is not None:
             self.session.verify = verify_ssl
+        retry = Retry(
+            total=MAX_RETRIES,
+            connect=MAX_RETRIES,
+            read=MAX_RETRIES,
+            status=MAX_RETRIES,
+            backoff_factor=BACKOFF_FACTOR,
+            status_forcelist=RETRY_STATUS_CODES,
+            allowed_methods=frozenset(["GET"]),
+            raise_on_status=False,
+        )
+        adapter = HTTPAdapter(max_retries=retry)
+        self.session.mount("http://", adapter)
+        self.session.mount("https://", adapter)
 
     def close(self) -> None:
         if self.session:
             self.session.close()
 
     def _get(self, path: str, params: Optional[dict] = None) -> dict:
         url = f"{self.base_url}{path}"
-        resp = self.session.get(url, timeout=DEFAULT_TIMEOUT, params=params)
+        resp = self.session.get(
+            url, timeout=(CONNECT_TIMEOUT, READ_TIMEOUT), params=params
+        )
         resp.raise_for_status()
         return resp.json()
 
+    def _paginate(self, path: str, params: dict, resource_label: str) -> Iterable[dict]:
+        """Yield pages from an OData endpoint. Any per-page failure raises
+        ``SourceConnectionException`` so callers can surface it instead of
+        producing a silently truncated result set."""
+        skip = 0
+        while True:
+            page_params = {**params, "$top": str(PAGE_SIZE), "$skip": str(skip)}
+            try:
+                data = self._get(path, params=page_params)
+            except Exception as exc:
+                raise SourceConnectionException(
+                    f"Failed to fetch SSRS {resource_label} at skip={skip}: {exc}"
+                ) from exc
+            yield data
+            value = data.get("value") or []
+            if len(value) < PAGE_SIZE:
+                return
+            skip += PAGE_SIZE
+
     def test_access(self) -> None:
         try:
             self._get("/Folders", params={"$top": "1"})
@@ -72,40 +113,26 @@ def test_access(self) -> None:
                 f"Failed to connect to SSRS: {exc}"
             ) from exc
 
-    def get_folders(self) -> List[SsrsFolder]:
+    def test_get_reports(self) -> None:
         try:
-            results: List[SsrsFolder] = []
-            skip = 0
-            while True:
-                data = self._get(
-                    "/Folders", params={"$top": str(PAGE_SIZE), "$skip": str(skip)}
-                )
-                response = SsrsFolderListResponse(**data)
-                results.extend(response.value)
-                if len(response.value) < PAGE_SIZE:
-                    break
-                skip += PAGE_SIZE
-            return results
+            self._get("/Reports", params={"$top": "1"})
         except Exception as exc:
-            logger.debug(traceback.format_exc())
-            logger.warning("Failed to fetch SSRS folders: %s", exc)
-        return []
+            raise SourceConnectionException(
+                f"Failed to fetch SSRS reports: {exc}"
+            ) from exc
 
-    def get_reports(self) -> List[SsrsReport]:
-        try:
-            results: List[SsrsReport] = []
-            skip = 0
-            while True:
-                data = self._get(
-                    "/Reports", params={"$top": str(PAGE_SIZE), "$skip": str(skip)}
-                )
-                response = SsrsReportListResponse(**data)
-                results.extend(response.value)
-                if len(response.value) < PAGE_SIZE:
-                    break
-                skip += PAGE_SIZE
-            return results
-        except Exception as exc:
-            logger.debug(traceback.format_exc())
-            logger.warning("Failed to fetch SSRS reports: %s", exc)
-        return []
+    def get_folders(self) -> Iterator[SsrsFolder]:
+        params = {
+            "$orderby": "Id",
+            "$select": FOLDER_SELECT_FIELDS,
+        }
+        for data in self._paginate("/Folders", params, "folders"):
+            yield from SsrsFolderListResponse(**data).value
+
+    def get_reports(self) -> Iterator[SsrsReport]:
+        params = {
+            "$orderby": "Id",
+            "$select": REPORT_SELECT_FIELDS,
+        }
+        for data in self._paginate("/Reports", params, "reports"):
+            yield from SsrsReportListResponse(**data).value

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

@@ -46,7 +46,7 @@ def test_connection(
 ) -> TestConnectionResult:
     test_fn = {
         "CheckAccess": client.test_access,
-        "GetDashboards": client.get_reports,
+        "GetDashboards": client.test_get_reports,
     }
 
     return test_connection_steps(

ingestion/src/metadata/ingestion/source/dashboard/ssrs/metadata.py

@@ -12,7 +12,7 @@
 SSRS source module
 """
 import traceback
-from typing import Any, Dict, Iterable, List, Optional
+from typing import Any, Dict, Iterable, Optional
 
 from metadata.generated.schema.api.data.createChart import CreateChartRequest
 from metadata.generated.schema.api.data.createDashboard import CreateDashboardRequest
@@ -77,13 +77,15 @@ def __init__(
         self.folder_path_map: Dict[str, str] = {}
 
     def prepare(self):
-        folders = self.client.get_folders()
-        self.folder_path_map = {folder.path: folder.name for folder in folders}
+        self.folder_path_map = {
+            folder.path: folder.name for folder in self.client.get_folders()
+        }
         return super().prepare()
 
-    def get_dashboards_list(self) -> Optional[List[SsrsReport]]:
-        reports = self.client.get_reports()
-        return [r for r in reports if not r.hidden]
+    def get_dashboards_list(self) -> Iterable[SsrsReport]:
+        for report in self.client.get_reports():
+            if not report.hidden:
+                yield report
 
     def get_dashboard_name(self, dashboard: SsrsReport) -> str:
         return dashboard.name

ingestion/src/metadata/ingestion/source/database/common_db_source.py

@@ -62,7 +62,6 @@
 from metadata.ingestion.models.patch_request import PatchedEntity, PatchRequest
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.connections import get_connection
-from metadata.ingestion.source.connections_utils import kill_active_connections
 from metadata.ingestion.source.database.database_service import DatabaseServiceSource
 from metadata.ingestion.source.database.sql_column_handler import SqlColumnHandlerMixin
 from metadata.ingestion.source.database.sqlalchemy_source import SqlAlchemySource
@@ -152,15 +151,42 @@ def set_inspector(self, database_name: str) -> None:
         :param database_name: new database to set
         """
 
-        kill_active_connections(self.engine)
+        self._release_engine()
         logger.info(f"Ingesting from database: {database_name}")
 
         new_service_connection = deepcopy(self.service_connection)
         new_service_connection.database = database_name
         self.engine = get_connection(new_service_connection)
+        self.session = create_and_bind_thread_safe_session(self.engine)
+        self.connection_obj = self.engine
 
-        self._connection_map = {}  # Lazy init as well
+    def _release_engine(self) -> None:
+        # Close fairies first so _ConnectionRecord drops its pool reference;
+        # dispose alone leaves them orphaned and causes _finalize_fairy
+        # RecursionErrors at GC time. Clearing _inspector_map is what
+        # actually frees Inspector.info_cache — dispose() does not.
+        if getattr(self, "engine", None) is None:
+            return
+        for conn in self._connection_map.values():
+            try:
+                conn.close()
+            except Exception:  # pylint: disable=broad-except
+                logger.debug("Connection already closed", exc_info=True)
+        self._connection_map = {}
         self._inspector_map = {}
+        session = getattr(self, "session", None)
+        if session is not None:
+            try:
+                session.remove()
+            except Exception:  # pylint: disable=broad-except
+                logger.debug("Session cleanup failed", exc_info=True)
+            self.session = None
+        try:
+            self.engine.dispose()
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.warning(f"Failed to dispose engine: {exc}")
+        self.engine = None
+        self.connection_obj = None
 
     def get_database_names(self) -> Iterable[str]:
         """
@@ -780,14 +806,10 @@ def inspector(self) -> Inspector:
         return self._inspector_map[thread_id]
 
     def close(self):
-        if self.connection is not None:
-            self.connection.close()
-        for connection in self._connection_map.values():
-            connection.close()
+        self._release_engine()
         if hasattr(self, "ssl_manager") and self.ssl_manager:
             self.ssl_manager = cast(SSLManager, self.ssl_manager)
             self.ssl_manager.cleanup_temp_files()
-        self.engine.dispose()
 
     def fetch_table_tags(
         self,