/
opt
/
alt
/
python27
/
lib64
/
python2.7
/
site-packages
/
sqlalchemy
/
testing
/
up file
home
� 1��fc @ s- d d l Z d d l Z d d l m Z d d l m Z d d l m Z d d l m Z d d l m Z d d l m Z d d l m Z d d l m Z d d l m Z e j e � Z d Z d e f d � � YZ d � Z d � Z d � Z d � Z d � Z e j d � � Z d � Z e j d � � Z e j d � � Z d � Z e j d � � Z! e j d � � Z" e j d � � Z# e j d � � Z$ e j d � � Z% e j d � � Z& e j d � � Z' d � Z( e j d! � � Z) e j d" � � Z* e j d# � � Z+ e j d$ � � Z, e j d% � � Z- d S(& i����Ni ( t config( t engines( t utili ( t exc( t inspect( t url( t ddl( t schema( t compatt registerc B s2 e Z d � Z e d � � Z d � Z d � Z RS( c C s i | _ d S( N( t fns( t self( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt __init__ s c C s t � j d � | � S( Nt *( R t for_db( t clst fn( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt init s c s � � f d � } | S( Nc s" x � D] } | � j | <q W� S( N( R ( R t dbname( t dbnamesR ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt decorate# s ( ( R R R ( ( R R sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR " s c G s� t | t j � r$ t j | � } n' t | t j � r? | } n | j j } | j � } | | j k rz | j | | | � S| j d | | � Sd S( NR ( t isinstanceR t string_typest sa_urlt make_urlt URLt dbR t get_backend_nameR ( R t cfgt argR t backend( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt __call__* s ( t __name__t __module__R t classmethodR R R ( ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR s c C sD x= t � D]2 } t j d | | j j � t | | j | � q Wd S( Ns CREATE database %s, URI %r( t _configs_for_db_operationt logt infoR R t create_db( t follower_identR ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt create_follower_db8 s c C s� t j | � j � } | j � | r7 t | | � } n i } t | | � d | d <t j | | � } t | | | � | j � j � t j j | | | | � } | r� | t _ n | r� t | | � n | S( Nt globalt scope( R R t get_dialectt load_provisioningt follower_url_from_maint update_db_optsR t testing_enginet post_configure_enginet connectt closeR t ConfigR t identt configure_follower( t db_urlt optionst file_configR' t dialectt db_optst engR ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt setup_config> s c C sD x= t � D]2 } t j d | | j j � t | | j | � q Wd S( Ns DROP database %s, URI %r( R# R$ R% R R t drop_db( R' R ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt drop_follower_dbZ s c c s3 t � } t j t � } g g | D] } t j | � ^ q" D] } | | j � f ^ q; } x+ | D]# \ } } | | j j | j � q` Wi } x� | D]� \ } } | j } | j � | | k r� t | � j | | � | | <} n | | } x; t | | � D]* } | | k rq� n | j | � | Vq� Wq� Wd S( s& 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 N( t sett collectionst defaultdictR R R+ t namet addt driverR, t differencet _generate_driver_urls( t db_urlst extra_driverst urlst! backend_to_driver_we_already_haveR6 t url_objt urls_plus_dialectsR9 t backend_to_driver_we_needR t extra_per_backendt driver_url( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt generate_db_urls` s( ! > c c s� | j � } | j | � t | | d � } t | � Vxz t | � D]l } d | k rp | j d d � \ } } n | } d } t | | | � } | rC | j | � t | � VqC qC Wd S( Nt t ?i ( t get_driver_namet discardt generate_driver_urlt strt listt splitt Nonet remove( R RH t main_drivert drvt driver_onlyt query_strt new_url( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyRF � s c C sn | j � } | j d d | | f � } | r@ | j | � } n y | j � Wn t j k re d SX| Sd S( Nt drivernames %s+%s( R R? t update_query_stringR+ R t NoSuchModuleErrorRY ( R RD R^ R R_ ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyRU � s c c s� t � } x$ t j j � D] } | j j � q Wxk t j j � D]Z } | j j } | j � } | | j | j | j f } | | k r@ | V| j | � q@ q@ Wx$ t j j � D] } | j j � q� Wd S( N( R? R R3 t all_configsR t disposeR R t usernamet hostt databaseRC ( t hostsR R R t host_conf( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR# � s c C s d S( N( ( R R; ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt" drop_all_schema_objects_pre_tables� s c C s d S( N( ( R R; ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt# drop_all_schema_objects_post_tables� s c C sf t | | � t | � } y | j � } Wn t k r< nO X| j � �@ } x6 | D]. } | j t j t j | t j � � � � qS WWd QXt j j j | � r!y | j d d � } Wn t k r� q!X| j � �F } x< | D]4 } | j t j t j | t j � d d �� � q� WWd QXn t j | | � t j j j | � r{t j | | d | j �t j | | d | j �n t | | � t j j j | � rb| j � �� } x3 | j � D]% } | j t j t j | � � � q�Wt j j j | � rYx_ | j | j g D]H } x? | j d | � D]+ } | j t j t j | d | �� � q#Wq Wn Wd QXn d S( NR t test_schema( Rj R t get_view_namest NotImplementedErrort begint executeR t _DropViewR t Tablet MetaDataR t requirementst schemast enabled_for_configR t drop_all_tablesRl t test_schema_2Rk t sequencest get_sequence_namest DropSequencet Sequence( R R; t inspectort view_namest connt vnamet seqt schema_name( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt drop_all_schema_objects� sN , # c C s t d | j f � � d S( s� Dynamically create a database for testing. Used when a test run will employ multiple processes, e.g., when run via `tox` or `pytest -n4`. s" no DB creation routine for cfg: %sN( Rn R ( R R; R4 ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR& ! s c C s t d | j f � � d S( s8 Drop a database that we dynamically created for testing.s no DB drop routine for cfg: %sN( Rn R ( R R; R4 ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR= - s c C s d S( sC Set database options (db_opts) for a test database that we created.N( ( R6 R: ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR. 3 s c C s d S( s� Perform extra steps after configuring an engine for testing. (For the internal dialects, currently only used by sqlite, oracle, mssql) N( ( R t engineR' ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR0 9 s c C s t j | � } | j d | � S( s� 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 Rg ( R R R? ( R R4 ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR- B s c C s d S( s@ Create dialect-specific config settings for a follower database.N( ( R R4 ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyR5 N s c C s d S( s[ 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. N( ( R R4 ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt run_reap_dbsT s c C s+ t j d � t j t � } t j t � } i } t | � �� } x� | D]� } | j � } | j d � \ } } t j | � } | | k r� | j � | | <| | j � n | j � | j f } | | j | � | | j | � qG WWd QXx9 | D]1 } t | | � d } | | } t | | � q� Wd S( Ns Reaping databases...t i ( R$ R% R@ RA R? t opent stripRX R R R+ R, R Rf RC RW R� ( t idents_fileRI t identst dialectst file_t linet db_nameR6 RK t url_keyR R4 ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt reap_dbsa s& c C s t d | j f � � d S( sO 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 s. no temp table keyword args routine for cfg: %sN( Rn R ( R R; ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt temp_table_keyword_argsz s c C s d S( N( ( R t connection( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt prepare_for_drop_tables� s c C s d S( N( ( R R t testcls( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt stop_test_class_outside_fixtures� s c C s | S( s� 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. ( ( R R; t base_name( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt get_temp_table_name� s c C s t d | j j f � � d S( Ns9 backend does not implement a schema name set function: %s( Rn R R ( R t dbapi_connectionR� ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt set_default_schema_on_connection� s (. R@ t loggingRQ R R R R R R� R R t sqlR R R t getLoggerR R$ RY t FOLLOWER_IDENTt objectR R( R< R> RP RF R RU R# Rj Rk R� R& R= R. R0 R- R5 R� R� R� R� R� R� R� ( ( ( sO /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing/provision.pyt <module> sH A 9