/
test_db_create.py
239 lines (194 loc) · 7.38 KB
/
test_db_create.py
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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, division
import sys
import logging
from datetime import datetime, date
import sqlite3
from petl.io.db import fromdb, todb
from petl.io.db_create import make_sqlalchemy_column
from petl.test.helpers import ieq, eq_
from petl.util.vis import look
from petl.test.io.test_db_server import user, password, host, database
logger = logging.getLogger(__name__)
debug = logger.debug
def _test_create(dbo):
expect = (('foo', 'bar'),
('a', 1),
('b', 2))
expect_extended = (('foo', 'bar', 'baz'),
('a', 1, 2.3),
('b', 2, 4.1))
actual = fromdb(dbo, 'SELECT * FROM test_create')
debug('verify table does not exist to start with')
try:
debug(look(actual))
except Exception as e:
debug('expected exception: ' + str(e))
else:
raise Exception('expected exception not raised')
debug('verify cannot write without create')
try:
todb(expect, dbo, 'test_create')
except Exception as e:
debug('expected exception: ' + str(e))
else:
raise Exception('expected exception not raised')
debug('create table and verify')
todb(expect, dbo, 'test_create', create=True)
ieq(expect, actual)
debug(look(actual))
debug('verify cannot overwrite with new cols without recreate')
try:
todb(expect_extended, dbo, 'test_create')
except Exception as e:
debug('expected exception: ' + str(e))
else:
raise Exception('expected exception not raised')
debug('verify recreate')
todb(expect_extended, dbo, 'test_create', create=True, drop=True)
ieq(expect_extended, actual)
debug(look(actual))
debug('horrendous identifiers')
table = (('foo foo', 'bar.baz."spong`'),
('a', 1),
('b', 2),
('c', 2))
todb(table, dbo, 'foo " bar`', create=True)
actual = fromdb(dbo, 'SELECT * FROM "foo "" bar`"')
ieq(table, actual)
def _setup_mysql(dbapi_connection):
# setup table
cursor = dbapi_connection.cursor()
# deal with quote compatibility
cursor.execute('SET SQL_MODE=ANSI_QUOTES')
cursor.execute('DROP TABLE IF EXISTS test_create')
cursor.execute('DROP TABLE IF EXISTS "foo "" bar`"')
cursor.close()
dbapi_connection.commit()
def _setup_generic(dbapi_connection):
# setup table
cursor = dbapi_connection.cursor()
cursor.execute('DROP TABLE IF EXISTS test_create')
cursor.execute('DROP TABLE IF EXISTS "foo "" bar`"')
cursor.close()
dbapi_connection.commit()
try:
# noinspection PyUnresolvedReferences
import sqlalchemy
except ImportError as e:
print('SKIP generic create tests: %s' % e, file=sys.stderr)
else:
from sqlalchemy import Column, DateTime, Date
def test_make_datetime_column():
sql_col = make_sqlalchemy_column([datetime(2014, 1, 1, 1, 1, 1, 1),
datetime(2014, 1, 1, 1, 1, 1, 2)],
'name')
expect = Column('name', DateTime(), nullable=False)
eq_(str(expect.type), str(sql_col.type))
def test_make_date_column():
sql_col = make_sqlalchemy_column([date(2014, 1, 1),
date(2014, 1, 2)],
'name')
expect = Column('name', Date(), nullable=False)
eq_(str(expect.type), str(sql_col.type))
def test_sqlite3_create():
dbapi_connection = sqlite3.connect(':memory:')
# exercise using a dbapi_connection
_setup_generic(dbapi_connection)
_test_create(dbapi_connection)
# exercise using a dbapi_cursor
_setup_generic(dbapi_connection)
dbapi_cursor = dbapi_connection.cursor()
_test_create(dbapi_cursor)
dbapi_cursor.close()
try:
import pymysql
import sqlalchemy
pymysql.connect(host=host,
user=user,
password=password,
database=database)
except Exception as e:
print('SKIP pymysql create tests: %s' % e, file=sys.stderr)
else:
def test_mysql_create():
import pymysql
connect = pymysql.connect
# assume database already created
dbapi_connection = connect(host=host,
user=user,
password=password,
database=database)
# exercise using a dbapi_connection
_setup_mysql(dbapi_connection)
_test_create(dbapi_connection)
# exercise using a dbapi_cursor
_setup_mysql(dbapi_connection)
dbapi_cursor = dbapi_connection.cursor()
_test_create(dbapi_cursor)
dbapi_cursor.close()
# exercise sqlalchemy dbapi_connection
_setup_mysql(dbapi_connection)
from sqlalchemy import create_engine
sqlalchemy_engine = create_engine('mysql+pymysql://%s:%s@%s/%s'
% (user, password, host, database))
sqlalchemy_connection = sqlalchemy_engine.connect()
sqlalchemy_connection.execute('SET SQL_MODE=ANSI_QUOTES')
_test_create(sqlalchemy_connection)
sqlalchemy_connection.close()
# exercise sqlalchemy session
_setup_mysql(dbapi_connection)
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=sqlalchemy_engine)
sqlalchemy_session = Session()
_test_create(sqlalchemy_session)
sqlalchemy_session.close()
try:
import psycopg2
import sqlalchemy
psycopg2.connect(
'host=%s dbname=%s user=%s password=%s'
% (host, database, user, password)
)
except Exception as e:
print('SKIP psycopg2 create tests: %s' % e, file=sys.stderr)
else:
def test_postgresql_create():
import psycopg2
import psycopg2.extensions
psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
psycopg2.extensions.register_type(psycopg2.extensions.UNICODEARRAY)
# assume database already created
dbapi_connection = psycopg2.connect(
'host=%s dbname=%s user=%s password=%s'
% (host, database, user, password)
)
dbapi_connection.autocommit = True
# exercise using a dbapi_connection
_setup_generic(dbapi_connection)
_test_create(dbapi_connection)
# exercise using a dbapi_cursor
_setup_generic(dbapi_connection)
dbapi_cursor = dbapi_connection.cursor()
_test_create(dbapi_cursor)
dbapi_cursor.close()
# # ignore these for now, having trouble with autocommit
#
# # exercise sqlalchemy dbapi_connection
# _setup_generic(dbapi_connection)
# from sqlalchemy import create_engine
# sqlalchemy_engine = create_engine(
# 'postgresql+psycopg2://%s:%s@%s/%s'
# % (user, password, host, database)
# )
# sqlalchemy_connection = sqlalchemy_engine.connect()
# _test_create(sqlalchemy_connection)
# sqlalchemy_connection.close()
#
# # exercise sqlalchemy session
# _setup_generic(dbapi_connection)
# from sqlalchemy.orm import sessionmaker
# Session = sessionmaker(bind=sqlalchemy_engine)
# sqlalchemy_session = Session()
# _test_create(sqlalchemy_session)
# sqlalchemy_session.close()