@@ -47,7 +47,7 @@ def run_cmd(*args):
47
47
48
48
def run_mysql (* args ):
49
49
user = _settings .get ("USER" , None )
50
- if user :
50
+ if user : # pragma: no branch
51
51
args = ("-u" , user ) + tuple (args )
52
52
args = ("mysql" ,) + tuple (args )
53
53
return run_cmd (* args )
@@ -61,110 +61,97 @@ def skip_if_sqlite_in_memory():
61
61
pytest .skip ("Do not test db reuse since database does not support it" )
62
62
63
63
64
- def drop_database (name = TEST_DB_NAME , suffix = None ):
65
- assert bool ( name ) ^ bool ( suffix ), "name and suffix cannot be used together"
64
+ def drop_database (name = TEST_DB_NAME ):
65
+ db_engine = get_db_engine ()
66
66
67
- if suffix :
68
- name = "%s_%s" % (name , suffix )
69
-
70
- if get_db_engine () == "postgresql_psycopg2" :
67
+ if db_engine == "postgresql_psycopg2" :
71
68
r = run_cmd ("psql" , "postgres" , "-c" , "DROP DATABASE %s" % name )
72
69
assert "DROP DATABASE" in force_text (
73
70
r .std_out
74
71
) or "does not exist" in force_text (r .std_err )
75
72
return
76
73
77
- if get_db_engine () == "mysql" :
74
+ if db_engine == "mysql" :
78
75
r = run_mysql ("-e" , "DROP DATABASE %s" % name )
79
76
assert "database doesn't exist" in force_text (r .std_err ) or r .status_code == 0
80
77
return
81
78
82
- if get_db_engine () == "sqlite3" :
83
- if name == ":memory:" :
84
- raise AssertionError ("sqlite in-memory database cannot be dropped!" )
85
- if os .path .exists (name ):
86
- os .unlink (name )
87
- return
88
-
89
- raise AssertionError ("%s cannot be tested properly!" % get_db_engine ())
79
+ assert db_engine == "sqlite3" , "%s cannot be tested properly!" % db_engine
80
+ assert name != ":memory:" , "sqlite in-memory database cannot be dropped!"
81
+ if os .path .exists (name ): # pragma: no branch
82
+ os .unlink (name )
90
83
91
84
92
85
def db_exists (db_suffix = None ):
93
86
name = TEST_DB_NAME
87
+ db_engine = get_db_engine ()
94
88
95
89
if db_suffix :
96
90
name = "%s_%s" % (name , db_suffix )
97
91
98
- if get_db_engine () == "postgresql_psycopg2" :
92
+ if db_engine == "postgresql_psycopg2" :
99
93
r = run_cmd ("psql" , name , "-c" , "SELECT 1" )
100
94
return r .status_code == 0
101
95
102
- if get_db_engine () == "mysql" :
96
+ if db_engine == "mysql" :
103
97
r = run_mysql (name , "-e" , "SELECT 1" )
104
98
return r .status_code == 0
105
99
106
- if get_db_engine () == "sqlite3" :
107
- if TEST_DB_NAME == ":memory:" :
108
- raise AssertionError (
109
- "sqlite in-memory database cannot be checked for existence!"
110
- )
111
- return os .path .exists (name )
112
-
113
- raise AssertionError ("%s cannot be tested properly!" % get_db_engine ())
100
+ assert db_engine == "sqlite3" , "%s cannot be tested properly!" % db_engine
101
+ assert TEST_DB_NAME != ":memory:" , (
102
+ "sqlite in-memory database cannot be checked for existence!" )
103
+ return os .path .exists (name )
114
104
115
105
116
106
def mark_database ():
117
- if get_db_engine () == "postgresql_psycopg2" :
107
+ db_engine = get_db_engine ()
108
+
109
+ if db_engine == "postgresql_psycopg2" :
118
110
r = run_cmd ("psql" , TEST_DB_NAME , "-c" , "CREATE TABLE mark_table();" )
119
111
assert r .status_code == 0
120
112
return
121
113
122
- if get_db_engine () == "mysql" :
114
+ if db_engine == "mysql" :
123
115
r = run_mysql (TEST_DB_NAME , "-e" , "CREATE TABLE mark_table(kaka int);" )
124
116
assert r .status_code == 0
125
117
return
126
118
127
- if get_db_engine () == "sqlite3" :
128
- if TEST_DB_NAME == ":memory:" :
129
- raise AssertionError ("sqlite in-memory database cannot be marked!" )
130
-
131
- conn = sqlite3 .connect (TEST_DB_NAME )
132
- try :
133
- with conn :
134
- conn .execute ("CREATE TABLE mark_table(kaka int);" )
135
- finally : # Close the DB even if an error is raised
136
- conn .close ()
137
- return
119
+ assert db_engine == "sqlite3" , "%s cannot be tested properly!" % db_engine
120
+ assert TEST_DB_NAME != ":memory:" , (
121
+ "sqlite in-memory database cannot be marked!" )
138
122
139
- raise AssertionError ("%s cannot be tested properly!" % get_db_engine ())
123
+ conn = sqlite3 .connect (TEST_DB_NAME )
124
+ try :
125
+ with conn :
126
+ conn .execute ("CREATE TABLE mark_table(kaka int);" )
127
+ finally : # Close the DB even if an error is raised
128
+ conn .close ()
140
129
141
130
142
131
def mark_exists ():
143
- if get_db_engine () == "postgresql_psycopg2" :
132
+ db_engine = get_db_engine ()
133
+
134
+ if db_engine == "postgresql_psycopg2" :
144
135
r = run_cmd ("psql" , TEST_DB_NAME , "-c" , "SELECT 1 FROM mark_table" )
145
136
146
137
# When something pops out on std_out, we are good
147
138
return bool (r .std_out )
148
139
149
- if get_db_engine () == "mysql" :
140
+ if db_engine == "mysql" :
150
141
r = run_mysql (TEST_DB_NAME , "-e" , "SELECT 1 FROM mark_table" )
151
142
152
143
return r .status_code == 0
153
144
154
- if get_db_engine () == "sqlite3" :
155
- if TEST_DB_NAME == ":memory:" :
156
- raise AssertionError (
157
- "sqlite in-memory database cannot be checked for mark!"
158
- )
159
-
160
- conn = sqlite3 .connect (TEST_DB_NAME )
161
- try :
162
- with conn :
163
- conn .execute ("SELECT 1 FROM mark_table" )
164
- return True
165
- except sqlite3 .OperationalError :
166
- return False
167
- finally : # Close the DB even if an error is raised
168
- conn .close ()
169
-
170
- raise AssertionError ("%s cannot be tested properly!" % get_db_engine ())
145
+ assert db_engine == "sqlite3" , "%s cannot be tested properly!" % db_engine
146
+ assert TEST_DB_NAME != ":memory:" , (
147
+ "sqlite in-memory database cannot be checked for mark!" )
148
+
149
+ conn = sqlite3 .connect (TEST_DB_NAME )
150
+ try :
151
+ with conn :
152
+ conn .execute ("SELECT 1 FROM mark_table" )
153
+ return True
154
+ except sqlite3 .OperationalError :
155
+ return False
156
+ finally : # Close the DB even if an error is raised
157
+ conn .close ()
0 commit comments