refactor(errors): register SSH tunnel error as its own errorhandler (SC-115347)

Self-review on PR #42538 flagged that the isinstance check inside
show_unexpected_exception was inconsistent with this file's own pattern,
where every other specific exception type (SupersetErrorException,
CSRFError, HTTPException, CommandException, ...) gets its own
@app.errorhandler registration rather than a branch inside the catch-all.

Flask's error handler dispatch walks the exception's MRO to find the most
specific registered handler regardless of registration order, so
@app.errorhandler(sshtunnel.BaseSSHTunnelForwarderError) is dispatched
ahead of @app.errorhandler(Exception) with no behavior change. Verified by
rerunning all 3 tests in tests/unit_tests/views/test_error_handling.py.

Co-Authored-By: Claude <noreply@anthropic.com>

superset/views/error_handling.py

@@ -23,12 +23,14 @@ import typing
 from importlib.resources import files
 from typing import Any, Callable, cast
 
+import sshtunnel
 from flask import (
     Flask,
     request,
     Response,
     send_file,
 )
+from flask_babel import gettext as _
 from flask_wtf.csrf import CSRFError
 from sqlalchemy import exc
 from werkzeug.exceptions import HTTPException
@@ -86,6 +88,29 @@ def json_error_response(
     )
 
 
+def handle_ssh_tunnel_error(ex: sshtunnel.BaseSSHTunnelForwarderError) -> FlaskResponse:
+    """
+    Build the structured response for an unreachable/misconfigured SSH tunnel
+    gateway. This is an expected environmental failure (analogous to a
+    database connection failure), so it is logged at WARNING rather than
+    ERROR to avoid alarm fatigue on otherwise-actionable alerting.
+    """
+    logger.warning("BaseSSHTunnelForwarderError", exc_info=True)
+    return json_error_response(
+        [
+            SupersetError(
+                message=_(
+                    "Failed to establish an SSH tunnel to the database: %(reason)s",
+                    reason=str(ex),
+                ),
+                error_type=SupersetErrorType.CONNECTION_HOST_DOWN_ERROR,
+                level=ErrorLevel.WARNING,
+            ),
+        ],
+        status=400,
+    )
+
+
 def handle_api_exception(
     f: Callable[..., FlaskResponse],
 ) -> Callable[..., FlaskResponse]:
@@ -121,6 +146,8 @@ def handle_api_exception(
         except (exc.IntegrityError, exc.DatabaseError, exc.DataError) as ex:
             logger.exception(ex)
             return json_error_response(utils.error_msg_from_exception(ex), status=422)
+        except sshtunnel.BaseSSHTunnelForwarderError as ex:
+            return handle_ssh_tunnel_error(ex)
         except Exception as ex:  # pylint: disable=broad-except
             logger.exception(ex)
             return json_error_response(utils.error_msg_from_exception(ex))
@@ -211,6 +238,12 @@ def set_app_error_handlers(app: Flask) -> None:  # noqa: C901
             status=ex.status,
         )
 
+    @app.errorhandler(sshtunnel.BaseSSHTunnelForwarderError)
+    def show_ssh_tunnel_error(
+        ex: sshtunnel.BaseSSHTunnelForwarderError,
+    ) -> FlaskResponse:
+        return handle_ssh_tunnel_error(ex)
+
     @app.errorhandler(Exception)
     @app.errorhandler(500)
     def show_unexpected_exception(ex: Exception) -> FlaskResponse:

tests/unit_tests/views/test_error_handling.py

@@ -0,0 +1,112 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+import logging
+from typing import cast
+
+import pytest
+import sshtunnel
+from flask import Flask, Response
+from flask_babel import Babel
+
+from superset.errors import SupersetErrorType
+from superset.superset_typing import FlaskResponse
+from superset.utils import json
+from superset.views.error_handling import handle_api_exception, set_app_error_handlers
+
+
+class TestHandleApiExceptionSSHTunnelError:
+    def test_returns_400_with_connection_host_down_error_and_no_error_log(
+        self, app, caplog: pytest.LogCaptureFixture
+    ):
+        @handle_api_exception
+        def view(self: object) -> FlaskResponse:
+            raise sshtunnel.BaseSSHTunnelForwarderError(
+                "Could not establish session to SSH gateway"
+            )
+
+        with app.test_request_context():
+            with caplog.at_level(logging.WARNING):
+                response = cast(Response, view(self=object()))
+
+        assert response.status_code == 400
+        payload = json.loads(response.data)
+        assert (
+            payload["errors"][0]["error_type"]
+            == SupersetErrorType.CONNECTION_HOST_DOWN_ERROR.value
+        )
+        assert not any(record.levelno >= logging.ERROR for record in caplog.records)
+        assert any(
+            record.levelno == logging.WARNING
+            and "BaseSSHTunnelForwarderError" in record.message
+            for record in caplog.records
+        )
+
+
+class TestShowUnexpectedException:
+    def _build_app_with_handlers(self) -> Flask:
+        # A fresh, minimal Flask app per test: `set_app_error_handlers` can
+        # only register handlers before the app has served its first
+        # request, so it can't share the module-scoped `app` fixture across
+        # tests in this class.
+        test_app = Flask(__name__)
+        test_app.config["DEBUG"] = False
+        Babel(test_app)
+        set_app_error_handlers(test_app)
+
+        @test_app.route("/ssh-tunnel-error")
+        def ssh_tunnel_error_view() -> FlaskResponse:
+            raise sshtunnel.BaseSSHTunnelForwarderError(
+                "Could not establish session to SSH gateway"
+            )
+
+        @test_app.route("/generic-error")
+        def generic_error_view() -> FlaskResponse:
+            raise ValueError("boom")
+
+        return test_app
+
+    def test_ssh_tunnel_error_returns_structured_400(
+        self, caplog: pytest.LogCaptureFixture
+    ):
+        client = self._build_app_with_handlers().test_client()
+
+        with caplog.at_level(logging.WARNING):
+            response = client.get("/ssh-tunnel-error")
+
+        assert response.status_code == 400
+        payload = json.loads(response.data)
+        assert (
+            payload["errors"][0]["error_type"]
+            == SupersetErrorType.CONNECTION_HOST_DOWN_ERROR.value
+        )
+        assert not any(record.levelno >= logging.ERROR for record in caplog.records)
+
+    def test_generic_exception_still_returns_original_500_shape(
+        self, caplog: pytest.LogCaptureFixture
+    ):
+        client = self._build_app_with_handlers().test_client()
+
+        with caplog.at_level(logging.WARNING):
+            response = client.get("/generic-error")
+
+        assert response.status_code == 500
+        payload = json.loads(response.data)
+        assert (
+            payload["errors"][0]["error_type"]
+            == SupersetErrorType.GENERIC_BACKEND_ERROR.value
+        )
+        assert any(record.levelno >= logging.ERROR for record in caplog.records)