sqlalchemy.testing.provision

module documentation
(source)

Undocumented

Class register 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 drop_materialized_views Undocumented
Function drop_views 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 normalize_sequence Normalize sequence parameters for dialect that don't start with 1 by default.
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.
Function upsert return the backends insert..on conflict / on dupe etc. construct.
Constant FOLLOWER_IDENT Undocumented
Variable log Undocumented
Function _adapt_update_db_opts Undocumented
Function _configs_for_db_operation Undocumented
Function _generate_driver_urls Undocumented
@register.init
def configure_follower(cfg, ident): (source)

Create dialect-specific config settings for a follower database.

@register.init
def create_db(cfg, eng, ident): (source)

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): (source)

Undocumented

def drop_all_schema_objects(cfg, eng): (source)

Undocumented

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

Undocumented

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

Undocumented

@register.init
def drop_db(cfg, eng, ident): (source)

Drop a database that we dynamically created for testing.

def drop_follower_db(follower_ident): (source)

Undocumented

def drop_materialized_views(cfg, eng): (source)

Undocumented

def drop_views(cfg, eng): (source)

Undocumented

@register.init
def follower_url_from_main(url, ident): (source)

Create a connection URL for a dynamically-created test database. :param url: the connection URL specified when the test run was invoked :param ident: the pytest-xdist "worker identifier" to be used as the database name

def generate_db_urls(db_urls, extra_drivers): (source)

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): (source)

Undocumented

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

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 normalize_sequence(cfg, sequence): (source)

Normalize sequence parameters for dialect that don't start with 1 by default. The default implementation does nothing

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

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): (source)

Undocumented

def reap_dbs(idents_file): (source)

Undocumented

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

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): (source)

Undocumented

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

Undocumented

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

Undocumented

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

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_decorator(_adapt_update_db_opts)
def update_db_opts(db_url, db_opts, options): (source)

Set database options (db_opts) for a test database that we created.

@register.init
def upsert(cfg, table, returning, set_lambda=None): (source)

return the backends insert..on conflict / on dupe etc. construct. while we should add a backend-neutral upsert construct as well, such as insert().upsert(), it's important that we continue to test the backend-specific insert() constructs since if we do implement insert().upsert(), that would be using a different codepath for the things we need to test like insertmanyvalues, etc.

FOLLOWER_IDENT = (source)

Undocumented

Value
None
log = (source)

Undocumented

def _adapt_update_db_opts(fn): (source)

Undocumented

def _configs_for_db_operation(): (source)

Undocumented

def _generate_driver_urls(url, extra_drivers): (source)

Undocumented