»Core Development>Code coverage>Lib/sqlite3/test/factory.py

Python code coverage for Lib/sqlite3/test/factory.py

#countcontent
1n/a#-*- coding: iso-8859-1 -*-
2n/a# pysqlite2/test/factory.py: tests for the various factories in pysqlite
3n/a#
4n/a# Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de>
5n/a#
6n/a# This file is part of pysqlite.
7n/a#
8n/a# This software is provided 'as-is', without any express or implied
9n/a# warranty. In no event will the authors be held liable for any damages
10n/a# arising from the use of this software.
11n/a#
12n/a# Permission is granted to anyone to use this software for any purpose,
13n/a# including commercial applications, and to alter it and redistribute it
14n/a# freely, subject to the following restrictions:
15n/a#
16n/a# 1. The origin of this software must not be misrepresented; you must not
17n/a# claim that you wrote the original software. If you use this software
18n/a# in a product, an acknowledgment in the product documentation would be
19n/a# appreciated but is not required.
20n/a# 2. Altered source versions must be plainly marked as such, and must not be
21n/a# misrepresented as being the original software.
22n/a# 3. This notice may not be removed or altered from any source distribution.
23n/a
24n/aimport unittest
25n/aimport sqlite3 as sqlite
26n/afrom collections.abc import Sequence
27n/a
28n/aclass MyConnection(sqlite.Connection):
29n/a def __init__(self, *args, **kwargs):
30n/a sqlite.Connection.__init__(self, *args, **kwargs)
31n/a
32n/adef dict_factory(cursor, row):
33n/a d = {}
34n/a for idx, col in enumerate(cursor.description):
35n/a d[col[0]] = row[idx]
36n/a return d
37n/a
38n/aclass MyCursor(sqlite.Cursor):
39n/a def __init__(self, *args, **kwargs):
40n/a sqlite.Cursor.__init__(self, *args, **kwargs)
41n/a self.row_factory = dict_factory
42n/a
43n/aclass ConnectionFactoryTests(unittest.TestCase):
44n/a def setUp(self):
45n/a self.con = sqlite.connect(":memory:", factory=MyConnection)
46n/a
47n/a def tearDown(self):
48n/a self.con.close()
49n/a
50n/a def CheckIsInstance(self):
51n/a self.assertIsInstance(self.con, MyConnection)
52n/a
53n/aclass CursorFactoryTests(unittest.TestCase):
54n/a def setUp(self):
55n/a self.con = sqlite.connect(":memory:")
56n/a
57n/a def tearDown(self):
58n/a self.con.close()
59n/a
60n/a def CheckIsInstance(self):
61n/a cur = self.con.cursor()
62n/a self.assertIsInstance(cur, sqlite.Cursor)
63n/a cur = self.con.cursor(MyCursor)
64n/a self.assertIsInstance(cur, MyCursor)
65n/a cur = self.con.cursor(factory=lambda con: MyCursor(con))
66n/a self.assertIsInstance(cur, MyCursor)
67n/a
68n/a def CheckInvalidFactory(self):
69n/a # not a callable at all
70n/a self.assertRaises(TypeError, self.con.cursor, None)
71n/a # invalid callable with not exact one argument
72n/a self.assertRaises(TypeError, self.con.cursor, lambda: None)
73n/a # invalid callable returning non-cursor
74n/a self.assertRaises(TypeError, self.con.cursor, lambda con: None)
75n/a
76n/aclass RowFactoryTestsBackwardsCompat(unittest.TestCase):
77n/a def setUp(self):
78n/a self.con = sqlite.connect(":memory:")
79n/a
80n/a def CheckIsProducedByFactory(self):
81n/a cur = self.con.cursor(factory=MyCursor)
82n/a cur.execute("select 4+5 as foo")
83n/a row = cur.fetchone()
84n/a self.assertIsInstance(row, dict)
85n/a cur.close()
86n/a
87n/a def tearDown(self):
88n/a self.con.close()
89n/a
90n/aclass RowFactoryTests(unittest.TestCase):
91n/a def setUp(self):
92n/a self.con = sqlite.connect(":memory:")
93n/a
94n/a def CheckCustomFactory(self):
95n/a self.con.row_factory = lambda cur, row: list(row)
96n/a row = self.con.execute("select 1, 2").fetchone()
97n/a self.assertIsInstance(row, list)
98n/a
99n/a def CheckSqliteRowIndex(self):
100n/a self.con.row_factory = sqlite.Row
101n/a row = self.con.execute("select 1 as a, 2 as b").fetchone()
102n/a self.assertIsInstance(row, sqlite.Row)
103n/a
104n/a col1, col2 = row["a"], row["b"]
105n/a self.assertEqual(col1, 1, "by name: wrong result for column 'a'")
106n/a self.assertEqual(col2, 2, "by name: wrong result for column 'a'")
107n/a
108n/a col1, col2 = row["A"], row["B"]
109n/a self.assertEqual(col1, 1, "by name: wrong result for column 'A'")
110n/a self.assertEqual(col2, 2, "by name: wrong result for column 'B'")
111n/a
112n/a self.assertEqual(row[0], 1, "by index: wrong result for column 0")
113n/a self.assertEqual(row[1], 2, "by index: wrong result for column 1")
114n/a self.assertEqual(row[-1], 2, "by index: wrong result for column -1")
115n/a self.assertEqual(row[-2], 1, "by index: wrong result for column -2")
116n/a
117n/a with self.assertRaises(IndexError):
118n/a row['c']
119n/a with self.assertRaises(IndexError):
120n/a row[2]
121n/a with self.assertRaises(IndexError):
122n/a row[-3]
123n/a with self.assertRaises(IndexError):
124n/a row[2**1000]
125n/a
126n/a def CheckSqliteRowSlice(self):
127n/a # A sqlite.Row can be sliced like a list.
128n/a self.con.row_factory = sqlite.Row
129n/a row = self.con.execute("select 1, 2, 3, 4").fetchone()
130n/a self.assertEqual(row[0:0], ())
131n/a self.assertEqual(row[0:1], (1,))
132n/a self.assertEqual(row[1:3], (2, 3))
133n/a self.assertEqual(row[3:1], ())
134n/a # Explicit bounds are optional.
135n/a self.assertEqual(row[1:], (2, 3, 4))
136n/a self.assertEqual(row[:3], (1, 2, 3))
137n/a # Slices can use negative indices.
138n/a self.assertEqual(row[-2:-1], (3,))
139n/a self.assertEqual(row[-2:], (3, 4))
140n/a # Slicing supports steps.
141n/a self.assertEqual(row[0:4:2], (1, 3))
142n/a self.assertEqual(row[3:0:-2], (4, 2))
143n/a
144n/a def CheckSqliteRowIter(self):
145n/a """Checks if the row object is iterable"""
146n/a self.con.row_factory = sqlite.Row
147n/a row = self.con.execute("select 1 as a, 2 as b").fetchone()
148n/a for col in row:
149n/a pass
150n/a
151n/a def CheckSqliteRowAsTuple(self):
152n/a """Checks if the row object can be converted to a tuple"""
153n/a self.con.row_factory = sqlite.Row
154n/a row = self.con.execute("select 1 as a, 2 as b").fetchone()
155n/a t = tuple(row)
156n/a self.assertEqual(t, (row['a'], row['b']))
157n/a
158n/a def CheckSqliteRowAsDict(self):
159n/a """Checks if the row object can be correctly converted to a dictionary"""
160n/a self.con.row_factory = sqlite.Row
161n/a row = self.con.execute("select 1 as a, 2 as b").fetchone()
162n/a d = dict(row)
163n/a self.assertEqual(d["a"], row["a"])
164n/a self.assertEqual(d["b"], row["b"])
165n/a
166n/a def CheckSqliteRowHashCmp(self):
167n/a """Checks if the row object compares and hashes correctly"""
168n/a self.con.row_factory = sqlite.Row
169n/a row_1 = self.con.execute("select 1 as a, 2 as b").fetchone()
170n/a row_2 = self.con.execute("select 1 as a, 2 as b").fetchone()
171n/a row_3 = self.con.execute("select 1 as a, 3 as b").fetchone()
172n/a
173n/a self.assertEqual(row_1, row_1)
174n/a self.assertEqual(row_1, row_2)
175n/a self.assertTrue(row_2 != row_3)
176n/a
177n/a self.assertFalse(row_1 != row_1)
178n/a self.assertFalse(row_1 != row_2)
179n/a self.assertFalse(row_2 == row_3)
180n/a
181n/a self.assertEqual(row_1, row_2)
182n/a self.assertEqual(hash(row_1), hash(row_2))
183n/a self.assertNotEqual(row_1, row_3)
184n/a self.assertNotEqual(hash(row_1), hash(row_3))
185n/a
186n/a def CheckSqliteRowAsSequence(self):
187n/a """ Checks if the row object can act like a sequence """
188n/a self.con.row_factory = sqlite.Row
189n/a row = self.con.execute("select 1 as a, 2 as b").fetchone()
190n/a
191n/a as_tuple = tuple(row)
192n/a self.assertEqual(list(reversed(row)), list(reversed(as_tuple)))
193n/a self.assertIsInstance(row, Sequence)
194n/a
195n/a def CheckFakeCursorClass(self):
196n/a # Issue #24257: Incorrect use of PyObject_IsInstance() caused
197n/a # segmentation fault.
198n/a # Issue #27861: Also applies for cursor factory.
199n/a class FakeCursor(str):
200n/a __class__ = sqlite.Cursor
201n/a self.con.row_factory = sqlite.Row
202n/a self.assertRaises(TypeError, self.con.cursor, FakeCursor)
203n/a self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ())
204n/a
205n/a def tearDown(self):
206n/a self.con.close()
207n/a
208n/aclass TextFactoryTests(unittest.TestCase):
209n/a def setUp(self):
210n/a self.con = sqlite.connect(":memory:")
211n/a
212n/a def CheckUnicode(self):
213n/a austria = "Österreich"
214n/a row = self.con.execute("select ?", (austria,)).fetchone()
215n/a self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
216n/a
217n/a def CheckString(self):
218n/a self.con.text_factory = bytes
219n/a austria = "Österreich"
220n/a row = self.con.execute("select ?", (austria,)).fetchone()
221n/a self.assertEqual(type(row[0]), bytes, "type of row[0] must be bytes")
222n/a self.assertEqual(row[0], austria.encode("utf-8"), "column must equal original data in UTF-8")
223n/a
224n/a def CheckCustom(self):
225n/a self.con.text_factory = lambda x: str(x, "utf-8", "ignore")
226n/a austria = "Österreich"
227n/a row = self.con.execute("select ?", (austria,)).fetchone()
228n/a self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
229n/a self.assertTrue(row[0].endswith("reich"), "column must contain original data")
230n/a
231n/a def CheckOptimizedUnicode(self):
232n/a # In py3k, str objects are always returned when text_factory
233n/a # is OptimizedUnicode
234n/a self.con.text_factory = sqlite.OptimizedUnicode
235n/a austria = "Österreich"
236n/a germany = "Deutchland"
237n/a a_row = self.con.execute("select ?", (austria,)).fetchone()
238n/a d_row = self.con.execute("select ?", (germany,)).fetchone()
239n/a self.assertEqual(type(a_row[0]), str, "type of non-ASCII row must be str")
240n/a self.assertEqual(type(d_row[0]), str, "type of ASCII-only row must be str")
241n/a
242n/a def tearDown(self):
243n/a self.con.close()
244n/a
245n/aclass TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase):
246n/a def setUp(self):
247n/a self.con = sqlite.connect(":memory:")
248n/a self.con.execute("create table test (value text)")
249n/a self.con.execute("insert into test (value) values (?)", ("a\x00b",))
250n/a
251n/a def CheckString(self):
252n/a # text_factory defaults to str
253n/a row = self.con.execute("select value from test").fetchone()
254n/a self.assertIs(type(row[0]), str)
255n/a self.assertEqual(row[0], "a\x00b")
256n/a
257n/a def CheckBytes(self):
258n/a self.con.text_factory = bytes
259n/a row = self.con.execute("select value from test").fetchone()
260n/a self.assertIs(type(row[0]), bytes)
261n/a self.assertEqual(row[0], b"a\x00b")
262n/a
263n/a def CheckBytearray(self):
264n/a self.con.text_factory = bytearray
265n/a row = self.con.execute("select value from test").fetchone()
266n/a self.assertIs(type(row[0]), bytearray)
267n/a self.assertEqual(row[0], b"a\x00b")
268n/a
269n/a def CheckCustom(self):
270n/a # A custom factory should receive a bytes argument
271n/a self.con.text_factory = lambda x: x
272n/a row = self.con.execute("select value from test").fetchone()
273n/a self.assertIs(type(row[0]), bytes)
274n/a self.assertEqual(row[0], b"a\x00b")
275n/a
276n/a def tearDown(self):
277n/a self.con.close()
278n/a
279n/adef suite():
280n/a connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check")
281n/a cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check")
282n/a row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check")
283n/a row_suite = unittest.makeSuite(RowFactoryTests, "Check")
284n/a text_suite = unittest.makeSuite(TextFactoryTests, "Check")
285n/a text_zero_bytes_suite = unittest.makeSuite(TextFactoryTestsWithEmbeddedZeroBytes, "Check")
286n/a return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite, text_zero_bytes_suite))
287n/a
288n/adef test():
289n/a runner = unittest.TextTestRunner()
290n/a runner.run(suite())
291n/a
292n/aif __name__ == "__main__":
293n/a test()