Function conn

tests/conftest.py:453–479

Provides a new DB-API compliant connection to Impala as a pytest fixture. The same connection is used for all test methods in a class. The class may provide the following customizations: - get_db_name(): The name of the database to connect to. - auto_create_db(): If declared

(request)

@pytest.fixture(scope="class")
def conn(request):
  """Provides a new DB-API compliant connection to Impala as a pytest fixture. The
     same connection is used for all test methods in a class. The class may provide the
     following customizations:
       - get_db_name(): The name of the database to connect to.
       - auto_create_db(): If declared and the method returns True, the database will
         be created before tests run and dropped afterwards. If a database name is
         provided by get_db_name(), it must not exist. Classes that use both
         auto_create_db() and get_db_name() should generate a random name in
         get_db_name() and cache it.
       - get_conn_timeout(): The timeout, in seconds, to use for this connection.
     The returned connection will have a 'db_name' property.

     DEPRECATED:
     See the 'unique_database' fixture above to use Impala's custom python
     API instead of DB-API.
  """
  db_name = __call_cls_method_if_exists(request.cls, "get_db_name")
  use_unique_conn = __call_cls_method_if_exists(request.cls, "auto_create_db")
  timeout = __call_cls_method_if_exists(request.cls, "get_conn_timeout") or \
      DEFAULT_CONN_TIMEOUT
  if use_unique_conn:
    with __unique_conn(request.config, db_name=db_name, timeout=timeout) as conn:
      yield conn
  else:
    with __auto_closed_conn(request.config, db_name=db_name, timeout=timeout) as conn:
      yield conn


def __call_cls_method_if_exists(cls, method_name):