Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 1 | #-*- coding: ISO-8859-1 -*- |
| 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): |
| 50 | self.failUnless(isinstance(self.con, |
| 51 | MyConnection), |
| 52 | "connection is not instance of MyConnection") |
| 53 | |
| 54 | class 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.failUnless(isinstance(cur, |
| 64 | MyCursor), |
| 65 | "cursor is not instance of MyCursor") |
| 66 | |
| 67 | class 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.failUnless(isinstance(row, |
| 76 | dict), |
| 77 | "row is not instance of dict") |
| 78 | cur.close() |
| 79 | |
| 80 | def tearDown(self): |
| 81 | self.con.close() |
| 82 | |
| 83 | class 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.failUnless(isinstance(row, |
| 91 | list), |
| 92 | "row is not instance of list") |
| 93 | |
Thomas Wouters | fc7bb8c | 2007-01-15 15:49:28 +0000 | [diff] [blame] | 94 | def CheckSqliteRowIndex(self): |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 95 | self.con.row_factory = sqlite.Row |
| 96 | row = self.con.execute("select 1 as a, 2 as b").fetchone() |
| 97 | self.failUnless(isinstance(row, |
| 98 | sqlite.Row), |
| 99 | "row is not instance of sqlite.Row") |
| 100 | |
| 101 | col1, col2 = row["a"], row["b"] |
| 102 | self.failUnless(col1 == 1, "by name: wrong result for column 'a'") |
| 103 | self.failUnless(col2 == 2, "by name: wrong result for column 'a'") |
| 104 | |
| 105 | col1, col2 = row["A"], row["B"] |
| 106 | self.failUnless(col1 == 1, "by name: wrong result for column 'A'") |
| 107 | self.failUnless(col2 == 2, "by name: wrong result for column 'B'") |
| 108 | |
| 109 | col1, col2 = row[0], row[1] |
| 110 | self.failUnless(col1 == 1, "by index: wrong result for column 0") |
| 111 | self.failUnless(col2 == 2, "by index: wrong result for column 1") |
| 112 | |
Thomas Wouters | fc7bb8c | 2007-01-15 15:49:28 +0000 | [diff] [blame] | 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.failUnlessEqual(d["a"], row["a"]) |
| 132 | self.failUnlessEqual(d["b"], row["b"]) |
| 133 | |
Christian Heimes | 81ee3ef | 2008-05-04 22:42:01 +0000 | [diff] [blame^] | 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.failUnless(row_1 == row_1) |
| 142 | self.failUnless(row_1 == row_2) |
| 143 | self.failUnless(row_2 != row_3) |
| 144 | |
| 145 | self.failIf(row_1 != row_1) |
| 146 | self.failIf(row_1 != row_2) |
| 147 | self.failIf(row_2 == row_3) |
| 148 | |
| 149 | self.failUnlessEqual(row_1, row_2) |
| 150 | self.failUnlessEqual(hash(row_1), hash(row_2)) |
| 151 | self.failIfEqual(row_1, row_3) |
| 152 | self.failIfEqual(hash(row_1), hash(row_3)) |
| 153 | |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 154 | def tearDown(self): |
| 155 | self.con.close() |
| 156 | |
| 157 | class TextFactoryTests(unittest.TestCase): |
| 158 | def setUp(self): |
| 159 | self.con = sqlite.connect(":memory:") |
| 160 | |
| 161 | def CheckUnicode(self): |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 162 | austria = "Österreich" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 163 | row = self.con.execute("select ?", (austria,)).fetchone() |
Guido van Rossum | ef87d6e | 2007-05-02 19:09:54 +0000 | [diff] [blame] | 164 | self.failUnless(type(row[0]) == str, "type of row[0] must be unicode") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 165 | |
| 166 | def CheckString(self): |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 167 | self.con.text_factory = bytes |
| 168 | austria = "Österreich" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 169 | row = self.con.execute("select ?", (austria,)).fetchone() |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 170 | self.failUnless(type(row[0]) == bytes, "type of row[0] must be bytes") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 171 | self.failUnless(row[0] == austria.encode("utf-8"), "column must equal original data in UTF-8") |
| 172 | |
| 173 | def CheckCustom(self): |
Guido van Rossum | ef87d6e | 2007-05-02 19:09:54 +0000 | [diff] [blame] | 174 | self.con.text_factory = lambda x: str(x, "utf-8", "ignore") |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 175 | austria = "Österreich" |
| 176 | row = self.con.execute("select ?", (austria,)).fetchone() |
Guido van Rossum | ef87d6e | 2007-05-02 19:09:54 +0000 | [diff] [blame] | 177 | self.failUnless(type(row[0]) == str, "type of row[0] must be unicode") |
| 178 | self.failUnless(row[0].endswith("reich"), "column must contain original data") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 179 | |
| 180 | def CheckOptimizedUnicode(self): |
| 181 | self.con.text_factory = sqlite.OptimizedUnicode |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame] | 182 | austria = "Österreich" |
| 183 | germany = "Deutchland" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 184 | a_row = self.con.execute("select ?", (austria,)).fetchone() |
| 185 | d_row = self.con.execute("select ?", (germany,)).fetchone() |
Guido van Rossum | 98297ee | 2007-11-06 21:34:58 +0000 | [diff] [blame] | 186 | self.failUnless(type(a_row[0]) == str, "type of non-ASCII row must be str") |
| 187 | self.failUnless(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] | 188 | |
| 189 | def tearDown(self): |
| 190 | self.con.close() |
| 191 | |
| 192 | def suite(): |
| 193 | connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check") |
| 194 | cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check") |
| 195 | row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check") |
| 196 | row_suite = unittest.makeSuite(RowFactoryTests, "Check") |
| 197 | text_suite = unittest.makeSuite(TextFactoryTests, "Check") |
| 198 | return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite)) |
| 199 | |
| 200 | def test(): |
| 201 | runner = unittest.TextTestRunner() |
| 202 | runner.run(suite()) |
| 203 | |
| 204 | if __name__ == "__main__": |
| 205 | test() |