EVOLUTION-NINJA

Edit File: factory.py

#-*- coding: iso-8859-1 -*- # pysqlite2/test/factory.py: tests for the various factories in pysqlite # # Copyright (C) 2005-2007 Gerhard H�ring <gh@ghaering.de> # # This file is part of pysqlite. # # This software is provided 'as-is', without any express or implied # warranty. In no event will the authors be held liable for any damages # arising from the use of this software. # # Permission is granted to anyone to use this software for any purpose, # including commercial applications, and to alter it and redistribute it # freely, subject to the following restrictions: # # 1. The origin of this software must not be misrepresented; you must not # claim that you wrote the original software. If you use this software # in a product, an acknowledgment in the product documentation would be # appreciated but is not required. # 2. Altered source versions must be plainly marked as such, and must not be # misrepresented as being the original software. # 3. This notice may not be removed or altered from any source distribution. import unittest import sqlite3 as sqlite from collections.abc import Sequence class MyConnection(sqlite.Connection): def __init__(self, *args, **kwargs): sqlite.Connection.__init__(self, *args, **kwargs) def dict_factory(cursor, row): d = {} for idx, col in enumerate(cursor.description): d[col[0]] = row[idx] return d class MyCursor(sqlite.Cursor): def __init__(self, *args, **kwargs): sqlite.Cursor.__init__(self, *args, **kwargs) self.row_factory = dict_factory class ConnectionFactoryTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:", factory=MyConnection) def tearDown(self): self.con.close() def CheckIsInstance(self): self.assertIsInstance(self.con, MyConnection) class CursorFactoryTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") def tearDown(self): self.con.close() def CheckIsInstance(self): cur = self.con.cursor() self.assertIsInstance(cur, sqlite.Cursor) cur = self.con.cursor(MyCursor) self.assertIsInstance(cur, MyCursor) cur = self.con.cursor(factory=lambda con: MyCursor(con)) self.assertIsInstance(cur, MyCursor) def CheckInvalidFactory(self): # not a callable at all self.assertRaises(TypeError, self.con.cursor, None) # invalid callable with not exact one argument self.assertRaises(TypeError, self.con.cursor, lambda: None) # invalid callable returning non-cursor self.assertRaises(TypeError, self.con.cursor, lambda con: None) class RowFactoryTestsBackwardsCompat(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") def CheckIsProducedByFactory(self): cur = self.con.cursor(factory=MyCursor) cur.execute("select 4+5 as foo") row = cur.fetchone() self.assertIsInstance(row, dict) cur.close() def tearDown(self): self.con.close() class RowFactoryTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") def CheckCustomFactory(self): self.con.row_factory = lambda cur, row: list(row) row = self.con.execute("select 1, 2").fetchone() self.assertIsInstance(row, list) def CheckSqliteRowIndex(self): self.con.row_factory = sqlite.Row row = self.con.execute("select 1 as a_1, 2 as b").fetchone() self.assertIsInstance(row, sqlite.Row) self.assertEqual(row["a_1"], 1, "by name: wrong result for column 'a_1'") self.assertEqual(row["b"], 2, "by name: wrong result for column 'b'") self.assertEqual(row["A_1"], 1, "by name: wrong result for column 'A_1'") self.assertEqual(row["B"], 2, "by name: wrong result for column 'B'") self.assertEqual(row[0], 1, "by index: wrong result for column 0") self.assertEqual(row[1], 2, "by index: wrong result for column 1") self.assertEqual(row[-1], 2, "by index: wrong result for column -1") self.assertEqual(row[-2], 1, "by index: wrong result for column -2") with self.assertRaises(IndexError): row['c'] with self.assertRaises(IndexError): row['a_\x11'] with self.assertRaises(IndexError): row['a\x7f1'] with self.assertRaises(IndexError): row[2] with self.assertRaises(IndexError): row[-3] with self.assertRaises(IndexError): row[2**1000] def CheckSqliteRowIndexUnicode(self): self.con.row_factory = sqlite.Row row = self.con.execute("select 1 as \xff").fetchone() self.assertEqual(row["\xff"], 1) with self.assertRaises(IndexError): row['\u0178'] with self.assertRaises(IndexError): row['\xdf'] def CheckSqliteRowSlice(self): # A sqlite.Row can be sliced like a list. self.con.row_factory = sqlite.Row row = self.con.execute("select 1, 2, 3, 4").fetchone() self.assertEqual(row[0:0], ()) self.assertEqual(row[0:1], (1,)) self.assertEqual(row[1:3], (2, 3)) self.assertEqual(row[3:1], ()) # Explicit bounds are optional. self.assertEqual(row[1:], (2, 3, 4)) self.assertEqual(row[:3], (1, 2, 3)) # Slices can use negative indices. self.assertEqual(row[-2:-1], (3,)) self.assertEqual(row[-2:], (3, 4)) # Slicing supports steps. self.assertEqual(row[0:4:2], (1, 3)) self.assertEqual(row[3:0:-2], (4, 2)) def CheckSqliteRowIter(self): """Checks if the row object is iterable""" self.con.row_factory = sqlite.Row row = self.con.execute("select 1 as a, 2 as b").fetchone() for col in row: pass def CheckSqliteRowAsTuple(self): """Checks if the row object can be converted to a tuple""" self.con.row_factory = sqlite.Row row = self.con.execute("select 1 as a, 2 as b").fetchone() t = tuple(row) self.assertEqual(t, (row['a'], row['b'])) def CheckSqliteRowAsDict(self): """Checks if the row object can be correctly converted to a dictionary""" self.con.row_factory = sqlite.Row row = self.con.execute("select 1 as a, 2 as b").fetchone() d = dict(row) self.assertEqual(d["a"], row["a"]) self.assertEqual(d["b"], row["b"]) def CheckSqliteRowHashCmp(self): """Checks if the row object compares and hashes correctly""" self.con.row_factory = sqlite.Row row_1 = self.con.execute("select 1 as a, 2 as b").fetchone() row_2 = self.con.execute("select 1 as a, 2 as b").fetchone() row_3 = self.con.execute("select 1 as a, 3 as b").fetchone() row_4 = self.con.execute("select 1 as b, 2 as a").fetchone() row_5 = self.con.execute("select 2 as b, 1 as a").fetchone() self.assertTrue(row_1 == row_1) self.assertTrue(row_1 == row_2) self.assertFalse(row_1 == row_3) self.assertFalse(row_1 == row_4) self.assertFalse(row_1 == row_5) self.assertFalse(row_1 == object()) self.assertFalse(row_1 != row_1) self.assertFalse(row_1 != row_2) self.assertTrue(row_1 != row_3) self.assertTrue(row_1 != row_4) self.assertTrue(row_1 != row_5) self.assertTrue(row_1 != object()) with self.assertRaises(TypeError): row_1 > row_2 with self.assertRaises(TypeError): row_1 < row_2 with self.assertRaises(TypeError): row_1 >= row_2 with self.assertRaises(TypeError): row_1 <= row_2 self.assertEqual(hash(row_1), hash(row_2)) def CheckSqliteRowAsSequence(self): """ Checks if the row object can act like a sequence """ self.con.row_factory = sqlite.Row row = self.con.execute("select 1 as a, 2 as b").fetchone() as_tuple = tuple(row) self.assertEqual(list(reversed(row)), list(reversed(as_tuple))) self.assertIsInstance(row, Sequence) def CheckFakeCursorClass(self): # Issue #24257: Incorrect use of PyObject_IsInstance() caused # segmentation fault. # Issue #27861: Also applies for cursor factory. class FakeCursor(str): __class__ = sqlite.Cursor self.con.row_factory = sqlite.Row self.assertRaises(TypeError, self.con.cursor, FakeCursor) self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ()) def tearDown(self): self.con.close() class TextFactoryTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") def CheckUnicode(self): austria = "�sterreich" row = self.con.execute("select ?", (austria,)).fetchone() self.assertEqual(type(row[0]), str, "type of row[0] must be unicode") def CheckString(self): self.con.text_factory = bytes austria = "�sterreich" row = self.con.execute("select ?", (austria,)).fetchone() self.assertEqual(type(row[0]), bytes, "type of row[0] must be bytes") self.assertEqual(row[0], austria.encode("utf-8"), "column must equal original data in UTF-8") def CheckCustom(self): self.con.text_factory = lambda x: str(x, "utf-8", "ignore") austria = "�sterreich" row = self.con.execute("select ?", (austria,)).fetchone() self.assertEqual(type(row[0]), str, "type of row[0] must be unicode") self.assertTrue(row[0].endswith("reich"), "column must contain original data") def CheckOptimizedUnicode(self): # In py3k, str objects are always returned when text_factory # is OptimizedUnicode self.con.text_factory = sqlite.OptimizedUnicode austria = "�sterreich" germany = "Deutchland" a_row = self.con.execute("select ?", (austria,)).fetchone() d_row = self.con.execute("select ?", (germany,)).fetchone() self.assertEqual(type(a_row[0]), str, "type of non-ASCII row must be str") self.assertEqual(type(d_row[0]), str, "type of ASCII-only row must be str") def tearDown(self): self.con.close() class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") self.con.execute("create table test (value text)") self.con.execute("insert into test (value) values (?)", ("a\x00b",)) def CheckString(self): # text_factory defaults to str row = self.con.execute("select value from test").fetchone() self.assertIs(type(row[0]), str) self.assertEqual(row[0], "a\x00b") def CheckBytes(self): self.con.text_factory = bytes row = self.con.execute("select value from test").fetchone() self.assertIs(type(row[0]), bytes) self.assertEqual(row[0], b"a\x00b") def CheckBytearray(self): self.con.text_factory = bytearray row = self.con.execute("select value from test").fetchone() self.assertIs(type(row[0]), bytearray) self.assertEqual(row[0], b"a\x00b") def CheckCustom(self): # A custom factory should receive a bytes argument self.con.text_factory = lambda x: x row = self.con.execute("select value from test").fetchone() self.assertIs(type(row[0]), bytes) self.assertEqual(row[0], b"a\x00b") def tearDown(self): self.con.close() def suite(): connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check") cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check") row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check") row_suite = unittest.makeSuite(RowFactoryTests, "Check") text_suite = unittest.makeSuite(TextFactoryTests, "Check") text_zero_bytes_suite = unittest.makeSuite(TextFactoryTestsWithEmbeddedZeroBytes, "Check") return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite, text_zero_bytes_suite)) def test(): runner = unittest.TextTestRunner() runner.run(suite()) if __name__ == "__main__": test()