diff -r 03ff92b26fa9 Lib/sqlite3/test/dbapi.py --- a/Lib/sqlite3/test/dbapi.py Mon Apr 13 17:48:40 2015 -0400 +++ b/Lib/sqlite3/test/dbapi.py Wed Apr 15 17:05:08 2015 -0400 @@ -511,6 +511,152 @@ except TypeError: pass +class SqliteOnConflictTests(unittest.TestCase): + + def setUp(self): + self.cx = sqlite.connect(":memory:") + self.cu = self.cx.cursor() + self.cu.execute("create table test(id integer primary key, name text, unique_name text unique)") + + def tearDown(self): + self.cu.close() + self.cx.close() + + def CheckOnConflictRollbackWithExplicitTransaction(self): + """ + Info at https://www.sqlite.org/lang_conflict.html + """ + # Start transaction. + self.cx.isolation_level = None + self.cu = self.cx.cursor() + self.cu.execute('begin') + self.cu.execute("INSERT INTO test(name) values ('abort_test')") + self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) values ('foo')") + # Cause abort + try: + # Shouldn't insert anything. + self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) values ('foo')") + except sqlite.IntegrityError: + pass + # Use connection to commit. + self.cx.commit() + self.cu.execute("SELECT name, unique_name from test") + # Transaction should have rolled back and nothing should be in table. + returned_rows = [] + for row in self.cu: + returned_rows.append(row) + expected_list = [] + self.assertEqual(returned_rows, expected_list) + + def CheckOnConflictAbortRaisesWithExplicitTransactions(self): + """ + Info at https://www.sqlite.org/lang_conflict.html + Abort cancels the current sql statement but doesn't change anything + about the current transaction. + """ + self.cx.isolation_level = None + self.cu = self.cx.cursor() + # Start transaction. + self.cu.execute('begin') + self.cu.execute("INSERT INTO test(name) values ('abort_test')") + self.cu.execute("INSERT OR ABORT INTO test(unique_name) values ('foo')") + # Cause abort + try: + # Shouldn't insert anything. + self.cu.execute("INSERT OR ABORT INTO test(unique_name) values ('foo')") + except sqlite.IntegrityError: + pass + # Use connection to commit. + self.cx.commit() + self.cu.execute("SELECT name, unique_name from test") + returned_rows = [] + for row in self.cu: + returned_rows.append(row) + # Expect the first two inserts to work, third to do nothing. + expected_list = [('abort_test', None), (None, 'foo',)] + self.assertEqual(returned_rows, expected_list) + + def CheckOnConflictRollbackWithoutTransaction(self): + """ + Info at https://www.sqlite.org/lang_conflict.html + """ + # Start of implicit transaction + self.cu.execute("INSERT INTO test(name) values ('abort_test')") + self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) values ('foo')") + # Cause abort + try: + # Shouldn't insert anything. + self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) values ('foo')") + except sqlite.IntegrityError: + pass + # Make sure all other values were inserted. + self.cu.execute("SELECT name, unique_name from test") + # No wrapping transaction, rollback only on statement. + returned_rows = [] + for row in self.cu: + returned_rows.append(row) + # Implicit transaction is rolled back on Error. + expected_list = [] + self.assertEqual(returned_rows, expected_list) + + def CheckOnConflictAbortRaisesWithoutTransactions(self): + """ + Info at https://www.sqlite.org/lang_conflict.html + Abort cancels the current sql statement but doesn't change anything + about the current transaction. + """ + self.cu.execute("INSERT INTO test(name) values ('abort_test')") + self.cu.execute("INSERT OR ABORT INTO test(unique_name) values ('foo')") + # Cause abort + try: + # Shouldn't insert anything. + self.cu.execute("INSERT OR ABORT INTO test(unique_name) values ('foo')") + except sqlite.IntegrityError: + pass + # Make sure all other values were inserted. + self.cu.execute("SELECT name, unique_name from test") + returned_rows = [] + for row in self.cu: + returned_rows.append(row) + expected_list = [('abort_test', None), (None, 'foo',)] + self.assertEqual(returned_rows, expected_list) + + def CheckOnConflictFail(self): + """ + Info at https://www.sqlite.org/lang_conflict.html + """ + with self.assertRaises(sqlite.IntegrityError): + self.cu.execute("INSERT OR FAIL INTO test(unique_name) values ('foo')") + self.cu.execute("INSERT OR FAIL INTO test(unique_name) values ('foo')") + + def CheckOnConflictIgnore(self): + """ + Info at https://www.sqlite.org/lang_conflict.html + """ + self.cu.execute("INSERT OR IGNORE INTO test(unique_name) values ('foo')") + # Nothing should happen. + self.cu.execute("INSERT OR IGNORE INTO test(unique_name) values ('foo')") + self.cu.execute("SELECT unique_name from test") + returned_rows = [] + for row in self.cu: + returned_rows.append(row) + expected_list = [('foo',)] + self.assertEqual(returned_rows, expected_list) + + def CheckOnConflictReplace(self): + """ + Info at https://www.sqlite.org/lang_conflict.html + """ + self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) values ('Data!', 'foo')") + self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) values ('Very different data!', 'foo')") + # There shouldn't be an exception. + self.cu.execute("SELECT name, unique_name from test") + returned_rows = [] + for row in self.cu: + returned_rows.append(row) + expected_list = [('Very different data!', 'foo')] + self.assertEqual(returned_rows, expected_list) + @unittest.skipUnless(threading, 'This test requires threading.') class ThreadTests(unittest.TestCase): def setUp(self): @@ -901,7 +1047,8 @@ ext_suite = unittest.makeSuite(ExtensionTests, "Check") closed_con_suite = unittest.makeSuite(ClosedConTests, "Check") closed_cur_suite = unittest.makeSuite(ClosedCurTests, "Check") - return unittest.TestSuite((module_suite, connection_suite, cursor_suite, thread_suite, constructor_suite, ext_suite, closed_con_suite, closed_cur_suite)) + sqlite_conflict_suit = unittest.makeSuite(SqliteOnConflictTests, "Check") + return unittest.TestSuite((module_suite, connection_suite, cursor_suite, thread_suite, constructor_suite, ext_suite, closed_con_suite, closed_cur_suite, sqlite_conflict_suit)) def test(): runner = unittest.TextTestRunner()