Lib/sqlite3/test/factory.py

   1 #-*- coding: ISO-8859-1 -*-
   2 # pysqlite2/test/factory.py: tests for the various factories in pysqlite
   3 #
   4 # Copyright (C) 2005-2007 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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(isinstance(row,
  91                                    list),
  92                         "row is not instance of list")
  93
  94     def CheckSqliteRowIndex(self):
  95         self.con.row_factory = sqlite.Row
  96         row = self.con.execute("select 1 as a, 2 as b").fetchone()
  97         self.assertTrue(isinstance(row,
  98                                    sqlite.Row),
  99                         "row is not instance of sqlite.Row")
 100
 101         col1, col2 = row["a"], row["b"]
 102         self.assertTrue(col1 == 1, "by name: wrong result for column 'a'")
 103         self.assertTrue(col2 == 2, "by name: wrong result for column 'a'")
 104
 105         col1, col2 = row["A"], row["B"]
 106         self.assertTrue(col1 == 1, "by name: wrong result for column 'A'")
 107         self.assertTrue(col2 == 2, "by name: wrong result for column 'B'")
 108
 109         col1, col2 = row[0], row[1]
 110         self.assertTrue(col1 == 1, "by index: wrong result for column 0")
 111         self.assertTrue(col2 == 2, "by index: wrong result for column 1")
 112
 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.assertEqual(d["a"], row["a"])
 132         self.assertEqual(d["b"], row["b"])
 133
 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.assertTrue(row_1 == row_1)
 142         self.assertTrue(row_1 == row_2)
 143         self.assertTrue(row_2 != row_3)
 144
 145         self.assertFalse(row_1 != row_1)
 146         self.assertFalse(row_1 != row_2)
 147         self.assertFalse(row_2 == row_3)
 148
 149         self.assertEqual(row_1, row_2)
 150         self.assertEqual(hash(row_1), hash(row_2))
 151         self.assertNotEqual(row_1, row_3)
 152         self.assertNotEqual(hash(row_1), hash(row_3))
 153
 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):
 162         austria = unicode("Österreich", "latin1")
 163         row = self.con.execute("select ?", (austria,)).fetchone()
 164         self.assertTrue(type(row[0]) == unicode, "type of row[0] must be unicode")
 165
 166     def CheckString(self):
 167         self.con.text_factory = str
 168         austria = unicode("Österreich", "latin1")
 169         row = self.con.execute("select ?", (austria,)).fetchone()
 170         self.assertTrue(type(row[0]) == str, "type of row[0] must be str")
 171         self.assertTrue(row[0] == austria.encode("utf-8"), "column must equal original data in UTF-8")
 172
 173     def CheckCustom(self):
 174         self.con.text_factory = lambda x: unicode(x, "utf-8", "ignore")
 175         austria = unicode("Österreich", "latin1")
 176         row = self.con.execute("select ?", (austria.encode("latin1"),)).fetchone()
 177         self.assertTrue(type(row[0]) == unicode, "type of row[0] must be unicode")
 178         self.assertTrue(row[0].endswith(u"reich"), "column must contain original data")
 179
 180     def CheckOptimizedUnicode(self):
 181         self.con.text_factory = sqlite.OptimizedUnicode
 182         austria = unicode("Österreich", "latin1")
 183         germany = unicode("Deutchland")
 184         a_row = self.con.execute("select ?", (austria,)).fetchone()
 185         d_row = self.con.execute("select ?", (germany,)).fetchone()
 186         self.assertTrue(type(a_row[0]) == unicode, "type of non-ASCII row must be unicode")
 187         self.assertTrue(type(d_row[0]) == str, "type of ASCII-only row must be str")
 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()