<s> import sys <EOL> import time <EOL> from django . apps import apps <EOL> from django . conf import settings <EOL> from django . core import serializers <EOL> from django . db import router <EOL> from django . utils . six import StringIO <EOL> from django . utils . six . moves import input <EOL> TEST_DATABASE_PREFIX = '<STR_LIT>' <EOL> class BaseDatabaseCreation ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , connection ) : <EOL> self . connection = connection <EOL> @ property <EOL> def _nodb_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . connection . _nodb_connection <EOL> def create_test_db ( self , verbosity = <NUM_LIT:1> , autoclobber = False , serialize = True , keepdb = False ) : <EOL> """<STR_LIT>""" <EOL> from django . core . management import call_command <EOL> test_database_name = self . _get_test_db_name ( ) <EOL> if verbosity >= <NUM_LIT:1> : <EOL> action = '<STR_LIT>' <EOL> if keepdb : <EOL> action = "<STR_LIT>" <EOL> print ( "<STR_LIT>" % ( <EOL> action , <EOL> self . _get_database_display_str ( verbosity , test_database_name ) , <EOL> ) ) <EOL> self . _create_test_db ( verbosity , autoclobber , keepdb ) <EOL> self . connection . close ( ) <EOL> settings . DATABASES [ self . connection . alias ] [ "<STR_LIT>" ] = test_database_name <EOL> self . connection . settings_dict [ "<STR_LIT>" ] = test_database_name <EOL> call_command ( <EOL> '<STR_LIT>' , <EOL> verbosity = max ( verbosity - <NUM_LIT:1> , <NUM_LIT:0> ) , <EOL> interactive = False , <EOL> database = self . connection . alias , <EOL> run_syncdb = True , <EOL> ) <EOL> if serialize : <EOL> self . connection . _test_serialized_contents = self . serialize_db_to_string ( ) <EOL> call_command ( '<STR_LIT>' , database = self . connection . alias ) <EOL> self . connection . ensure_connection ( ) <EOL> return test_database_name <EOL> def set_as_test_mirror ( self , primary_settings_dict ) : <EOL> """<STR_LIT>""" <EOL> self . connection . settings_dict [ '<STR_LIT>' ] = primary_settings_dict [ '<STR_LIT>' ] <EOL> def serialize_db_to_string ( self ) : <EOL> """<STR_LIT>""" <EOL> from django . db . migrations . loader import MigrationLoader <EOL> loader = MigrationLoader ( self . connection ) <EOL> app_list = [ ] <EOL> for app_config in apps . get_app_configs ( ) : <EOL> if ( <EOL> app_config . models_module is not None and <EOL> app_config . label in loader . migrated_apps and <EOL> app_config . name not in settings . TEST_NON_SERIALIZED_APPS <EOL> ) : <EOL> app_list . append ( ( app_config , None ) ) <EOL> def get_objects ( ) : <EOL> for model in serializers . sort_dependencies ( app_list ) : <EOL> if ( model . _meta . can_migrate ( self . connection ) and <EOL> router . allow_migrate_model ( self . connection . alias , model ) ) : <EOL> queryset = model . _default_manager . using ( self . connection . alias ) . order_by ( model . _meta . pk . name ) <EOL> for obj in queryset . iterator ( ) : <EOL> yield obj <EOL> out = StringIO ( ) <EOL> serializers . serialize ( "<STR_LIT>" , get_objects ( ) , indent = None , stream = out ) <EOL> return out . getvalue ( ) <EOL> def deserialize_db_from_string ( self , data ) : <EOL> """<STR_LIT>""" <EOL> data = StringIO ( data ) <EOL> for obj in serializers . deserialize ( "<STR_LIT>" , data , using = self . connection . alias ) : <EOL> obj . save ( ) <EOL> def _get_database_display_str ( self , verbosity , database_name ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> self . connection . alias , <EOL> ( "<STR_LIT>" % database_name ) if verbosity >= <NUM_LIT:2> else '<STR_LIT>' , <EOL> ) <EOL> def _get_test_db_name ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . connection . settings_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> return self . connection . settings_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> return TEST_DATABASE_PREFIX + self . connection . settings_dict [ '<STR_LIT>' ] <EOL> def _create_test_db ( self , verbosity , autoclobber , keepdb = False ) : <EOL> """<STR_LIT>""" <EOL> suffix = self . sql_table_creation_suffix ( ) <EOL> test_database_name = self . _get_test_db_name ( ) <EOL> qn = self . connection . ops . quote_name <EOL> with self . _nodb_connection . cursor ( ) as cursor : <EOL> try : <EOL> cursor . execute ( <EOL> "<STR_LIT>" % ( qn ( test_database_name ) , suffix ) ) <EOL> except Exception as e : <EOL> if keepdb : <EOL> return test_database_name <EOL> sys . stderr . write ( <EOL> "<STR_LIT>" % e ) <EOL> if not autoclobber : <EOL> confirm = input ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % test_database_name ) <EOL> if autoclobber or confirm == '<STR_LIT:yes>' : <EOL> try : <EOL> if verbosity >= <NUM_LIT:1> : <EOL> print ( "<STR_LIT>" % ( <EOL> self . _get_database_display_str ( verbosity , test_database_name ) , <EOL> ) ) <EOL> cursor . execute ( <EOL> "<STR_LIT>" % qn ( test_database_name ) ) <EOL> cursor . execute ( <EOL> "<STR_LIT>" % ( qn ( test_database_name ) , <EOL> suffix ) ) <EOL> except Exception as e : <EOL> sys . stderr . write ( <EOL> "<STR_LIT>" % e ) <EOL> sys . exit ( <NUM_LIT:2> ) <EOL> else : <EOL> print ( "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> return test_database_name <EOL> def clone_test_db ( self , number , verbosity = <NUM_LIT:1> , autoclobber = False , keepdb = False ) : <EOL> """<STR_LIT>""" <EOL> source_database_name = self . connection . settings_dict [ '<STR_LIT>' ] <EOL> if verbosity >= <NUM_LIT:1> : <EOL> action = '<STR_LIT>' <EOL> if keepdb : <EOL> action = '<STR_LIT>' <EOL> print ( "<STR_LIT>" % ( <EOL> action , <EOL> self . _get_database_display_str ( verbosity , source_database_name ) , <EOL> ) ) <EOL> self . _clone_test_db ( number , verbosity , keepdb ) <EOL> def get_test_db_clone_settings ( self , number ) : <EOL> """<STR_LIT>""" <EOL> orig_settings_dict = self . connection . settings_dict <EOL> new_settings_dict = orig_settings_dict . copy ( ) <EOL> new_settings_dict [ '<STR_LIT>' ] = '<STR_LIT>' . format ( orig_settings_dict [ '<STR_LIT>' ] , number ) <EOL> return new_settings_dict <EOL> def _clone_test_db ( self , number , verbosity , keepdb = False ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def destroy_test_db ( self , old_database_name = None , verbosity = <NUM_LIT:1> , keepdb = False , number = None ) : <EOL> """<STR_LIT>""" <EOL> self . connection . close ( ) <EOL> if number is None : <EOL> test_database_name = self . connection . settings_dict [ '<STR_LIT>' ] <EOL> else : <EOL> test_database_name = self . get_test_db_clone_settings ( number ) [ '<STR_LIT>' ] <EOL> if verbosity >= <NUM_LIT:1> : <EOL> action = '<STR_LIT>' <EOL> if keepdb : <EOL> action = '<STR_LIT>' <EOL> print ( "<STR_LIT>" % ( <EOL> action , <EOL> self . _get_database_display_str ( verbosity , test_database_name ) , <EOL> ) ) <EOL> if not keepdb : <EOL> self . _destroy_test_db ( test_database_name , verbosity ) <EOL> if old_database_name is not None : <EOL> settings . DATABASES [ self . connection . alias ] [ "<STR_LIT>" ] = old_database_name <EOL> self . connection . settings_dict [ "<STR_LIT>" ] = old_database_name <EOL> def _destroy_test_db ( self , test_database_name , verbosity ) : <EOL> """<STR_LIT>""" <EOL> with self . connection . _nodb_connection . cursor ( ) as cursor : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> cursor . execute ( "<STR_LIT>" <EOL> % self . connection . ops . quote_name ( test_database_name ) ) <EOL> def sql_table_creation_suffix ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' <EOL> def test_db_signature ( self ) : <EOL> """<STR_LIT>""" <EOL> settings_dict = self . connection . settings_dict <EOL> return ( <EOL> settings_dict [ '<STR_LIT>' ] , <EOL> settings_dict [ '<STR_LIT>' ] , <EOL> settings_dict [ '<STR_LIT>' ] , <EOL> settings_dict [ '<STR_LIT>' ] <EOL> ) </s>
