module documentation

Undocumented

Constant FOLLOWER​_IDENT Undocumented
Variable log Undocumented
Class register Undocumented
Function ​_configs​_for​_db​_operation Undocumented
Function ​_generate​_driver​_urls Undocumented
Function configure​_follower Create dialect-specific config settings for a follower database.
Function create​_db Dynamically create a database for testing.
Function create​_follower​_db Undocumented
Function drop​_all​_schema​_objects Undocumented
Function drop​_all​_schema​_objects​_post​_tables Undocumented
Function drop​_all​_schema​_objects​_pre​_tables Undocumented
Function drop​_db Drop a database that we dynamically created for testing.
Function drop​_follower​_db Undocumented
Function follower​_url​_from​_main Create a connection URL for a dynamically-created test database.
Function generate​_db​_urls Generate a set of URLs to test given configured URLs plus additional driver names.
Function generate​_driver​_url Undocumented
Function get​_temp​_table​_name Specify table name for creating a temporary Table.
Function post​_configure​_engine Perform extra steps after configuring an engine for testing.
Function prepare​_for​_drop​_tables Undocumented
Function reap​_dbs Undocumented
Function run​_reap​_dbs Remove databases that were created during the test process, after the process has ended.
Function set​_default​_schema​_on​_connection Undocumented
Function setup​_config Undocumented
Function stop​_test​_class​_outside​_fixtures Undocumented
Function temp​_table​_keyword​_args Specify keyword arguments for creating a temporary Table.
Function update​_db​_opts Set database options (db_opts) for a test database that we created.
FOLLOWER_IDENT =

Undocumented

Value
None
log =

Undocumented

def _configs_for_db_operation():

Undocumented

def _generate_driver_urls(url, extra_drivers):

Undocumented

@register.init
def configure_follower(cfg, ident):
Create dialect-specific config settings for a follower database.
@register.init
def create_db(cfg, eng, ident):

Dynamically create a database for testing.

Used when a test run will employ multiple processes, e.g., when run via tox or pytest -n4.

def create_follower_db(follower_ident):

Undocumented

def drop_all_schema_objects(cfg, eng):

Undocumented

@register.init
def drop_all_schema_objects_post_tables(cfg, eng):

Undocumented

@register.init
def drop_all_schema_objects_pre_tables(cfg, eng):

Undocumented

@register.init
def drop_db(cfg, eng, ident):
Drop a database that we dynamically created for testing.
def drop_follower_db(follower_ident):

Undocumented

@register.init
def follower_url_from_main(url, ident):
Create a connection URL for a dynamically-created test database.
Parameters
urlthe connection URL specified when the test run was invoked
identthe pytest-xdist "worker identifier" to be used as the database name
def generate_db_urls(db_urls, extra_drivers):

Generate a set of URLs to test given configured URLs plus additional driver names.

Given:

--dburi postgresql://db1          --dburi postgresql://db2          --dburi postgresql://db2          --dbdriver=psycopg2 --dbdriver=asyncpg?async_fallback=true

Noting that the default postgresql driver is psycopg2, the output would be:

postgresql+psycopg2://db1
postgresql+asyncpg://db1
postgresql+psycopg2://db2
postgresql+psycopg2://db3

That is, for the driver in a --dburi, we want to keep that and use that driver for each URL it's part of . For a driver that is only in --dbdrivers, we want to use it just once for one of the URLs. for a driver that is both coming from --dburi as well as --dbdrivers, we want to keep it in that dburi.

Driver specific query options can be specified by added them to the driver name. For example, to enable the async fallback option for asyncpg:

--dburi postgresql://db1          --dbdriver=asyncpg?async_fallback=true
@register.init
def generate_driver_url(url, driver, query_str):

Undocumented

@register.init
def get_temp_table_name(cfg, eng, base_name):

Specify table name for creating a temporary Table.

Dialect-specific implementations of this method will return the name to use when creating a temporary table for testing, e.g., in the define_temp_tables method of the ComponentReflectionTest class in suite/test_reflection.py

Default to just the base name since that's what most dialects will use. The mssql dialect's implementation will need a "#" prepended.

@register.init
def post_configure_engine(url, engine, follower_ident):

Perform extra steps after configuring an engine for testing.

(For the internal dialects, currently only used by sqlite, oracle)

@register.init
def prepare_for_drop_tables(config, connection):

Undocumented

def reap_dbs(idents_file):

Undocumented

@register.init
def run_reap_dbs(url, ident):

Remove databases that were created during the test process, after the process has ended.

This is an optional step that is invoked for certain backends that do not reliably release locks on the database as long as a process is still in use. For the internal dialects, this is currently only necessary for mssql and oracle.

@register.init
def set_default_schema_on_connection(cfg, dbapi_connection, schema_name):

Undocumented

def setup_config(db_url, options, file_config, follower_ident):

Undocumented

@register.init
def stop_test_class_outside_fixtures(config, db, testcls):

Undocumented

@register.init
def temp_table_keyword_args(cfg, eng):

Specify keyword arguments for creating a temporary Table.

Dialect-specific implementations of this method will return the kwargs that are passed to the Table method when creating a temporary table for testing, e.g., in the define_temp_tables method of the ComponentReflectionTest class in suite/test_reflection.py

@register.init
def update_db_opts(db_url, db_opts):
Set database options (db_opts) for a test database that we created.