1# pysqlite2/test/factory.py: tests for the various factories in pysqlite
2#
3# Copyright (C) 2005-2007 Gerhard Häring <[email protected]>
4#
5# This file is part of pysqlite.
6#
7# This software is provided 'as-is', without any express or implied
8# warranty.  In no event will the authors be held liable for any damages
9# arising from the use of this software.
10#
11# Permission is granted to anyone to use this software for any purpose,
12# including commercial applications, and to alter it and redistribute it
13# freely, subject to the following restrictions:
14#
15# 1. The origin of this software must not be misrepresented; you must not
16#    claim that you wrote the original software. If you use this software
17#    in a product, an acknowledgment in the product documentation would be
18#    appreciated but is not required.
19# 2. Altered source versions must be plainly marked as such, and must not be
20#    misrepresented as being the original software.
21# 3. This notice may not be removed or altered from any source distribution.
22
23import unittest
24import sqlite3 as sqlite
25from collections.abc import Sequence
26
27
28def dict_factory(cursor, row):
29    d = {}
30    for idx, col in enumerate(cursor.description):
31        d[col[0]] = row[idx]
32    return d
33
34class MyCursor(sqlite.Cursor):
35    def __init__(self, *args, **kwargs):
36        sqlite.Cursor.__init__(self, *args, **kwargs)
37        self.row_factory = dict_factory
38
39class ConnectionFactoryTests(unittest.TestCase):
40    def test_connection_factories(self):
41        class DefectFactory(sqlite.Connection):
42            def __init__(self, *args, **kwargs):
43                return None
44        class OkFactory(sqlite.Connection):
45            def __init__(self, *args, **kwargs):
46                sqlite.Connection.__init__(self, *args, **kwargs)
47
48        for factory in DefectFactory, OkFactory:
49            with self.subTest(factory=factory):
50                con = sqlite.connect(":memory:", factory=factory)
51                self.assertIsInstance(con, factory)
52
53    def test_connection_factory_relayed_call(self):
54        # gh-95132: keyword args must not be passed as positional args
55        class Factory(sqlite.Connection):
56            def __init__(self, *args, **kwargs):
57                kwargs["isolation_level"] = None
58                super(Factory, self).__init__(*args, **kwargs)
59
60        con = sqlite.connect(":memory:", factory=Factory)
61        self.assertIsNone(con.isolation_level)
62        self.assertIsInstance(con, Factory)
63
64    def test_connection_factory_as_positional_arg(self):
65        class Factory(sqlite.Connection):
66            def __init__(self, *args, **kwargs):
67                super(Factory, self).__init__(*args, **kwargs)
68
69        con = sqlite.connect(":memory:", 5.0, 0, None, True, Factory)
70        self.assertIsNone(con.isolation_level)
71        self.assertIsInstance(con, Factory)
72
73
74class CursorFactoryTests(unittest.TestCase):
75    def setUp(self):
76        self.con = sqlite.connect(":memory:")
77
78    def tearDown(self):
79        self.con.close()
80
81    def test_is_instance(self):
82        cur = self.con.cursor()
83        self.assertIsInstance(cur, sqlite.Cursor)
84        cur = self.con.cursor(MyCursor)
85        self.assertIsInstance(cur, MyCursor)
86        cur = self.con.cursor(factory=lambda con: MyCursor(con))
87        self.assertIsInstance(cur, MyCursor)
88
89    def test_invalid_factory(self):
90        # not a callable at all
91        self.assertRaises(TypeError, self.con.cursor, None)
92        # invalid callable with not exact one argument
93        self.assertRaises(TypeError, self.con.cursor, lambda: None)
94        # invalid callable returning non-cursor
95        self.assertRaises(TypeError, self.con.cursor, lambda con: None)
96
97class RowFactoryTestsBackwardsCompat(unittest.TestCase):
98    def setUp(self):
99        self.con = sqlite.connect(":memory:")
100
101    def test_is_produced_by_factory(self):
102        cur = self.con.cursor(factory=MyCursor)
103        cur.execute("select 4+5 as foo")
104        row = cur.fetchone()
105        self.assertIsInstance(row, dict)
106        cur.close()
107
108    def tearDown(self):
109        self.con.close()
110
111class RowFactoryTests(unittest.TestCase):
112    def setUp(self):
113        self.con = sqlite.connect(":memory:")
114
115    def test_custom_factory(self):
116        self.con.row_factory = lambda cur, row: list(row)
117        row = self.con.execute("select 1, 2").fetchone()
118        self.assertIsInstance(row, list)
119
120    def test_sqlite_row_index(self):
121        self.con.row_factory = sqlite.Row
122        row = self.con.execute("select 1 as a_1, 2 as b").fetchone()
123        self.assertIsInstance(row, sqlite.Row)
124
125        self.assertEqual(row["a_1"], 1, "by name: wrong result for column 'a_1'")
126        self.assertEqual(row["b"], 2, "by name: wrong result for column 'b'")
127
128        self.assertEqual(row["A_1"], 1, "by name: wrong result for column 'A_1'")
129        self.assertEqual(row["B"], 2, "by name: wrong result for column 'B'")
130
131        self.assertEqual(row[0], 1, "by index: wrong result for column 0")
132        self.assertEqual(row[1], 2, "by index: wrong result for column 1")
133        self.assertEqual(row[-1], 2, "by index: wrong result for column -1")
134        self.assertEqual(row[-2], 1, "by index: wrong result for column -2")
135
136        with self.assertRaises(IndexError):
137            row['c']
138        with self.assertRaises(IndexError):
139            row['a_\x11']
140        with self.assertRaises(IndexError):
141            row['a\x7f1']
142        with self.assertRaises(IndexError):
143            row[2]
144        with self.assertRaises(IndexError):
145            row[-3]
146        with self.assertRaises(IndexError):
147            row[2**1000]
148        with self.assertRaises(IndexError):
149            row[complex()]  # index must be int or string
150
151    def test_sqlite_row_index_unicode(self):
152        self.con.row_factory = sqlite.Row
153        row = self.con.execute("select 1 as \xff").fetchone()
154        self.assertEqual(row["\xff"], 1)
155        with self.assertRaises(IndexError):
156            row['\u0178']
157        with self.assertRaises(IndexError):
158            row['\xdf']
159
160    def test_sqlite_row_slice(self):
161        # A sqlite.Row can be sliced like a list.
162        self.con.row_factory = sqlite.Row
163        row = self.con.execute("select 1, 2, 3, 4").fetchone()
164        self.assertEqual(row[0:0], ())
165        self.assertEqual(row[0:1], (1,))
166        self.assertEqual(row[1:3], (2, 3))
167        self.assertEqual(row[3:1], ())
168        # Explicit bounds are optional.
169        self.assertEqual(row[1:], (2, 3, 4))
170        self.assertEqual(row[:3], (1, 2, 3))
171        # Slices can use negative indices.
172        self.assertEqual(row[-2:-1], (3,))
173        self.assertEqual(row[-2:], (3, 4))
174        # Slicing supports steps.
175        self.assertEqual(row[0:4:2], (1, 3))
176        self.assertEqual(row[3:0:-2], (4, 2))
177
178    def test_sqlite_row_iter(self):
179        """Checks if the row object is iterable"""
180        self.con.row_factory = sqlite.Row
181        row = self.con.execute("select 1 as a, 2 as b").fetchone()
182
183        # Is iterable in correct order and produces valid results:
184        items = [col for col in row]
185        self.assertEqual(items, [1, 2])
186
187        # Is iterable the second time:
188        items = [col for col in row]
189        self.assertEqual(items, [1, 2])
190
191    def test_sqlite_row_as_tuple(self):
192        """Checks if the row object can be converted to a tuple"""
193        self.con.row_factory = sqlite.Row
194        row = self.con.execute("select 1 as a, 2 as b").fetchone()
195        t = tuple(row)
196        self.assertEqual(t, (row['a'], row['b']))
197
198    def test_sqlite_row_as_dict(self):
199        """Checks if the row object can be correctly converted to a dictionary"""
200        self.con.row_factory = sqlite.Row
201        row = self.con.execute("select 1 as a, 2 as b").fetchone()
202        d = dict(row)
203        self.assertEqual(d["a"], row["a"])
204        self.assertEqual(d["b"], row["b"])
205
206    def test_sqlite_row_hash_cmp(self):
207        """Checks if the row object compares and hashes correctly"""
208        self.con.row_factory = sqlite.Row
209        row_1 = self.con.execute("select 1 as a, 2 as b").fetchone()
210        row_2 = self.con.execute("select 1 as a, 2 as b").fetchone()
211        row_3 = self.con.execute("select 1 as a, 3 as b").fetchone()
212        row_4 = self.con.execute("select 1 as b, 2 as a").fetchone()
213        row_5 = self.con.execute("select 2 as b, 1 as a").fetchone()
214
215        self.assertTrue(row_1 == row_1)
216        self.assertTrue(row_1 == row_2)
217        self.assertFalse(row_1 == row_3)
218        self.assertFalse(row_1 == row_4)
219        self.assertFalse(row_1 == row_5)
220        self.assertFalse(row_1 == object())
221
222        self.assertFalse(row_1 != row_1)
223        self.assertFalse(row_1 != row_2)
224        self.assertTrue(row_1 != row_3)
225        self.assertTrue(row_1 != row_4)
226        self.assertTrue(row_1 != row_5)
227        self.assertTrue(row_1 != object())
228
229        with self.assertRaises(TypeError):
230            row_1 > row_2
231        with self.assertRaises(TypeError):
232            row_1 < row_2
233        with self.assertRaises(TypeError):
234            row_1 >= row_2
235        with self.assertRaises(TypeError):
236            row_1 <= row_2
237
238        self.assertEqual(hash(row_1), hash(row_2))
239
240    def test_sqlite_row_as_sequence(self):
241        """ Checks if the row object can act like a sequence """
242        self.con.row_factory = sqlite.Row
243        row = self.con.execute("select 1 as a, 2 as b").fetchone()
244
245        as_tuple = tuple(row)
246        self.assertEqual(list(reversed(row)), list(reversed(as_tuple)))
247        self.assertIsInstance(row, Sequence)
248
249    def test_fake_cursor_class(self):
250        # Issue #24257: Incorrect use of PyObject_IsInstance() caused
251        # segmentation fault.
252        # Issue #27861: Also applies for cursor factory.
253        class FakeCursor(str):
254            __class__ = sqlite.Cursor
255        self.con.row_factory = sqlite.Row
256        self.assertRaises(TypeError, self.con.cursor, FakeCursor)
257        self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ())
258
259    def tearDown(self):
260        self.con.close()
261
262class TextFactoryTests(unittest.TestCase):
263    def setUp(self):
264        self.con = sqlite.connect(":memory:")
265
266    def test_unicode(self):
267        austria = "Österreich"
268        row = self.con.execute("select ?", (austria,)).fetchone()
269        self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
270
271    def test_string(self):
272        self.con.text_factory = bytes
273        austria = "Österreich"
274        row = self.con.execute("select ?", (austria,)).fetchone()
275        self.assertEqual(type(row[0]), bytes, "type of row[0] must be bytes")
276        self.assertEqual(row[0], austria.encode("utf-8"), "column must equal original data in UTF-8")
277
278    def test_custom(self):
279        self.con.text_factory = lambda x: str(x, "utf-8", "ignore")
280        austria = "Österreich"
281        row = self.con.execute("select ?", (austria,)).fetchone()
282        self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
283        self.assertTrue(row[0].endswith("reich"), "column must contain original data")
284
285    def test_optimized_unicode(self):
286        # OptimizedUnicode is deprecated as of Python 3.10
287        with self.assertWarns(DeprecationWarning) as cm:
288            self.con.text_factory = sqlite.OptimizedUnicode
289        self.assertIn("factory.py", cm.filename)
290        austria = "Österreich"
291        germany = "Deutchland"
292        a_row = self.con.execute("select ?", (austria,)).fetchone()
293        d_row = self.con.execute("select ?", (germany,)).fetchone()
294        self.assertEqual(type(a_row[0]), str, "type of non-ASCII row must be str")
295        self.assertEqual(type(d_row[0]), str, "type of ASCII-only row must be str")
296
297    def tearDown(self):
298        self.con.close()
299
300class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase):
301    def setUp(self):
302        self.con = sqlite.connect(":memory:")
303        self.con.execute("create table test (value text)")
304        self.con.execute("insert into test (value) values (?)", ("a\x00b",))
305
306    def test_string(self):
307        # text_factory defaults to str
308        row = self.con.execute("select value from test").fetchone()
309        self.assertIs(type(row[0]), str)
310        self.assertEqual(row[0], "a\x00b")
311
312    def test_bytes(self):
313        self.con.text_factory = bytes
314        row = self.con.execute("select value from test").fetchone()
315        self.assertIs(type(row[0]), bytes)
316        self.assertEqual(row[0], b"a\x00b")
317
318    def test_bytearray(self):
319        self.con.text_factory = bytearray
320        row = self.con.execute("select value from test").fetchone()
321        self.assertIs(type(row[0]), bytearray)
322        self.assertEqual(row[0], b"a\x00b")
323
324    def test_custom(self):
325        # A custom factory should receive a bytes argument
326        self.con.text_factory = lambda x: x
327        row = self.con.execute("select value from test").fetchone()
328        self.assertIs(type(row[0]), bytes)
329        self.assertEqual(row[0], b"a\x00b")
330
331    def tearDown(self):
332        self.con.close()
333
334
335if __name__ == "__main__":
336    unittest.main()
337