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 | # |
| 4 | # Copyright (C) 2005 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 | |
| 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 | |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 134 | def tearDown(self): |
| 135 | self.con.close() |
| 136 | |
| 137 | class TextFactoryTests(unittest.TestCase): |
| 138 | def setUp(self): |
| 139 | self.con = sqlite.connect(":memory:") |
| 140 | |
| 141 | def CheckUnicode(self): |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame^] | 142 | austria = "Österreich" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 143 | row = self.con.execute("select ?", (austria,)).fetchone() |
Guido van Rossum | ef87d6e | 2007-05-02 19:09:54 +0000 | [diff] [blame] | 144 | 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] | 145 | |
| 146 | def CheckString(self): |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame^] | 147 | self.con.text_factory = bytes |
| 148 | austria = "Österreich" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 149 | row = self.con.execute("select ?", (austria,)).fetchone() |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame^] | 150 | 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] | 151 | self.failUnless(row[0] == austria.encode("utf-8"), "column must equal original data in UTF-8") |
| 152 | |
| 153 | def CheckCustom(self): |
Guido van Rossum | ef87d6e | 2007-05-02 19:09:54 +0000 | [diff] [blame] | 154 | self.con.text_factory = lambda x: str(x, "utf-8", "ignore") |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame^] | 155 | austria = "Österreich" |
| 156 | row = self.con.execute("select ?", (austria,)).fetchone() |
Guido van Rossum | ef87d6e | 2007-05-02 19:09:54 +0000 | [diff] [blame] | 157 | self.failUnless(type(row[0]) == str, "type of row[0] must be unicode") |
| 158 | self.failUnless(row[0].endswith("reich"), "column must contain original data") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 159 | |
| 160 | def CheckOptimizedUnicode(self): |
| 161 | self.con.text_factory = sqlite.OptimizedUnicode |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame^] | 162 | austria = "Österreich" |
| 163 | germany = "Deutchland" |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 164 | a_row = self.con.execute("select ?", (austria,)).fetchone() |
| 165 | d_row = self.con.execute("select ?", (germany,)).fetchone() |
Guido van Rossum | ef87d6e | 2007-05-02 19:09:54 +0000 | [diff] [blame] | 166 | self.failUnless(type(a_row[0]) == str, "type of non-ASCII row must be unicode") |
Gerhard Häring | 6d21456 | 2007-08-10 18:15:11 +0000 | [diff] [blame^] | 167 | self.failUnless(type(d_row[0]) == str8, "type of ASCII-only row must be str8") |
Thomas Wouters | 49fd7fa | 2006-04-21 10:40:58 +0000 | [diff] [blame] | 168 | |
| 169 | def tearDown(self): |
| 170 | self.con.close() |
| 171 | |
| 172 | def suite(): |
| 173 | connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check") |
| 174 | cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check") |
| 175 | row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check") |
| 176 | row_suite = unittest.makeSuite(RowFactoryTests, "Check") |
| 177 | text_suite = unittest.makeSuite(TextFactoryTests, "Check") |
| 178 | return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite)) |
| 179 | |
| 180 | def test(): |
| 181 | runner = unittest.TextTestRunner() |
| 182 | runner.run(suite()) |
| 183 | |
| 184 | if __name__ == "__main__": |
| 185 | test() |