Skip to content

Commit

Permalink
gh-105539: Explict resource management for connection objects in sqli…
Browse files Browse the repository at this point in the history
…te3 tests (#108017)

- Use memory_database() helper
- Move test utility functions to util.py
- Add convenience memory database mixin
- Add check() helper for closed connection tests
  • Loading branch information
erlend-aasland committed Aug 17, 2023
1 parent c9d83f9 commit 1344cfa
Show file tree
Hide file tree
Showing 9 changed files with 373 additions and 387 deletions.
52 changes: 27 additions & 25 deletions Lib/test/test_sqlite3/test_backup.py
@@ -1,6 +1,8 @@
import sqlite3 as sqlite
import unittest

from .util import memory_database


class BackupTests(unittest.TestCase):
def setUp(self):
Expand Down Expand Up @@ -32,32 +34,32 @@ def test_bad_target_same_connection(self):
self.cx.backup(self.cx)

def test_bad_target_closed_connection(self):
bck = sqlite.connect(':memory:')
bck.close()
with self.assertRaises(sqlite.ProgrammingError):
self.cx.backup(bck)
with memory_database() as bck:
bck.close()
with self.assertRaises(sqlite.ProgrammingError):
self.cx.backup(bck)

def test_bad_source_closed_connection(self):
bck = sqlite.connect(':memory:')
source = sqlite.connect(":memory:")
source.close()
with self.assertRaises(sqlite.ProgrammingError):
source.backup(bck)
with memory_database() as bck:
source = sqlite.connect(":memory:")
source.close()
with self.assertRaises(sqlite.ProgrammingError):
source.backup(bck)

def test_bad_target_in_transaction(self):
bck = sqlite.connect(':memory:')
bck.execute('CREATE TABLE bar (key INTEGER)')
bck.executemany('INSERT INTO bar (key) VALUES (?)', [(3,), (4,)])
with self.assertRaises(sqlite.OperationalError) as cm:
self.cx.backup(bck)
with memory_database() as bck:
bck.execute('CREATE TABLE bar (key INTEGER)')
bck.executemany('INSERT INTO bar (key) VALUES (?)', [(3,), (4,)])
with self.assertRaises(sqlite.OperationalError) as cm:
self.cx.backup(bck)

def test_keyword_only_args(self):
with self.assertRaises(TypeError):
with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, 1)

def test_simple(self):
with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck)
self.verify_backup(bck)

Expand All @@ -67,7 +69,7 @@ def test_progress(self):
def progress(status, remaining, total):
journal.append(status)

with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, pages=1, progress=progress)
self.verify_backup(bck)

Expand All @@ -81,7 +83,7 @@ def test_progress_all_pages_at_once_1(self):
def progress(status, remaining, total):
journal.append(remaining)

with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, progress=progress)
self.verify_backup(bck)

Expand All @@ -94,7 +96,7 @@ def test_progress_all_pages_at_once_2(self):
def progress(status, remaining, total):
journal.append(remaining)

with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, pages=-1, progress=progress)
self.verify_backup(bck)

Expand All @@ -103,7 +105,7 @@ def progress(status, remaining, total):

def test_non_callable_progress(self):
with self.assertRaises(TypeError) as cm:
with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, pages=1, progress='bar')
self.assertEqual(str(cm.exception), 'progress argument must be a callable')

Expand All @@ -116,7 +118,7 @@ def progress(status, remaining, total):
self.cx.commit()
journal.append(remaining)

with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, pages=1, progress=progress)
self.verify_backup(bck)

Expand All @@ -140,20 +142,20 @@ def progress(status, remaining, total):
self.assertEqual(str(err.exception), 'nearly out of space')

def test_database_source_name(self):
with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, name='main')
with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, name='temp')
with self.assertRaises(sqlite.OperationalError) as cm:
with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, name='non-existing')
self.assertIn("unknown database", str(cm.exception))

self.cx.execute("ATTACH DATABASE ':memory:' AS attached_db")
self.cx.execute('CREATE TABLE attached_db.foo (key INTEGER)')
self.cx.executemany('INSERT INTO attached_db.foo (key) VALUES (?)', [(3,), (4,)])
self.cx.commit()
with sqlite.connect(':memory:') as bck:
with memory_database() as bck:
self.cx.backup(bck, name='attached_db')
self.verify_backup(bck)

Expand Down

0 comments on commit 1344cfa

Please sign in to comment.