9
9
10
10
11
11
# Construct names for the "inner" database used in runpytest tests
12
- _settings = settings .DATABASES [' default' ]
12
+ _settings = settings .DATABASES [" default" ]
13
13
14
- DB_NAME = _settings [' NAME' ]
15
- TEST_DB_NAME = _settings [' TEST' ][ ' NAME' ]
14
+ DB_NAME = _settings [" NAME" ]
15
+ TEST_DB_NAME = _settings [" TEST" ][ " NAME" ]
16
16
17
- if _settings [' ENGINE' ] == ' django.db.backends.sqlite3' and TEST_DB_NAME is None :
18
- TEST_DB_NAME = ' :memory:'
17
+ if _settings [" ENGINE" ] == " django.db.backends.sqlite3" and TEST_DB_NAME is None :
18
+ TEST_DB_NAME = " :memory:"
19
19
else :
20
- DB_NAME += ' _inner'
20
+ DB_NAME += " _inner"
21
21
22
22
if TEST_DB_NAME is None :
23
23
# No explicit test db name was given, construct a default one
24
- TEST_DB_NAME = ' test_{}_inner' .format (DB_NAME )
24
+ TEST_DB_NAME = " test_{}_inner" .format (DB_NAME )
25
25
else :
26
26
# An explicit test db name was given, is that as the base name
27
- TEST_DB_NAME = ' {}_inner' .format (TEST_DB_NAME )
27
+ TEST_DB_NAME = " {}_inner" .format (TEST_DB_NAME )
28
28
29
29
30
30
def get_db_engine ():
31
- return _settings [' ENGINE' ].split ('.' )[- 1 ]
31
+ return _settings [" ENGINE" ].split ("." )[- 1 ]
32
32
33
33
34
34
class CmdResult (object ):
@@ -46,121 +46,125 @@ def run_cmd(*args):
46
46
47
47
48
48
def run_mysql (* args ):
49
- user = _settings .get (' USER' , None )
49
+ user = _settings .get (" USER" , None )
50
50
if user :
51
- args = ('-u' , user ) + tuple (args )
52
- args = (' mysql' ,) + tuple (args )
51
+ args = ("-u" , user ) + tuple (args )
52
+ args = (" mysql" ,) + tuple (args )
53
53
return run_cmd (* args )
54
54
55
55
56
56
def skip_if_sqlite_in_memory ():
57
- if _settings ['ENGINE' ] == 'django.db.backends.sqlite3' and _settings ['TEST' ]['NAME' ] is None :
58
- pytest .skip ('Do not test db reuse since database does not support it' )
57
+ if (
58
+ _settings ["ENGINE" ] == "django.db.backends.sqlite3"
59
+ and _settings ["TEST" ]["NAME" ] is None
60
+ ):
61
+ pytest .skip ("Do not test db reuse since database does not support it" )
59
62
60
63
61
64
def drop_database (name = TEST_DB_NAME , suffix = None ):
62
- assert bool (name ) ^ bool (suffix ), ' name and suffix cannot be used together'
65
+ assert bool (name ) ^ bool (suffix ), " name and suffix cannot be used together"
63
66
64
67
if suffix :
65
- name = ' %s_%s' % (name , suffix )
68
+ name = " %s_%s" % (name , suffix )
66
69
67
- if get_db_engine () == ' postgresql_psycopg2' :
68
- r = run_cmd (' psql' , ' postgres' , '-c' , ' DROP DATABASE %s' % name )
70
+ if get_db_engine () == " postgresql_psycopg2" :
71
+ r = run_cmd (" psql" , " postgres" , "-c" , " DROP DATABASE %s" % name )
69
72
assert "DROP DATABASE" in force_text (
70
73
r .std_out
71
74
) or "does not exist" in force_text (r .std_err )
72
75
return
73
76
74
- if get_db_engine () == ' mysql' :
75
- r = run_mysql ('-e' , ' DROP DATABASE %s' % name )
77
+ if get_db_engine () == " mysql" :
78
+ r = run_mysql ("-e" , " DROP DATABASE %s" % name )
76
79
assert "database doesn't exist" in force_text (r .std_err ) or r .status_code == 0
77
80
return
78
81
79
- if get_db_engine () == 'sqlite3' :
80
- if name == ':memory:' :
81
- raise AssertionError (
82
- 'sqlite in-memory database cannot be dropped!' )
82
+ if get_db_engine () == "sqlite3" :
83
+ if name == ":memory:" :
84
+ raise AssertionError ("sqlite in-memory database cannot be dropped!" )
83
85
if os .path .exists (name ):
84
86
os .unlink (name )
85
87
return
86
88
87
- raise AssertionError (' %s cannot be tested properly!' % get_db_engine ())
89
+ raise AssertionError (" %s cannot be tested properly!" % get_db_engine ())
88
90
89
91
90
92
def db_exists (db_suffix = None ):
91
93
name = TEST_DB_NAME
92
94
93
95
if db_suffix :
94
- name = ' %s_%s' % (name , db_suffix )
96
+ name = " %s_%s" % (name , db_suffix )
95
97
96
- if get_db_engine () == ' postgresql_psycopg2' :
97
- r = run_cmd (' psql' , name , '-c' , ' SELECT 1' )
98
+ if get_db_engine () == " postgresql_psycopg2" :
99
+ r = run_cmd (" psql" , name , "-c" , " SELECT 1" )
98
100
return r .status_code == 0
99
101
100
- if get_db_engine () == ' mysql' :
101
- r = run_mysql (name , '-e' , ' SELECT 1' )
102
+ if get_db_engine () == " mysql" :
103
+ r = run_mysql (name , "-e" , " SELECT 1" )
102
104
return r .status_code == 0
103
105
104
- if get_db_engine () == ' sqlite3' :
105
- if TEST_DB_NAME == ' :memory:' :
106
+ if get_db_engine () == " sqlite3" :
107
+ if TEST_DB_NAME == " :memory:" :
106
108
raise AssertionError (
107
- 'sqlite in-memory database cannot be checked for existence!' )
109
+ "sqlite in-memory database cannot be checked for existence!"
110
+ )
108
111
return os .path .exists (name )
109
112
110
- raise AssertionError (' %s cannot be tested properly!' % get_db_engine ())
113
+ raise AssertionError (" %s cannot be tested properly!" % get_db_engine ())
111
114
112
115
113
116
def mark_database ():
114
- if get_db_engine () == ' postgresql_psycopg2' :
115
- r = run_cmd (' psql' , TEST_DB_NAME , '-c' , ' CREATE TABLE mark_table();' )
117
+ if get_db_engine () == " postgresql_psycopg2" :
118
+ r = run_cmd (" psql" , TEST_DB_NAME , "-c" , " CREATE TABLE mark_table();" )
116
119
assert r .status_code == 0
117
120
return
118
121
119
- if get_db_engine () == ' mysql' :
120
- r = run_mysql (TEST_DB_NAME , '-e' , ' CREATE TABLE mark_table(kaka int);' )
122
+ if get_db_engine () == " mysql" :
123
+ r = run_mysql (TEST_DB_NAME , "-e" , " CREATE TABLE mark_table(kaka int);" )
121
124
assert r .status_code == 0
122
125
return
123
126
124
- if get_db_engine () == ' sqlite3' :
125
- if TEST_DB_NAME == ' :memory:' :
126
- raise AssertionError (' sqlite in-memory database cannot be marked!' )
127
+ if get_db_engine () == " sqlite3" :
128
+ if TEST_DB_NAME == " :memory:" :
129
+ raise AssertionError (" sqlite in-memory database cannot be marked!" )
127
130
128
131
conn = sqlite3 .connect (TEST_DB_NAME )
129
132
try :
130
133
with conn :
131
- conn .execute (' CREATE TABLE mark_table(kaka int);' )
134
+ conn .execute (" CREATE TABLE mark_table(kaka int);" )
132
135
finally : # Close the DB even if an error is raised
133
136
conn .close ()
134
137
return
135
138
136
- raise AssertionError (' %s cannot be tested properly!' % get_db_engine ())
139
+ raise AssertionError (" %s cannot be tested properly!" % get_db_engine ())
137
140
138
141
139
142
def mark_exists ():
140
- if get_db_engine () == ' postgresql_psycopg2' :
141
- r = run_cmd (' psql' , TEST_DB_NAME , '-c' , ' SELECT 1 FROM mark_table' )
143
+ if get_db_engine () == " postgresql_psycopg2" :
144
+ r = run_cmd (" psql" , TEST_DB_NAME , "-c" , " SELECT 1 FROM mark_table" )
142
145
143
146
# When something pops out on std_out, we are good
144
147
return bool (r .std_out )
145
148
146
- if get_db_engine () == ' mysql' :
147
- r = run_mysql (TEST_DB_NAME , '-e' , ' SELECT 1 FROM mark_table' )
149
+ if get_db_engine () == " mysql" :
150
+ r = run_mysql (TEST_DB_NAME , "-e" , " SELECT 1 FROM mark_table" )
148
151
149
152
return r .status_code == 0
150
153
151
- if get_db_engine () == ' sqlite3' :
152
- if TEST_DB_NAME == ' :memory:' :
154
+ if get_db_engine () == " sqlite3" :
155
+ if TEST_DB_NAME == " :memory:" :
153
156
raise AssertionError (
154
- 'sqlite in-memory database cannot be checked for mark!' )
157
+ "sqlite in-memory database cannot be checked for mark!"
158
+ )
155
159
156
160
conn = sqlite3 .connect (TEST_DB_NAME )
157
161
try :
158
162
with conn :
159
- conn .execute (' SELECT 1 FROM mark_table' )
163
+ conn .execute (" SELECT 1 FROM mark_table" )
160
164
return True
161
165
except sqlite3 .OperationalError :
162
166
return False
163
167
finally : # Close the DB even if an error is raised
164
168
conn .close ()
165
169
166
- raise AssertionError (' %s cannot be tested properly!' % get_db_engine ())
170
+ raise AssertionError (" %s cannot be tested properly!" % get_db_engine ())
0 commit comments