Function __unique_conn

tests/conftest.py:492–525

Connects to Impala and creates a new database, then returns a connection to it. This is intended to be used in a "with" block. Upon exit, the database will be dropped. A database name can be provided by 'db_name', a database by that name must not exist prior to calling this method.

(pytest_config, db_name=None, timeout=DEFAULT_CONN_TIMEOUT)

@contextlib.contextmanager
def __unique_conn(pytest_config, db_name=None, timeout=DEFAULT_CONN_TIMEOUT):
  """Connects to Impala and creates a new database, then returns a connection to it.
     This is intended to be used in a "with" block. Upon exit, the database will be
     dropped. A database name can be provided by 'db_name', a database by that name
     must not exist prior to calling this method.

     with __unique_conn() as conn:
       # Use conn
     # The database no longer exists and the conn is closed.

     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.
  """
  if not db_name:
    db_name = choice(ascii_lowercase) + "".join(sample(ascii_lowercase + digits, 5))
  with __auto_closed_conn(pytest_config, timeout=timeout) as conn:
    with __auto_closed_cursor(conn) as cur:
      cur.execute("CREATE DATABASE %s" % db_name)
  with __auto_closed_conn(pytest_config, db_name=db_name, timeout=timeout) as conn:
    try:
      yield conn
    finally:
      try:
        with __auto_closed_cursor(conn) as cur:
          try:
            cur.execute("USE DEFAULT")
            cur.execute("DROP DATABASE IF EXISTS %s CASCADE" % db_name)
          except Exception as e:
            LOG.warn("Error dropping database: %s", e)
      except Exception as e:
        LOG.warn("Error creating a cursor: %s", e)


@contextlib.contextmanager