How to use test_connection_error method in locust

Best Python code snippet using locust

test_connection.py

Source:test_connection.py

1# Licensed to the Apache Software Foundation (ASF) under one2# or more contributor license agreements.  See the NOTICE file3# distributed with this work for additional information4# regarding copyright ownership.  The ASF licenses this file5# to you under the Apache License, Version 2.0 (the6# "License"); you may not use this file except in compliance7# with the License.  You may obtain a copy of the License at8#9#   http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing,12# software distributed under the License is distributed on an13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY14# KIND, either express or implied.  See the License for the15# specific language governing permissions and limitations16# under the License.17import logging18import sqlite319from contextlib import closing20from typing import Any, Dict, Optional21from flask import current_app as app22from flask_appbuilder.security.sqla.models import User23from flask_babel import gettext as _24from func_timeout import func_timeout, FunctionTimedOut25from sqlalchemy.engine.url import make_url26from sqlalchemy.exc import DBAPIError, NoSuchModuleError27from superset.commands.base import BaseCommand28from superset.databases.commands.exceptions import (29    DatabaseSecurityUnsafeError,30    DatabaseTestConnectionDriverError,31    DatabaseTestConnectionFailedError,32    DatabaseTestConnectionUnexpectedError,33)34from superset.databases.dao import DatabaseDAO35from superset.errors import ErrorLevel, SupersetErrorType36from superset.exceptions import SupersetSecurityException, SupersetTimeoutException37from superset.extensions import event_logger38from superset.models.core import Database39logger = logging.getLogger(__name__)40class TestConnectionDatabaseCommand(BaseCommand):41    def __init__(self, user: User, data: Dict[str, Any]):42        self._actor = user43        self._properties = data.copy()44        self._model: Optional[Database] = None45    def run(self) -> None:46        self.validate()47        uri = self._properties.get("sqlalchemy_uri", "")48        if self._model and uri == self._model.safe_sqlalchemy_uri():49            uri = self._model.sqlalchemy_uri_decrypted50        # context for error messages51        url = make_url(uri)52        context = {53            "hostname": url.host,54            "password": url.password,55            "port": url.port,56            "username": url.username,57            "database": url.database,58        }59        try:60            database = DatabaseDAO.build_db_for_connection_test(61                server_cert=self._properties.get("server_cert", ""),62                extra=self._properties.get("extra", "{}"),63                impersonate_user=self._properties.get("impersonate_user", False),64                encrypted_extra=self._properties.get("encrypted_extra", "{}"),65            )66            database.set_sqlalchemy_uri(uri)67            database.db_engine_spec.mutate_db_for_connection_test(database)68            username = self._actor.username if self._actor is not None else None69            engine = database.get_sqla_engine(user_name=username)70            event_logger.log_with_context(71                action="test_connection_attempt",72                engine=database.db_engine_spec.__name__,73            )74            with closing(engine.raw_connection()) as conn:75                try:76                    alive = func_timeout(77                        int(78                            app.config[79                                "TEST_DATABASE_CONNECTION_TIMEOUT"80                            ].total_seconds()81                        ),82                        engine.dialect.do_ping,83                        args=(conn,),84                    )85                except sqlite3.ProgrammingError:86                    # SQLite can't run on a separate thread, so ``func_timeout`` fails87                    alive = engine.dialect.do_ping(conn)88                except FunctionTimedOut as ex:89                    raise SupersetTimeoutException(90                        error_type=SupersetErrorType.CONNECTION_DATABASE_TIMEOUT,91                        message=(92                            "Please check your connection details and database settings, "93                            "and ensure that your database is accepting connections, "94                            "then try connecting again."95                        ),96                        level=ErrorLevel.ERROR,97                        extra={"sqlalchemy_uri": database.sqlalchemy_uri},98                    ) from ex99                except Exception:  # pylint: disable=broad-except100                    alive = False101                if not alive:102                    raise DBAPIError(None, None, None)103            # Log succesful connection test with engine104            event_logger.log_with_context(105                action="test_connection_success",106                engine=database.db_engine_spec.__name__,107            )108        except (NoSuchModuleError, ModuleNotFoundError) as ex:109            event_logger.log_with_context(110                action=f"test_connection_error.{ex.__class__.__name__}",111                engine=database.db_engine_spec.__name__,112            )113            raise DatabaseTestConnectionDriverError(114                message=_("Could not load database driver: {}").format(115                    database.db_engine_spec.__name__116                ),117            ) from ex118        except DBAPIError as ex:119            event_logger.log_with_context(120                action=f"test_connection_error.{ex.__class__.__name__}",121                engine=database.db_engine_spec.__name__,122            )123            # check for custom errors (wrong username, wrong password, etc)124            errors = database.db_engine_spec.extract_errors(ex, context)125            raise DatabaseTestConnectionFailedError(errors) from ex126        except SupersetSecurityException as ex:127            event_logger.log_with_context(128                action=f"test_connection_error.{ex.__class__.__name__}",129                engine=database.db_engine_spec.__name__,130            )131            raise DatabaseSecurityUnsafeError(message=str(ex)) from ex132        except SupersetTimeoutException as ex:133            event_logger.log_with_context(134                action=f"test_connection_error.{ex.__class__.__name__}",135                engine=database.db_engine_spec.__name__,136            )137            # bubble up the exception to return a 408138            raise ex139        except Exception as ex:140            event_logger.log_with_context(141                action=f"test_connection_error.{ex.__class__.__name__}",142                engine=database.db_engine_spec.__name__,143            )144            errors = database.db_engine_spec.extract_errors(ex, context)145            raise DatabaseTestConnectionUnexpectedError(errors) from ex146    def validate(self) -> None:147        database_name = self._properties.get("database_name")148        if database_name is not None:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.