forked from vikram/monetdb-python
-
Notifications
You must be signed in to change notification settings - Fork 1
/
tests.py
269 lines (212 loc) · 6.81 KB
/
tests.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
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
# Copyright (c) 2009 - 2010, Mark Bucciarelli <mkbucc@gmail.com>
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
import subprocess
import sys
import unittest
from django.conf import settings
#
# Must configure settings before opening database connection.
#
db = 'testdjangodb1'
schema = 'django1'
user = 'django1'
passwd = 'django1'
settings.configure(
DEBUG=True,
DATABASE_ENGINE='django_monetdb',
DATABASE_NAME=db,
DATABASE_USER=user,
DATABASE_PASSWORD=passwd,
INSTALLED_APPS = ('testapp',),
)
#
# In order for Django to find our driver for the unit tests, the parent
# directory of the source must be on the system path, as Django tries
# to import the module django_monetdb.base.
#
sys.path.append('..')
class SubprocessError(Exception): pass
def run(cmd):
'''Wrapper for subprocess.call that handles errors.'''
emsg = ''
try:
rc = subprocess.call(cmd, shell=True)
if rc == 0:
pass # normal
elif rc < 0:
emsg = "%s, child terminated by signal %s" \
% (cmd, -rc)
else:
emsg = "%s, child returned error %s" % (cmd, rc)
except OSError, e:
emsg = "%s, execution failed: %s" % (cdm, e)
if emsg:
raise SubprocessError(emsg)
class TestMonetDjango(unittest.TestCase):
'''Basic tests of MonetDB driver.'''
def setUp(self):
cmd = './createdb.sh "%s" "%s" "%s" "%s"' % \
(db, user, passwd, schema)
run(cmd)
def tearDown(self):
from django.db import connection
connection.close()
cmd = './zapdb.sh "%s" "%s" "%s" "%s"' % \
(db, user, passwd, schema)
#run(cmd)
def testinit(self):
'''instantiate our custom database wrapper.'''
from django_monetdb.base import DatabaseWrapper
db = DatabaseWrapper({})
def testcreate(self):
'''instantiate a cursor.'''
import django_monetdb
w = django_monetdb.base.DatabaseWrapper({})
c = w.cursor()
self.failUnless(c)
def testbasicsql(self):
'''Run some base SQL through cursor.
This time we get the cursor (and connection) the Django way.
'''
from django.db import connection
c = connection.cursor()
s = "CREATE TABLE test (id int, name varchar(10))"
c.execute(s)
s = "INSERT INTO test values (1, 'one')"
c.execute(s)
s = "INSERT INTO test values (2, 'two')"
c.execute(s)
s = "INSERT INTO test values (3, 'three')"
c.execute(s)
s = "SELECT name FROM test WHERE id = 2"
self.failUnless(c.execute(s) == 1)
row = c.fetchone()
self.failUnless(row[0] == 'two')
def testconnection(self):
'''Get a database connection the Django way.'''
from django.db import connection
c = connection.cursor()
self.assert_(c)
def testsyncdb(self):
'''Does syncdb run (using models.py in the testapp subdir)?
Note that this does not actually test that fields are created,
just that syncdb command exits without error. For example,
while developing this driver I noticed that the FloatField was
not actually created, although syncdb completed.
As a workaround for this particular issue, I used a float field
in the unique_together section of the Meta subclass. But that's
a one-off test hack; to really test, I should get a model instance
and verify it's attributes match what was inserted into the db.
The core issue is that if db_type() returns None, the field is
skipped. And db_type() can return none if you have an error
in your driver's data_type dictionary.
'''
from django.core.management import call_command
call_command('syncdb')
def testget_or_create(self):
'''get_or_create requires driver to have an "ops" dictionary.
ref: django/db/models/sql/where.py, make_atom()
'''
from django.core.management import call_command
call_command('syncdb')
from testapp.models import Simple
obj, created = Simple.objects.get_or_create(name='one')
def testcascadingdelete(self):
'''in Django, deletes cascade. test this works.'''
from testapp.models import Simple, Parent, Aunt, GrandParent
from django.core.management import call_command
call_command('syncdb')
s = Simple(name='one')
s.save()
s = Simple(name='two')
s.save()
#
# Simple record should now be in database.
#
try:
s = Simple.objects.get(name='one')
except Simple.DoesNotExist:
self.fail("didn't save Simple record")
p = Parent(simple=s, name='p')
p.save()
try:
tst = Parent.objects.get(name='p')
except Parent.DoesNotExist:
self.fail("didn't save Parent record")
a = Aunt(simple=s, name='a')
a.save()
try:
tst = Aunt.objects.get(name='a')
except Aunt.DoesNotExist:
self.fail("didn't save Aunt record")
gp = GrandParent(parent=p, name='gp')
gp.save()
try:
tst = GrandParent.objects.get(name='gp')
except GrandParent.DoesNotExist:
self.fail("didn't save GrandParent record")
s.delete()
#
# Parent, Aunt and GrandParent records should now be gone from database.
#
ok = False
try:
tst = Parent.objects.get(name='p')
except Parent.DoesNotExist:
ok = True
self.failIf(not ok, "delete did not cascade");
ok = False
try:
tst = GrandParent.objects.get(name='gp')
except GrandParent.DoesNotExist:
ok = True
self.failIf(not ok, "delete did not cascade");
ok = False
try:
tst = Aunt.objects.get(name='a')
except Aunt.DoesNotExist:
ok = True
self.failIf(not ok, "delete did not cascade");
def testutf8(self):
'''test that we can save and retrieve utf-8 characters.
This was an attempt to duplicate a problem I hit with
a real Django app; this test passed though, so I didn't
capture the issue properly.
'''
from testapp.models import Simple, Parent, Aunt, GrandParent
from django.core.management import call_command
call_command('syncdb')
#
# cafe is in unicode here.
#
unicode_cafe = u"caf" + unichr(0x00E9)
utf8_cafe = unicode_cafe.encode('utf8')
print "cafe =", unicode_cafe.encode('utf8')
s = Simple(name=unicode_cafe)
s.save()
#
# At this point, string should be in database as UTF-8.
# When we retrieve it through Django, it should be
# back to unicode.
#
# Note:
# unicode(s) calls s.__unicode__()
#
o = Simple.objects.get(pk=1)
django_cafe = o.name
self.assertEqual(type(django_cafe), type(unicode_cafe))
self.assertEqual(django_cafe, unicode_cafe)
if __name__ == '__main__':
unittest.main()