Petri Lehtinen | f854799 | 2012-02-02 17:17:36 +0200 | [diff] [blame] | 1 | #-*- coding: iso-8859-1 -*- |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 2 | # pysqlite2/test/factory.py: tests for the various factories in pysqlite |
| 3 | # |
Gerhard Häring | e7ea745 | 2008-03-29 00:45:29 +0000 | [diff] [blame] | 4 | # Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de> |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 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 | |
| 24 | import unittest |
| 25 | import sqlite3 as sqlite |
| 26 | |
| 27 | class MyConnection(sqlite.Connection): |
| 28 | def __init__(self, *args, **kwargs): |
| 29 | sqlite.Connection.__init__(self, *args, **kwargs) |
| 30 | |
| 31 | def dict_factory(cursor, row): |
| 32 | d = {} |
| 33 | for idx, col in enumerate(cursor.description): |
| 34 | d[col[0]] = row[idx] |
| 35 | return d |
| 36 | |
| 37 | class MyCursor(sqlite.Cursor): |
| 38 | def __init__(self, *args, **kwargs): |
| 39 | sqlite.Cursor.__init__(self, *args, **kwargs) |
| 40 | self.row_factory = dict_factory |
| 41 | |
| 42 | class 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): |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 50 | self.assertIsInstance(self.con, MyConnection) |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 51 | |
| 52 | class CursorFactoryTests(unittest.TestCase): |
| 53 | def setUp(self): |
| 54 | self.con = sqlite.connect(":memory:") |
| 55 | |
| 56 | def tearDown(self): |
| 57 | self.con.close() |
| 58 | |
| 59 | def CheckIsInstance(self): |
| 60 | cur = self.con.cursor(factory=MyCursor) |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 61 | self.assertIsInstance(cur, MyCursor) |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 62 | |
| 63 | class RowFactoryTestsBackwardsCompat(unittest.TestCase): |
| 64 | def setUp(self): |
| 65 | self.con = sqlite.connect(":memory:") |
| 66 | |
| 67 | def CheckIsProducedByFactory(self): |
| 68 | cur = self.con.cursor(factory=MyCursor) |
| 69 | cur.execute("select 4+5 as foo") |
| 70 | row = cur.fetchone() |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 71 | self.assertIsInstance(row, dict) |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 72 | cur.close() |
| 73 | |
| 74 | def tearDown(self): |
| 75 | self.con.close() |
| 76 | |
| 77 | class RowFactoryTests(unittest.TestCase): |
| 78 | def setUp(self): |
| 79 | self.con = sqlite.connect(":memory:") |
| 80 | |
| 81 | def CheckCustomFactory(self): |
| 82 | self.con.row_factory = lambda cur, row: list(row) |
| 83 | row = self.con.execute("select 1, 2").fetchone() |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 84 | self.assertIsInstance(row, list) |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 85 | |
Thomas Wouters | fc7bb8c | 2007-01-15 15:49:28 +0000 | [diff] [blame] | 86 | def CheckSqliteRowIndex(self): |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 87 | self.con.row_factory = sqlite.Row |
| 88 | row = self.con.execute("select 1 as a, 2 as b").fetchone() |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 89 | self.assertIsInstance(row, sqlite.Row) |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 90 | |
| 91 | col1, col2 = row["a"], row["b"] |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 92 | self.assertEqual(col1, 1, "by name: wrong result for column 'a'") |
| 93 | self.assertEqual(col2, 2, "by name: wrong result for column 'a'") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 94 | |
| 95 | col1, col2 = row["A"], row["B"] |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 96 | self.assertEqual(col1, 1, "by name: wrong result for column 'A'") |
| 97 | self.assertEqual(col2, 2, "by name: wrong result for column 'B'") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 98 | |
| 99 | col1, col2 = row[0], row[1] |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 100 | self.assertEqual(col1, 1, "by index: wrong result for column 0") |
| 101 | self.assertEqual(col2, 2, "by index: wrong result for column 1") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 102 | |
Thomas Wouters | fc7bb8c | 2007-01-15 15:49:28 +0000 | [diff] [blame] | 103 | def CheckSqliteRowIter(self): |
| 104 | """Checks if the row object is iterable""" |
| 105 | self.con.row_factory = sqlite.Row |
| 106 | row = self.con.execute("select 1 as a, 2 as b").fetchone() |
| 107 | for col in row: |
| 108 | pass |
| 109 | |
| 110 | def CheckSqliteRowAsTuple(self): |
| 111 | """Checks if the row object can be converted to a tuple""" |
| 112 | self.con.row_factory = sqlite.Row |
| 113 | row = self.con.execute("select 1 as a, 2 as b").fetchone() |
| 114 | t = tuple(row) |
| 115 | |
| 116 | def CheckSqliteRowAsDict(self): |
| 117 | """Checks if the row object can be correctly converted to a dictionary""" |
| 118 | self.con.row_factory = sqlite.Row |
| 119 | row = self.con.execute("select 1 as a, 2 as b").fetchone() |
| 120 | d = dict(row) |
Gregory P. Smith | 04cecaf | 2009-07-04 08:32:15 +0000 | [diff] [blame] | 121 | self.assertEqual(d["a"], row["a"]) |
| 122 | self.assertEqual(d["b"], row["b"]) |
Thomas Wouters | fc7bb8c | 2007-01-15 15:49:28 +0000 | [diff] [blame] | 123 | |
Christian Heimes | 81ee3ef | 2008-05-04 22:42:01 +0000 | [diff] [blame] | 124 | def CheckSqliteRowHashCmp(self): |
| 125 | """Checks if the row object compares and hashes correctly""" |
| 126 | self.con.row_factory = sqlite.Row |
| 127 | row_1 = self.con.execute("select 1 as a, 2 as b").fetchone() |
| 128 | row_2 = self.con.execute("select 1 as a, 2 as b").fetchone() |
| 129 | row_3 = self.con.execute("select 1 as a, 3 as b").fetchone() |
| 130 | |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 131 | self.assertEqual(row_1, row_1) |
| 132 | self.assertEqual(row_1, row_2) |
Gregory P. Smith | 04cecaf | 2009-07-04 08:32:15 +0000 | [diff] [blame] | 133 | self.assertTrue(row_2 != row_3) |
Christian Heimes | 81ee3ef | 2008-05-04 22:42:01 +0000 | [diff] [blame] | 134 | |
Gregory P. Smith | 04cecaf | 2009-07-04 08:32:15 +0000 | [diff] [blame] | 135 | self.assertFalse(row_1 != row_1) |
| 136 | self.assertFalse(row_1 != row_2) |
| 137 | self.assertFalse(row_2 == row_3) |
Christian Heimes | 81ee3ef | 2008-05-04 22:42:01 +0000 | [diff] [blame] | 138 | |
Gregory P. Smith | 04cecaf | 2009-07-04 08:32:15 +0000 | [diff] [blame] | 139 | self.assertEqual(row_1, row_2) |
| 140 | self.assertEqual(hash(row_1), hash(row_2)) |
| 141 | self.assertNotEqual(row_1, row_3) |
| 142 | self.assertNotEqual(hash(row_1), hash(row_3)) |
Christian Heimes | 81ee3ef | 2008-05-04 22:42:01 +0000 | [diff] [blame] | 143 | |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 144 | def tearDown(self): |
| 145 | self.con.close() |
| 146 | |
| 147 | class TextFactoryTests(unittest.TestCase): |
| 148 | def setUp(self): |
| 149 | self.con = sqlite.connect(":memory:") |
| 150 | |
| 151 | def CheckUnicode(self): |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 152 | austria = "Österreich" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 153 | row = self.con.execute("select ?", (austria,)).fetchone() |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 154 | self.assertEqual(type(row[0]), str, "type of row[0] must be unicode") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 155 | |
| 156 | def CheckString(self): |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 157 | self.con.text_factory = bytes |
| 158 | austria = "Österreich" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 159 | row = self.con.execute("select ?", (austria,)).fetchone() |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 160 | self.assertEqual(type(row[0]), bytes, "type of row[0] must be bytes") |
| 161 | self.assertEqual(row[0], austria.encode("utf-8"), "column must equal original data in UTF-8") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 162 | |
| 163 | def CheckCustom(self): |
Guido van Rossum | ef87d6e | 2007-05-02 19:09:54 +0000 | [diff] [blame] | 164 | self.con.text_factory = lambda x: str(x, "utf-8", "ignore") |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 165 | austria = "Österreich" |
| 166 | row = self.con.execute("select ?", (austria,)).fetchone() |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 167 | self.assertEqual(type(row[0]), str, "type of row[0] must be unicode") |
Gregory P. Smith | 04cecaf | 2009-07-04 08:32:15 +0000 | [diff] [blame] | 168 | self.assertTrue(row[0].endswith("reich"), "column must contain original data") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 169 | |
| 170 | def CheckOptimizedUnicode(self): |
Petri Lehtinen | bc35beb | 2012-02-09 21:09:03 +0200 | [diff] [blame] | 171 | # In py3k, str objects are always returned when text_factory |
| 172 | # is OptimizedUnicode |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 173 | self.con.text_factory = sqlite.OptimizedUnicode |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 174 | austria = "Österreich" |
| 175 | germany = "Deutchland" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 176 | a_row = self.con.execute("select ?", (austria,)).fetchone() |
| 177 | d_row = self.con.execute("select ?", (germany,)).fetchone() |
Serhiy Storchaka | 78ee078 | 2013-11-17 00:39:12 +0200 | [diff] [blame^] | 178 | self.assertEqual(type(a_row[0]), str, "type of non-ASCII row must be str") |
| 179 | self.assertEqual(type(d_row[0]), str, "type of ASCII-only row must be str") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 180 | |
| 181 | def tearDown(self): |
| 182 | self.con.close() |
| 183 | |
Petri Lehtinen | 023fe33 | 2012-02-01 22:18:19 +0200 | [diff] [blame] | 184 | class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase): |
| 185 | def setUp(self): |
| 186 | self.con = sqlite.connect(":memory:") |
| 187 | self.con.execute("create table test (value text)") |
| 188 | self.con.execute("insert into test (value) values (?)", ("a\x00b",)) |
| 189 | |
| 190 | def CheckString(self): |
| 191 | # text_factory defaults to str |
| 192 | row = self.con.execute("select value from test").fetchone() |
| 193 | self.assertIs(type(row[0]), str) |
| 194 | self.assertEqual(row[0], "a\x00b") |
| 195 | |
| 196 | def CheckBytes(self): |
| 197 | self.con.text_factory = bytes |
| 198 | row = self.con.execute("select value from test").fetchone() |
| 199 | self.assertIs(type(row[0]), bytes) |
| 200 | self.assertEqual(row[0], b"a\x00b") |
| 201 | |
| 202 | def CheckBytearray(self): |
| 203 | self.con.text_factory = bytearray |
| 204 | row = self.con.execute("select value from test").fetchone() |
| 205 | self.assertIs(type(row[0]), bytearray) |
| 206 | self.assertEqual(row[0], b"a\x00b") |
| 207 | |
| 208 | def CheckCustom(self): |
| 209 | # A custom factory should receive a bytes argument |
| 210 | self.con.text_factory = lambda x: x |
| 211 | row = self.con.execute("select value from test").fetchone() |
| 212 | self.assertIs(type(row[0]), bytes) |
| 213 | self.assertEqual(row[0], b"a\x00b") |
| 214 | |
| 215 | def tearDown(self): |
| 216 | self.con.close() |
| 217 | |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 218 | def suite(): |
| 219 | connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check") |
| 220 | cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check") |
| 221 | row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check") |
| 222 | row_suite = unittest.makeSuite(RowFactoryTests, "Check") |
| 223 | text_suite = unittest.makeSuite(TextFactoryTests, "Check") |
Petri Lehtinen | 023fe33 | 2012-02-01 22:18:19 +0200 | [diff] [blame] | 224 | text_zero_bytes_suite = unittest.makeSuite(TextFactoryTestsWithEmbeddedZeroBytes, "Check") |
| 225 | return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite, text_zero_bytes_suite)) |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 226 | |
| 227 | def test(): |
| 228 | runner = unittest.TextTestRunner() |
| 229 | runner.run(suite()) |
| 230 | |
| 231 | if __name__ == "__main__": |
| 232 | test() |