1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
import pytest
import sqlalchemy as sa
from sqlalchemy_utils import create_database, database_exists, drop_database
from sqlalchemy_utils.compat import get_sqlalchemy_version
pymysql = None
try:
import pymysql # noqa
except ImportError:
pass
sqlalchemy_version = get_sqlalchemy_version()
class DatabaseTest:
def test_create_and_drop(self, dsn):
assert not database_exists(dsn)
create_database(dsn)
assert database_exists(dsn)
drop_database(dsn)
assert not database_exists(dsn)
@pytest.mark.usefixtures('sqlite_memory_dsn')
class TestDatabaseSQLiteMemory:
def test_exists_memory(self, dsn):
assert database_exists(dsn)
@pytest.mark.usefixtures('sqlite_none_database_dsn')
class TestDatabaseSQLiteMemoryNoDatabaseString:
def test_exists_memory_none_database(self, sqlite_none_database_dsn):
assert database_exists(sqlite_none_database_dsn)
@pytest.mark.usefixtures('sqlite_file_dsn')
class TestDatabaseSQLiteFile(DatabaseTest):
def test_existing_non_sqlite_file(self, dsn):
database = sa.engine.url.make_url(dsn).database
open(database, 'w').close()
self.test_create_and_drop(dsn)
@pytest.mark.skipif('pymysql is None')
@pytest.mark.usefixtures('mysql_dsn')
class TestDatabaseMySQL(DatabaseTest):
@pytest.fixture
def db_name(self):
return 'db_test_sqlalchemy_util'
@pytest.mark.skipif('pymysql is None')
@pytest.mark.usefixtures('mysql_dsn')
class TestDatabaseMySQLWithQuotedName(DatabaseTest):
@pytest.fixture
def db_name(self):
return 'db_test_sqlalchemy-util'
@pytest.mark.usefixtures('postgresql_dsn')
class TestDatabasePostgres(DatabaseTest):
@pytest.fixture
def db_name(self):
return 'db_test_sqlalchemy_util'
def test_template(self, postgresql_db_user, postgresql_db_password):
dsn = 'postgresql://{}:{}@localhost/db_test_sqlalchemy_util'.format(
postgresql_db_user,
postgresql_db_password
)
with pytest.raises(sa.exc.ProgrammingError) as excinfo:
create_database(dsn, template='my_template')
assert ("CREATE DATABASE db_test_sqlalchemy_util ENCODING 'utf8' "
"TEMPLATE my_template") in str(excinfo.value)
class TestDatabasePostgresPg8000(DatabaseTest):
@pytest.fixture
def dsn(self, postgresql_db_user, postgresql_db_password):
return 'postgresql+pg8000://{}:{}@localhost/{}'.format(
postgresql_db_user,
postgresql_db_password,
'db_to_test_create_and_drop_via_pg8000_driver'
)
class TestDatabasePostgresPsycoPG2CFFI(DatabaseTest):
@pytest.fixture
def dsn(self, postgresql_db_user, postgresql_db_password):
return 'postgresql+psycopg2cffi://{}:{}@localhost/{}'.format(
postgresql_db_user,
postgresql_db_password,
'db_to_test_create_and_drop_via_psycopg2cffi_driver'
)
@pytest.mark.skipif('sqlalchemy_version < (2, 0, 0)')
class TestDatabasePostgresPsycoPG3(DatabaseTest):
@pytest.fixture
def dsn(self, postgresql_db_user, postgresql_db_password):
return 'postgresql+psycopg://{}:{}@localhost/{}'.format(
postgresql_db_user,
postgresql_db_password,
'db_to_test_create_and_drop_via_psycopg3_driver'
)
@pytest.mark.usefixtures('postgresql_dsn')
class TestDatabasePostgresWithQuotedName(DatabaseTest):
@pytest.fixture
def db_name(self):
return 'db_test_sqlalchemy-util'
def test_template(self, postgresql_db_user, postgresql_db_password):
dsn = 'postgresql://{}:{}@localhost/db_test_sqlalchemy-util'.format(
postgresql_db_user,
postgresql_db_password
)
with pytest.raises(sa.exc.ProgrammingError) as excinfo:
create_database(dsn, template='my-template')
assert ('CREATE DATABASE "db_test_sqlalchemy-util" ENCODING \'utf8\' '
'TEMPLATE "my-template"') in str(excinfo.value)
class TestDatabasePostgresCreateDatabaseCloseConnection:
def test_create_database_twice(
self,
postgresql_db_user,
postgresql_db_password
):
dsn_list = [
'postgresql://{}:{}@localhost/db_test_sqlalchemy-util-a'.format(
postgresql_db_user,
postgresql_db_password
),
'postgresql://{}:{}@localhost/db_test_sqlalchemy-util-b'.format(
postgresql_db_user,
postgresql_db_password
),
]
for dsn_item in dsn_list:
assert not database_exists(dsn_item)
create_database(dsn_item, template="template1")
assert database_exists(dsn_item)
for dsn_item in dsn_list:
drop_database(dsn_item)
assert not database_exists(dsn_item)
@pytest.mark.usefixtures('mssql_dsn')
class TestDatabaseMssql(DatabaseTest):
@pytest.fixture
def db_name(self):
pytest.importorskip('pyodbc')
return 'db_test_sqlalchemy_util'
|
