<s> import os <EOL> import subprocess <EOL> import sqlite3 <EOL> import pytest <EOL> from django . conf import settings <EOL> from django . db import connection <EOL> from django . db import transaction <EOL> from . compat import force_text <EOL> from . app . models import Item <EOL> DB_NAME = settings . DATABASES [ '<STR_LIT:default>' ] [ '<STR_LIT>' ] <EOL> if DB_NAME == '<STR_LIT>' : <EOL> TEST_DB_NAME = DB_NAME <EOL> else : <EOL> DB_NAME += '<STR_LIT>' <EOL> TEST_DB_NAME = '<STR_LIT>' + DB_NAME <EOL> def get_db_engine ( ) : <EOL> from django . conf import settings <EOL> return settings . DATABASES [ '<STR_LIT:default>' ] [ '<STR_LIT>' ] . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:1> ] <EOL> class CmdResult ( object ) : <EOL> def __init__ ( self , status_code , std_out , std_err ) : <EOL> self . status_code = status_code <EOL> self . std_out = std_out <EOL> self . std_err = std_err <EOL> def run_cmd ( * args ) : <EOL> r = subprocess . Popen ( args , stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> stdoutdata , stderrdata = r . communicate ( ) <EOL> ret = r . wait ( ) <EOL> return CmdResult ( ret , stdoutdata , stderrdata ) <EOL> def run_mysql ( * args ) : <EOL> from django . conf import settings <EOL> user = settings . DATABASES [ '<STR_LIT:default>' ] . get ( '<STR_LIT>' , None ) <EOL> if user : <EOL> args = ( '<STR_LIT>' , user ) + tuple ( args ) <EOL> args = ( '<STR_LIT>' , ) + tuple ( args ) <EOL> return run_cmd ( * args ) <EOL> def skip_if_sqlite_in_memory ( ) : <EOL> from django . conf import settings <EOL> if settings . DATABASES [ '<STR_LIT:default>' ] [ '<STR_LIT>' ] == '<STR_LIT>' and settings . DATABASES [ '<STR_LIT:default>' ] [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> pytest . skip ( '<STR_LIT>' ) <EOL> def create_empty_production_database ( ) : <EOL> drop_database ( name = DB_NAME ) <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_cmd ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:-c>' , '<STR_LIT>' % DB_NAME ) <EOL> assert ( '<STR_LIT>' in force_text ( r . std_out ) or <EOL> '<STR_LIT>' in force_text ( r . std_err ) ) <EOL> return <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_mysql ( '<STR_LIT>' , '<STR_LIT>' % DB_NAME ) <EOL> assert ( r . status_code == <NUM_LIT:0> or <EOL> '<STR_LIT>' in force_text ( r . std_out ) or <EOL> '<STR_LIT>' in force_text ( r . std_err ) ) <EOL> return <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> if DB_NAME == '<STR_LIT>' : <EOL> raise AssertionError ( <EOL> '<STR_LIT>' ) <EOL> open ( DB_NAME , '<STR_LIT:a>' ) . close ( ) <EOL> return <EOL> raise AssertionError ( '<STR_LIT>' % get_db_engine ( ) ) <EOL> def drop_database ( name = TEST_DB_NAME , suffix = None ) : <EOL> assert bool ( name ) ^ bool ( suffix ) , '<STR_LIT>' <EOL> if suffix : <EOL> name = '<STR_LIT>' % ( name , suffix ) <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_cmd ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:-c>' , '<STR_LIT>' % name ) <EOL> assert ( '<STR_LIT>' in force_text ( r . std_out ) or <EOL> '<STR_LIT>' in force_text ( r . std_err ) ) <EOL> return <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_mysql ( '<STR_LIT>' , '<STR_LIT>' % name ) <EOL> assert ( '<STR_LIT>' in force_text ( r . std_err ) or <EOL> r . status_code == <NUM_LIT:0> ) <EOL> return <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> if name == '<STR_LIT>' : <EOL> raise AssertionError ( <EOL> '<STR_LIT>' ) <EOL> if os . path . exists ( name ) : <EOL> os . unlink ( name ) <EOL> return <EOL> raise AssertionError ( '<STR_LIT>' % get_db_engine ( ) ) <EOL> def db_exists ( db_suffix = None ) : <EOL> name = TEST_DB_NAME <EOL> if db_suffix : <EOL> name = '<STR_LIT>' % ( name , db_suffix ) <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_cmd ( '<STR_LIT>' , name , '<STR_LIT:-c>' , '<STR_LIT>' ) <EOL> return r . status_code == <NUM_LIT:0> <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_mysql ( name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> return r . status_code == <NUM_LIT:0> <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> if TEST_DB_NAME == '<STR_LIT>' : <EOL> raise AssertionError ( <EOL> '<STR_LIT>' ) <EOL> return os . path . exists ( name ) <EOL> raise AssertionError ( '<STR_LIT>' % get_db_engine ( ) ) <EOL> def mark_database ( ) : <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_cmd ( '<STR_LIT>' , TEST_DB_NAME , '<STR_LIT:-c>' , '<STR_LIT>' ) <EOL> assert r . status_code == <NUM_LIT:0> <EOL> return <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_mysql ( TEST_DB_NAME , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert r . status_code == <NUM_LIT:0> <EOL> return <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> if TEST_DB_NAME == '<STR_LIT>' : <EOL> raise AssertionError ( '<STR_LIT>' ) <EOL> conn = sqlite3 . connect ( TEST_DB_NAME ) <EOL> try : <EOL> with conn : <EOL> conn . execute ( '<STR_LIT>' ) <EOL> finally : <EOL> conn . close ( ) <EOL> return <EOL> raise AssertionError ( '<STR_LIT>' % get_db_engine ( ) ) <EOL> def mark_exists ( ) : <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_cmd ( '<STR_LIT>' , TEST_DB_NAME , '<STR_LIT:-c>' , '<STR_LIT>' ) <EOL> return bool ( r . std_out ) <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> r = run_mysql ( TEST_DB_NAME , '<STR_LIT>' , '<STR_LIT>' ) <EOL> return r . status_code == <NUM_LIT:0> <EOL> if get_db_engine ( ) == '<STR_LIT>' : <EOL> if TEST_DB_NAME == '<STR_LIT>' : <EOL> raise AssertionError ( <EOL> '<STR_LIT>' ) <EOL> conn = sqlite3 . connect ( TEST_DB_NAME ) <EOL> try : <EOL> with conn : <EOL> conn . execute ( '<STR_LIT>' ) <EOL> return True <EOL> except sqlite3 . OperationalError : <EOL> return False <EOL> finally : <EOL> conn . close ( ) <EOL> raise AssertionError ( '<STR_LIT>' % get_db_engine ( ) ) <EOL> def noop_transactions ( ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( connection , '<STR_LIT>' ) : <EOL> return connection . in_atomic_block <EOL> else : <EOL> with transaction . commit_manually ( ) : <EOL> Item . objects . create ( name = '<STR_LIT>' ) <EOL> transaction . rollback ( ) <EOL> try : <EOL> item = Item . objects . get ( name = '<STR_LIT>' ) <EOL> except Item . DoesNotExist : <EOL> return False <EOL> else : <EOL> item . delete ( ) <EOL> return True </s>
