diff options
Diffstat (limited to 'Lib/sqlite3')
| -rw-r--r-- | Lib/sqlite3/test/dbapi.py | 43 | ||||
| -rw-r--r-- | Lib/sqlite3/test/transactions.py | 36 |
2 files changed, 70 insertions, 9 deletions
diff --git a/Lib/sqlite3/test/dbapi.py b/Lib/sqlite3/test/dbapi.py index 7fb8d7e..cb85814 100644 --- a/Lib/sqlite3/test/dbapi.py +++ b/Lib/sqlite3/test/dbapi.py @@ -188,7 +188,10 @@ class CursorTests(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, income number)") + self.cu.execute( + "create table test(id integer primary key, name text, " + "income number, unique_test text unique)" + ) self.cu.execute("insert into test(name) values (?)", ("foo",)) def tearDown(self): @@ -462,6 +465,44 @@ class CursorTests(unittest.TestCase): with self.assertRaises(TypeError): cur = sqlite.Cursor(foo) + def CheckLastRowIDOnReplace(self): + """ + INSERT OR REPLACE and REPLACE INTO should produce the same behavior. + """ + sql = '{} INTO test(id, unique_test) VALUES (?, ?)' + for statement in ('INSERT OR REPLACE', 'REPLACE'): + with self.subTest(statement=statement): + self.cu.execute(sql.format(statement), (1, 'foo')) + self.assertEqual(self.cu.lastrowid, 1) + + def CheckLastRowIDOnIgnore(self): + self.cu.execute( + "insert or ignore into test(unique_test) values (?)", + ('test',)) + self.assertEqual(self.cu.lastrowid, 2) + self.cu.execute( + "insert or ignore into test(unique_test) values (?)", + ('test',)) + self.assertEqual(self.cu.lastrowid, 2) + + def CheckLastRowIDInsertOR(self): + results = [] + for statement in ('FAIL', 'ABORT', 'ROLLBACK'): + sql = 'INSERT OR {} INTO test(unique_test) VALUES (?)' + with self.subTest(statement='INSERT OR {}'.format(statement)): + self.cu.execute(sql.format(statement), (statement,)) + results.append((statement, self.cu.lastrowid)) + with self.assertRaises(sqlite.IntegrityError): + self.cu.execute(sql.format(statement), (statement,)) + results.append((statement, self.cu.lastrowid)) + expected = [ + ('FAIL', 2), ('FAIL', 2), + ('ABORT', 3), ('ABORT', 3), + ('ROLLBACK', 4), ('ROLLBACK', 4), + ] + self.assertEqual(results, expected) + + @unittest.skipUnless(threading, 'This test requires threading.') class ThreadTests(unittest.TestCase): def setUp(self): diff --git a/Lib/sqlite3/test/transactions.py b/Lib/sqlite3/test/transactions.py index a25360a..45f1b04 100644 --- a/Lib/sqlite3/test/transactions.py +++ b/Lib/sqlite3/test/transactions.py @@ -52,13 +52,13 @@ class TransactionTests(unittest.TestCase): except OSError: pass - def CheckDMLdoesAutoCommitBefore(self): + def CheckDMLDoesNotAutoCommitBefore(self): self.cur1.execute("create table test(i)") self.cur1.execute("insert into test(i) values (5)") self.cur1.execute("create table test2(j)") self.cur2.execute("select i from test") res = self.cur2.fetchall() - self.assertEqual(len(res), 1) + self.assertEqual(len(res), 0) def CheckInsertStartsTransaction(self): self.cur1.execute("create table test(i)") @@ -153,11 +153,6 @@ class SpecialCommandTests(unittest.TestCase): self.con = sqlite.connect(":memory:") self.cur = self.con.cursor() - def CheckVacuum(self): - self.cur.execute("create table test(i)") - self.cur.execute("insert into test(i) values (5)") - self.cur.execute("vacuum") - def CheckDropTable(self): self.cur.execute("create table test(i)") self.cur.execute("insert into test(i) values (5)") @@ -172,10 +167,35 @@ class SpecialCommandTests(unittest.TestCase): self.cur.close() self.con.close() +class TransactionalDDL(unittest.TestCase): + def setUp(self): + self.con = sqlite.connect(":memory:") + + def CheckDdlDoesNotAutostartTransaction(self): + # For backwards compatibility reasons, DDL statements should not + # implicitly start a transaction. + self.con.execute("create table test(i)") + self.con.rollback() + result = self.con.execute("select * from test").fetchall() + self.assertEqual(result, []) + + def CheckTransactionalDDL(self): + # You can achieve transactional DDL by issuing a BEGIN + # statement manually. + self.con.execute("begin") + self.con.execute("create table test(i)") + self.con.rollback() + with self.assertRaises(sqlite.OperationalError): + self.con.execute("select * from test") + + def tearDown(self): + self.con.close() + def suite(): default_suite = unittest.makeSuite(TransactionTests, "Check") special_command_suite = unittest.makeSuite(SpecialCommandTests, "Check") - return unittest.TestSuite((default_suite, special_command_suite)) + ddl_suite = unittest.makeSuite(TransactionalDDL, "Check") + return unittest.TestSuite((default_suite, special_command_suite, ddl_suite)) def test(): runner = unittest.TextTestRunner() |