|
22 | 22 |
|
23 | 23 | import os, unittest
|
24 | 24 | import sqlite3 as sqlite
|
| 25 | +from contextlib import contextmanager |
25 | 26 |
|
26 | 27 | from test.support import LOOPBACK_TIMEOUT
|
27 | 28 | from test.support.os_helper import TESTFN, unlink
|
@@ -327,5 +328,77 @@ def test_isolation_level_none(self):
|
327 | 328 | self.assertEqual(self.traced, [self.QUERY])
|
328 | 329 |
|
329 | 330 |
|
| 331 | +class AutocommitPEP249(unittest.TestCase): |
| 332 | + """Test PEP-249 compliant autocommit behaviour.""" |
| 333 | + @contextmanager |
| 334 | + def check_stmt_trace(self, cx, expected): |
| 335 | + try: |
| 336 | + traced = [] |
| 337 | + cx.set_trace_callback(lambda stmt: traced.append(stmt)) |
| 338 | + yield |
| 339 | + finally: |
| 340 | + self.assertEqual(traced, expected) |
| 341 | + cx.set_trace_callback(None) |
| 342 | + |
| 343 | + def test_autocommit_default(self): |
| 344 | + with memory_database() as cx: |
| 345 | + self.assertEqual(cx.autocommit, |
| 346 | + sqlite.DEPRECATED_TRANSACTION_CONTROL) |
| 347 | + |
| 348 | + def test_autocommit_setget(self): |
| 349 | + dataset = ( |
| 350 | + True, |
| 351 | + False, |
| 352 | + sqlite.DEPRECATED_TRANSACTION_CONTROL, |
| 353 | + ) |
| 354 | + for mode in dataset: |
| 355 | + with self.subTest(mode=mode): |
| 356 | + with memory_database(autocommit=mode) as cx: |
| 357 | + self.assertEqual(cx.autocommit, mode) |
| 358 | + with memory_database() as cx: |
| 359 | + cx.autocommit = mode |
| 360 | + self.assertEqual(cx.autocommit, mode) |
| 361 | + |
| 362 | + def test_autocommit_setget_invalid(self): |
| 363 | + msg = "autocommit must be True, False, or.*DEPRECATED" |
| 364 | + for mode in "a", 12, (), None: |
| 365 | + with self.subTest(mode=mode): |
| 366 | + with self.assertRaisesRegex(ValueError, msg): |
| 367 | + with memory_database(autocommit=mode) as cx: |
| 368 | + self.assertEqual(cx.autocommit, mode) |
| 369 | + |
| 370 | + def test_autocommit_disabled(self): |
| 371 | + expected = [ |
| 372 | + "SELECT 1", |
| 373 | + "COMMIT", |
| 374 | + "BEGIN", |
| 375 | + "ROLLBACK", |
| 376 | + "BEGIN", |
| 377 | + ] |
| 378 | + with memory_database(autocommit=False) as cx: |
| 379 | + self.assertTrue(cx.in_transaction) |
| 380 | + with self.check_stmt_trace(cx, expected): |
| 381 | + cx.execute("SELECT 1") |
| 382 | + cx.commit() |
| 383 | + cx.rollback() |
| 384 | + |
| 385 | + def test_autocommit_disabled_implicit_rollback(self): |
| 386 | + expected = ["ROLLBACK"] |
| 387 | + actual = [] |
| 388 | + with memory_database(autocommit=False) as cx: |
| 389 | + self.assertTrue(cx.in_transaction) |
| 390 | + cx.set_trace_callback(lambda stmt: actual.append(stmt)) |
| 391 | + cx.close() |
| 392 | + self.assertEqual(actual, expected) |
| 393 | + |
| 394 | + def test_autocommit_enabled(self): |
| 395 | + expected = ["SELECT 1"] |
| 396 | + with memory_database(autocommit=True) as cx: |
| 397 | + self.assertFalse(cx.in_transaction) |
| 398 | + with self.check_stmt_trace(cx, expected): |
| 399 | + cx.execute("SELECT 1") |
| 400 | + cx.commit() # expect this to pass silently |
| 401 | + |
| 402 | + |
330 | 403 | if __name__ == "__main__":
|
331 | 404 | unittest.main()
|
0 commit comments