source: python/trunk/Lib/sqlite3/test/factory.py

Last change on this file was 391, checked in by dmik, 11 years ago

python: Merge vendor 2.7.6 to trunk.
  • Property svn:eol-style set to native
File size: 9.4 KB
Line 
1#-*- coding: ISO-8859-1 -*-
2# pysqlite2/test/factory.py: tests for the various factories in pysqlite
3#
4# Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de>
5#
6# This file is part of pysqlite.
7#
8# This software is provided 'as-is', without any express or implied
9# warranty. In no event will the authors be held liable for any damages
10# arising from the use of this software.
11#
12# Permission is granted to anyone to use this software for any purpose,
13# including commercial applications, and to alter it and redistribute it
14# freely, subject to the following restrictions:
15#
16# 1. The origin of this software must not be misrepresented; you must not
17# claim that you wrote the original software. If you use this software
18# in a product, an acknowledgment in the product documentation would be
19# appreciated but is not required.
20# 2. Altered source versions must be plainly marked as such, and must not be
21# misrepresented as being the original software.
22# 3. This notice may not be removed or altered from any source distribution.
23
24import unittest
25import sqlite3 as sqlite
26
27class MyConnection(sqlite.Connection):
28 def __init__(self, *args, **kwargs):
29 sqlite.Connection.__init__(self, *args, **kwargs)
30
31def dict_factory(cursor, row):
32 d = {}
33 for idx, col in enumerate(cursor.description):
34 d[col[0]] = row[idx]
35 return d
36
37class MyCursor(sqlite.Cursor):
38 def __init__(self, *args, **kwargs):
39 sqlite.Cursor.__init__(self, *args, **kwargs)
40 self.row_factory = dict_factory
41
42class ConnectionFactoryTests(unittest.TestCase):
43 def setUp(self):
44 self.con = sqlite.connect(":memory:", factory=MyConnection)
45
46 def tearDown(self):
47 self.con.close()
48
49 def CheckIsInstance(self):
50 self.assertTrue(isinstance(self.con,
51 MyConnection),
52 "connection is not instance of MyConnection")
53
54class CursorFactoryTests(unittest.TestCase):
55 def setUp(self):
56 self.con = sqlite.connect(":memory:")
57
58 def tearDown(self):
59 self.con.close()
60
61 def CheckIsInstance(self):
62 cur = self.con.cursor(factory=MyCursor)
63 self.assertTrue(isinstance(cur,
64 MyCursor),
65 "cursor is not instance of MyCursor")
66
67class RowFactoryTestsBackwardsCompat(unittest.TestCase):
68 def setUp(self):
69 self.con = sqlite.connect(":memory:")
70
71 def CheckIsProducedByFactory(self):
72 cur = self.con.cursor(factory=MyCursor)
73 cur.execute("select 4+5 as foo")
74 row = cur.fetchone()
75 self.assertTrue(isinstance(row,
76 dict),
77 "row is not instance of dict")
78 cur.close()
79
80 def tearDown(self):
81 self.con.close()
82
83class RowFactoryTests(unittest.TestCase):
84 def setUp(self):
85 self.con = sqlite.connect(":memory:")
86
87 def CheckCustomFactory(self):
88 self.con.row_factory = lambda cur, row: list(row)
89 row = self.con.execute("select 1, 2").fetchone()
90 self.assertTrue(isinstance(row,
91 list),
92 "row is not instance of list")
93
94 def CheckSqliteRowIndex(self):
95 self.con.row_factory = sqlite.Row
96 row = self.con.execute("select 1 as a, 2 as b").fetchone()
97 self.assertTrue(isinstance(row,
98 sqlite.Row),
99 "row is not instance of sqlite.Row")
100
101 col1, col2 = row["a"], row["b"]
102 self.assertTrue(col1 == 1, "by name: wrong result for column 'a'")
103 self.assertTrue(col2 == 2, "by name: wrong result for column 'a'")
104
105 col1, col2 = row["A"], row["B"]
106 self.assertTrue(col1 == 1, "by name: wrong result for column 'A'")
107 self.assertTrue(col2 == 2, "by name: wrong result for column 'B'")
108
109 col1, col2 = row[0], row[1]
110 self.assertTrue(col1 == 1, "by index: wrong result for column 0")
111 self.assertTrue(col2 == 2, "by index: wrong result for column 1")
112
113 def CheckSqliteRowIter(self):
114 """Checks if the row object is iterable"""
115 self.con.row_factory = sqlite.Row
116 row = self.con.execute("select 1 as a, 2 as b").fetchone()
117 for col in row:
118 pass
119
120 def CheckSqliteRowAsTuple(self):
121 """Checks if the row object can be converted to a tuple"""
122 self.con.row_factory = sqlite.Row
123 row = self.con.execute("select 1 as a, 2 as b").fetchone()
124 t = tuple(row)
125
126 def CheckSqliteRowAsDict(self):
127 """Checks if the row object can be correctly converted to a dictionary"""
128 self.con.row_factory = sqlite.Row
129 row = self.con.execute("select 1 as a, 2 as b").fetchone()
130 d = dict(row)
131 self.assertEqual(d["a"], row["a"])
132 self.assertEqual(d["b"], row["b"])
133
134 def CheckSqliteRowHashCmp(self):
135 """Checks if the row object compares and hashes correctly"""
136 self.con.row_factory = sqlite.Row
137 row_1 = self.con.execute("select 1 as a, 2 as b").fetchone()
138 row_2 = self.con.execute("select 1 as a, 2 as b").fetchone()
139 row_3 = self.con.execute("select 1 as a, 3 as b").fetchone()
140
141 self.assertTrue(row_1 == row_1)
142 self.assertTrue(row_1 == row_2)
143 self.assertTrue(row_2 != row_3)
144
145 self.assertFalse(row_1 != row_1)
146 self.assertFalse(row_1 != row_2)
147 self.assertFalse(row_2 == row_3)
148
149 self.assertEqual(row_1, row_2)
150 self.assertEqual(hash(row_1), hash(row_2))
151 self.assertNotEqual(row_1, row_3)
152 self.assertNotEqual(hash(row_1), hash(row_3))
153
154 def tearDown(self):
155 self.con.close()
156
157class TextFactoryTests(unittest.TestCase):
158 def setUp(self):
159 self.con = sqlite.connect(":memory:")
160
161 def CheckUnicode(self):
162 austria = unicode("Österreich", "latin1")
163 row = self.con.execute("select ?", (austria,)).fetchone()
164 self.assertTrue(type(row[0]) == unicode, "type of row[0] must be unicode")
165
166 def CheckString(self):
167 self.con.text_factory = str
168 austria = unicode("Österreich", "latin1")
169 row = self.con.execute("select ?", (austria,)).fetchone()
170 self.assertTrue(type(row[0]) == str, "type of row[0] must be str")
171 self.assertTrue(row[0] == austria.encode("utf-8"), "column must equal original data in UTF-8")
172
173 def CheckCustom(self):
174 self.con.text_factory = lambda x: unicode(x, "utf-8", "ignore")
175 austria = unicode("Österreich", "latin1")
176 row = self.con.execute("select ?", (austria.encode("latin1"),)).fetchone()
177 self.assertTrue(type(row[0]) == unicode, "type of row[0] must be unicode")
178 self.assertTrue(row[0].endswith(u"reich"), "column must contain original data")
179
180 def CheckOptimizedUnicode(self):
181 self.con.text_factory = sqlite.OptimizedUnicode
182 austria = unicode("Österreich", "latin1")
183 germany = unicode("Deutchland")
184 a_row = self.con.execute("select ?", (austria,)).fetchone()
185 d_row = self.con.execute("select ?", (germany,)).fetchone()
186 self.assertTrue(type(a_row[0]) == unicode, "type of non-ASCII row must be unicode")
187 self.assertTrue(type(d_row[0]) == str, "type of ASCII-only row must be str")
188
189 def tearDown(self):
190 self.con.close()
191
192class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase):
193 def setUp(self):
194 self.con = sqlite.connect(":memory:")
195 self.con.execute("create table test (value text)")
196 self.con.execute("insert into test (value) values (?)", ("a\x00b",))
197
198 def CheckString(self):
199 # text_factory defaults to unicode
200 row = self.con.execute("select value from test").fetchone()
201 self.assertIs(type(row[0]), unicode)
202 self.assertEqual(row[0], "a\x00b")
203
204 def CheckCustom(self):
205 # A custom factory should receive an str argument
206 self.con.text_factory = lambda x: x
207 row = self.con.execute("select value from test").fetchone()
208 self.assertIs(type(row[0]), str)
209 self.assertEqual(row[0], "a\x00b")
210
211 def CheckOptimizedUnicodeAsString(self):
212 # ASCII -> str argument
213 self.con.text_factory = sqlite.OptimizedUnicode
214 row = self.con.execute("select value from test").fetchone()
215 self.assertIs(type(row[0]), str)
216 self.assertEqual(row[0], "a\x00b")
217
218 def CheckOptimizedUnicodeAsUnicode(self):
219 # Non-ASCII -> unicode argument
220 self.con.text_factory = sqlite.OptimizedUnicode
221 self.con.execute("delete from test")
222 self.con.execute("insert into test (value) values (?)", (u'ä\0ö',))
223 row = self.con.execute("select value from test").fetchone()
224 self.assertIs(type(row[0]), unicode)
225 self.assertEqual(row[0], u"ä\x00ö")
226
227 def tearDown(self):
228 self.con.close()
229
230def suite():
231 connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check")
232 cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check")
233 row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check")
234 row_suite = unittest.makeSuite(RowFactoryTests, "Check")
235 text_suite = unittest.makeSuite(TextFactoryTests, "Check")
236 text_zero_bytes_suite = unittest.makeSuite(TextFactoryTestsWithEmbeddedZeroBytes, "Check")
237 return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite, text_zero_bytes_suite))
238
239def test():
240 runner = unittest.TextTestRunner()
241 runner.run(suite())
242
243if __name__ == "__main__":
244 test()
Note: See TracBrowser for help on using the repository browser.