Mercurial > p > mysql-python > mysqldb-2

comparison tests/test_MySQLdb_dbapi20.py @ 18:d55bfb1a4701 MySQLdb

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Tons of changes from major refactoring/cleanup. This is all really broken right now. In particular, all results are returned as strings.
author adustman
date Fri, 14 Mar 2008 23:06:29 +0000
parents
children
comparison
equal deleted inserted replaced
17:7c7a89123d65 18:d55bfb1a4701
1 #!/usr/bin/env python
2 import dbapi20
3 import unittest
4 import MySQLdb
5
6 class test_MySQLdb(dbapi20.DatabaseAPI20Test):
7 driver = MySQLdb
8 connect_args = ()
9 connect_kw_args = dict(db='test',
10 read_default_file='~/.my.cnf',
11 charset='utf8',
12 sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL")
13
14 def test_setoutputsize(self): pass
15 def test_setoutputsize_basic(self): pass
16 def test_nextset(self): pass
17
18 """The tests on fetchone and fetchall and rowcount bogusly
19 test for an exception if the statement cannot return a
20 result set. MySQL always returns a result set; it's just that
21 some things return empty result sets."""
22
23 def test_fetchall(self):
24 con = self._connect()
25 try:
26 cur = con.cursor()
27 # cursor.fetchall should raise an Error if called
28 # without executing a query that may return rows (such
29 # as a select)
30 self.assertRaises(self.driver.Error, cur.fetchall)
31
32 self.executeDDL1(cur)
33 for sql in self._populate():
34 cur.execute(sql)
35
36 # cursor.fetchall should raise an Error if called
37 # after executing a a statement that cannot return rows
38 ## self.assertRaises(self.driver.Error,cur.fetchall)
39
40 cur.execute('select name from %sbooze' % self.table_prefix)
41 rows = cur.fetchall()
42 self.failUnless(cur.rowcount in (-1,len(self.samples)))
43 self.assertEqual(len(rows),len(self.samples),
44 'cursor.fetchall did not retrieve all rows'
45 )
46 rows = [r[0] for r in rows]
47 rows.sort()
48 for i in range(0,len(self.samples)):
49 self.assertEqual(rows[i],self.samples[i],
50 'cursor.fetchall retrieved incorrect rows'
51 )
52 rows = cur.fetchall()
53 self.assertEqual(
54 len(rows),0,
55 'cursor.fetchall should return an empty list if called '
56 'after the whole result set has been fetched'
57 )
58 self.failUnless(cur.rowcount in (-1,len(self.samples)))
59
60 self.executeDDL2(cur)
61 cur.execute('select name from %sbarflys' % self.table_prefix)
62 rows = cur.fetchall()
63 self.failUnless(cur.rowcount in (-1,0))
64 self.assertEqual(len(rows),0,
65 'cursor.fetchall should return an empty list if '
66 'a select query returns no rows'
67 )
68
69 finally:
70 con.close()
71
72 def test_fetchone(self):
73 con = self._connect()
74 try:
75 cur = con.cursor()
76
77 # cursor.fetchone should raise an Error if called before
78 # executing a select-type query
79 self.assertRaises(self.driver.Error,cur.fetchone)
80
81 # cursor.fetchone should raise an Error if called after
82 # executing a query that cannnot return rows
83 self.executeDDL1(cur)
84 ## self.assertRaises(self.driver.Error,cur.fetchone)
85
86 cur.execute('select name from %sbooze' % self.table_prefix)
87 self.assertEqual(cur.fetchone(),None,
88 'cursor.fetchone should return None if a query retrieves '
89 'no rows'
90 )
91 self.failUnless(cur.rowcount in (-1,0))
92
93 # cursor.fetchone should raise an Error if called after
94 # executing a query that cannnot return rows
95 cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
96 self.table_prefix
97 ))
98 ## self.assertRaises(self.driver.Error,cur.fetchone)
99
100 cur.execute('select name from %sbooze' % self.table_prefix)
101 r = cur.fetchone()
102 self.assertEqual(len(r),1,
103 'cursor.fetchone should have retrieved a single row'
104 )
105 self.assertEqual(r[0],'Victoria Bitter',
106 'cursor.fetchone retrieved incorrect data'
107 )
108 ## self.assertEqual(cur.fetchone(),None,
109 ## 'cursor.fetchone should return None if no more rows available'
110 ## )
111 self.failUnless(cur.rowcount in (-1,1))
112 finally:
113 con.close()
114
115 # Same complaint as for fetchall and fetchone
116 def test_rowcount(self):
117 con = self._connect()
118 try:
119 cur = con.cursor()
120 self.executeDDL1(cur)
121 ## self.assertEqual(cur.rowcount,-1,
122 ## 'cursor.rowcount should be -1 after executing no-result '
123 ## 'statements'
124 ## )
125 cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
126 self.table_prefix
127 ))
128 ## self.failUnless(cur.rowcount in (-1,1),
129 ## 'cursor.rowcount should == number or rows inserted, or '
130 ## 'set to -1 after executing an insert statement'
131 ## )
132 cur.execute("select name from %sbooze" % self.table_prefix)
133 self.failUnless(cur.rowcount in (-1,1),
134 'cursor.rowcount should == number of rows returned, or '
135 'set to -1 after executing a select statement'
136 )
137 self.executeDDL2(cur)
138 ## self.assertEqual(cur.rowcount,-1,
139 ## 'cursor.rowcount not being reset to -1 after executing '
140 ## 'no-result statements'
141 ## )
142 finally:
143 con.close()
144
145 def test_callproc(self):
146 pass # performed in test_MySQL_capabilities
147
148 def help_nextset_setUp(self,cur):
149 ''' Should create a procedure called deleteme
150 that returns two result sets, first the
151 number of rows in booze then "name from booze"
152 '''
153 sql="""
154 create procedure deleteme()
155 begin
156 select count(*) from %(tp)sbooze;
157 select name from %(tp)sbooze;
158 end
159 """ % dict(tp=self.table_prefix)
160 cur.execute(sql)
161
162 def help_nextset_tearDown(self,cur):
163 'If cleaning up is needed after nextSetTest'
164 cur.execute("drop procedure deleteme")
165
166 def test_nextset(self):
167 from warnings import warn
168 con = self._connect()
169 try:
170 cur = con.cursor()
171 if not hasattr(cur,'nextset'):
172 return
173
174 try:
175 self.executeDDL1(cur)
176 sql=self._populate()
177 for sql in self._populate():
178 cur.execute(sql)
179
180 self.help_nextset_setUp(cur)
181
182 cur.callproc('deleteme')
183 numberofrows=cur.fetchone()
184 assert numberofrows[0]== len(self.samples)
185 assert cur.nextset()
186 names=cur.fetchall()
187 assert len(names) == len(self.samples)
188 s=cur.nextset()
189 if s:
190 empty = cur.fetchall()
191 self.assertEquals(len(empty), 0,
192 "non-empty result set after other result sets")
193 #warn("Incompatibility: MySQL returns an empty result set for the CALL itself",
194 # Warning)
195 #assert s == None,'No more return sets, should return None'
196 finally:
197 self.help_nextset_tearDown(cur)
198
199 finally:
200 con.close()
201
202
203 if __name__ == '__main__':
204 unittest.main()
205 print '''"Huh-huh, he said 'unit'." -- Butthead'''