• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
26from collections.abc import Sequence
27
28class MyConnection(sqlite.Connection):
29    def __init__(self, *args, **kwargs):
30        sqlite.Connection.__init__(self, *args, **kwargs)
31
32def dict_factory(cursor, row):
33    d = {}
34    for idx, col in enumerate(cursor.description):
35        d[col[0]] = row[idx]
36    return d
37
38class MyCursor(sqlite.Cursor):
39    def __init__(self, *args, **kwargs):
40        sqlite.Cursor.__init__(self, *args, **kwargs)
41        self.row_factory = dict_factory
42
43class ConnectionFactoryTests(unittest.TestCase):
44    def setUp(self):
45        self.con = sqlite.connect(":memory:", factory=MyConnection)
46
47    def tearDown(self):
48        self.con.close()
49
50    def CheckIsInstance(self):
51        self.assertIsInstance(self.con, MyConnection)
52
53class CursorFactoryTests(unittest.TestCase):
54    def setUp(self):
55        self.con = sqlite.connect(":memory:")
56
57    def tearDown(self):
58        self.con.close()
59
60    def CheckIsInstance(self):
61        cur = self.con.cursor()
62        self.assertIsInstance(cur, sqlite.Cursor)
63        cur = self.con.cursor(MyCursor)
64        self.assertIsInstance(cur, MyCursor)
65        cur = self.con.cursor(factory=lambda con: MyCursor(con))
66        self.assertIsInstance(cur, MyCursor)
67
68    def CheckInvalidFactory(self):
69        # not a callable at all
70        self.assertRaises(TypeError, self.con.cursor, None)
71        # invalid callable with not exact one argument
72        self.assertRaises(TypeError, self.con.cursor, lambda: None)
73        # invalid callable returning non-cursor
74        self.assertRaises(TypeError, self.con.cursor, lambda con: None)
75
76class RowFactoryTestsBackwardsCompat(unittest.TestCase):
77    def setUp(self):
78        self.con = sqlite.connect(":memory:")
79
80    def CheckIsProducedByFactory(self):
81        cur = self.con.cursor(factory=MyCursor)
82        cur.execute("select 4+5 as foo")
83        row = cur.fetchone()
84        self.assertIsInstance(row, dict)
85        cur.close()
86
87    def tearDown(self):
88        self.con.close()
89
90class RowFactoryTests(unittest.TestCase):
91    def setUp(self):
92        self.con = sqlite.connect(":memory:")
93
94    def CheckCustomFactory(self):
95        self.con.row_factory = lambda cur, row: list(row)
96        row = self.con.execute("select 1, 2").fetchone()
97        self.assertIsInstance(row, list)
98
99    def CheckSqliteRowIndex(self):
100        self.con.row_factory = sqlite.Row
101        row = self.con.execute("select 1 as a_1, 2 as b").fetchone()
102        self.assertIsInstance(row, sqlite.Row)
103
104        self.assertEqual(row["a_1"], 1, "by name: wrong result for column 'a_1'")
105        self.assertEqual(row["b"], 2, "by name: wrong result for column 'b'")
106
107        self.assertEqual(row["A_1"], 1, "by name: wrong result for column 'A_1'")
108        self.assertEqual(row["B"], 2, "by name: wrong result for column 'B'")
109
110        self.assertEqual(row[0], 1, "by index: wrong result for column 0")
111        self.assertEqual(row[1], 2, "by index: wrong result for column 1")
112        self.assertEqual(row[-1], 2, "by index: wrong result for column -1")
113        self.assertEqual(row[-2], 1, "by index: wrong result for column -2")
114
115        with self.assertRaises(IndexError):
116            row['c']
117        with self.assertRaises(IndexError):
118            row['a_\x11']
119        with self.assertRaises(IndexError):
120            row['a\x7f1']
121        with self.assertRaises(IndexError):
122            row[2]
123        with self.assertRaises(IndexError):
124            row[-3]
125        with self.assertRaises(IndexError):
126            row[2**1000]
127
128    def CheckSqliteRowIndexUnicode(self):
129        self.con.row_factory = sqlite.Row
130        row = self.con.execute("select 1 as \xff").fetchone()
131        self.assertEqual(row["\xff"], 1)
132        with self.assertRaises(IndexError):
133            row['\u0178']
134        with self.assertRaises(IndexError):
135            row['\xdf']
136
137    def CheckSqliteRowSlice(self):
138        # A sqlite.Row can be sliced like a list.
139        self.con.row_factory = sqlite.Row
140        row = self.con.execute("select 1, 2, 3, 4").fetchone()
141        self.assertEqual(row[0:0], ())
142        self.assertEqual(row[0:1], (1,))
143        self.assertEqual(row[1:3], (2, 3))
144        self.assertEqual(row[3:1], ())
145        # Explicit bounds are optional.
146        self.assertEqual(row[1:], (2, 3, 4))
147        self.assertEqual(row[:3], (1, 2, 3))
148        # Slices can use negative indices.
149        self.assertEqual(row[-2:-1], (3,))
150        self.assertEqual(row[-2:], (3, 4))
151        # Slicing supports steps.
152        self.assertEqual(row[0:4:2], (1, 3))
153        self.assertEqual(row[3:0:-2], (4, 2))
154
155    def CheckSqliteRowIter(self):
156        """Checks if the row object is iterable"""
157        self.con.row_factory = sqlite.Row
158        row = self.con.execute("select 1 as a, 2 as b").fetchone()
159        for col in row:
160            pass
161
162    def CheckSqliteRowAsTuple(self):
163        """Checks if the row object can be converted to a tuple"""
164        self.con.row_factory = sqlite.Row
165        row = self.con.execute("select 1 as a, 2 as b").fetchone()
166        t = tuple(row)
167        self.assertEqual(t, (row['a'], row['b']))
168
169    def CheckSqliteRowAsDict(self):
170        """Checks if the row object can be correctly converted to a dictionary"""
171        self.con.row_factory = sqlite.Row
172        row = self.con.execute("select 1 as a, 2 as b").fetchone()
173        d = dict(row)
174        self.assertEqual(d["a"], row["a"])
175        self.assertEqual(d["b"], row["b"])
176
177    def CheckSqliteRowHashCmp(self):
178        """Checks if the row object compares and hashes correctly"""
179        self.con.row_factory = sqlite.Row
180        row_1 = self.con.execute("select 1 as a, 2 as b").fetchone()
181        row_2 = self.con.execute("select 1 as a, 2 as b").fetchone()
182        row_3 = self.con.execute("select 1 as a, 3 as b").fetchone()
183        row_4 = self.con.execute("select 1 as b, 2 as a").fetchone()
184        row_5 = self.con.execute("select 2 as b, 1 as a").fetchone()
185
186        self.assertTrue(row_1 == row_1)
187        self.assertTrue(row_1 == row_2)
188        self.assertFalse(row_1 == row_3)
189        self.assertFalse(row_1 == row_4)
190        self.assertFalse(row_1 == row_5)
191        self.assertFalse(row_1 == object())
192
193        self.assertFalse(row_1 != row_1)
194        self.assertFalse(row_1 != row_2)
195        self.assertTrue(row_1 != row_3)
196        self.assertTrue(row_1 != row_4)
197        self.assertTrue(row_1 != row_5)
198        self.assertTrue(row_1 != object())
199
200        with self.assertRaises(TypeError):
201            row_1 > row_2
202        with self.assertRaises(TypeError):
203            row_1 < row_2
204        with self.assertRaises(TypeError):
205            row_1 >= row_2
206        with self.assertRaises(TypeError):
207            row_1 <= row_2
208
209        self.assertEqual(hash(row_1), hash(row_2))
210
211    def CheckSqliteRowAsSequence(self):
212        """ Checks if the row object can act like a sequence """
213        self.con.row_factory = sqlite.Row
214        row = self.con.execute("select 1 as a, 2 as b").fetchone()
215
216        as_tuple = tuple(row)
217        self.assertEqual(list(reversed(row)), list(reversed(as_tuple)))
218        self.assertIsInstance(row, Sequence)
219
220    def CheckFakeCursorClass(self):
221        # Issue #24257: Incorrect use of PyObject_IsInstance() caused
222        # segmentation fault.
223        # Issue #27861: Also applies for cursor factory.
224        class FakeCursor(str):
225            __class__ = sqlite.Cursor
226        self.con.row_factory = sqlite.Row
227        self.assertRaises(TypeError, self.con.cursor, FakeCursor)
228        self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ())
229
230    def tearDown(self):
231        self.con.close()
232
233class TextFactoryTests(unittest.TestCase):
234    def setUp(self):
235        self.con = sqlite.connect(":memory:")
236
237    def CheckUnicode(self):
238        austria = "�sterreich"
239        row = self.con.execute("select ?", (austria,)).fetchone()
240        self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
241
242    def CheckString(self):
243        self.con.text_factory = bytes
244        austria = "�sterreich"
245        row = self.con.execute("select ?", (austria,)).fetchone()
246        self.assertEqual(type(row[0]), bytes, "type of row[0] must be bytes")
247        self.assertEqual(row[0], austria.encode("utf-8"), "column must equal original data in UTF-8")
248
249    def CheckCustom(self):
250        self.con.text_factory = lambda x: str(x, "utf-8", "ignore")
251        austria = "�sterreich"
252        row = self.con.execute("select ?", (austria,)).fetchone()
253        self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
254        self.assertTrue(row[0].endswith("reich"), "column must contain original data")
255
256    def CheckOptimizedUnicode(self):
257        # In py3k, str objects are always returned when text_factory
258        # is OptimizedUnicode
259        self.con.text_factory = sqlite.OptimizedUnicode
260        austria = "�sterreich"
261        germany = "Deutchland"
262        a_row = self.con.execute("select ?", (austria,)).fetchone()
263        d_row = self.con.execute("select ?", (germany,)).fetchone()
264        self.assertEqual(type(a_row[0]), str, "type of non-ASCII row must be str")
265        self.assertEqual(type(d_row[0]), str, "type of ASCII-only row must be str")
266
267    def tearDown(self):
268        self.con.close()
269
270class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase):
271    def setUp(self):
272        self.con = sqlite.connect(":memory:")
273        self.con.execute("create table test (value text)")
274        self.con.execute("insert into test (value) values (?)", ("a\x00b",))
275
276    def CheckString(self):
277        # text_factory defaults to str
278        row = self.con.execute("select value from test").fetchone()
279        self.assertIs(type(row[0]), str)
280        self.assertEqual(row[0], "a\x00b")
281
282    def CheckBytes(self):
283        self.con.text_factory = bytes
284        row = self.con.execute("select value from test").fetchone()
285        self.assertIs(type(row[0]), bytes)
286        self.assertEqual(row[0], b"a\x00b")
287
288    def CheckBytearray(self):
289        self.con.text_factory = bytearray
290        row = self.con.execute("select value from test").fetchone()
291        self.assertIs(type(row[0]), bytearray)
292        self.assertEqual(row[0], b"a\x00b")
293
294    def CheckCustom(self):
295        # A custom factory should receive a bytes argument
296        self.con.text_factory = lambda x: x
297        row = self.con.execute("select value from test").fetchone()
298        self.assertIs(type(row[0]), bytes)
299        self.assertEqual(row[0], b"a\x00b")
300
301    def tearDown(self):
302        self.con.close()
303
304def suite():
305    connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check")
306    cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check")
307    row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check")
308    row_suite = unittest.makeSuite(RowFactoryTests, "Check")
309    text_suite = unittest.makeSuite(TextFactoryTests, "Check")
310    text_zero_bytes_suite = unittest.makeSuite(TextFactoryTestsWithEmbeddedZeroBytes, "Check")
311    return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite, text_zero_bytes_suite))
312
313def test():
314    runner = unittest.TextTestRunner()
315    runner.run(suite())
316
317if __name__ == "__main__":
318    test()
319