/usr/lib/python3.7/sqlite3/test/backup.py is in libpython3.7-testsuite 3.7.0~b3-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | import sqlite3 as sqlite
import unittest
@unittest.skipIf(sqlite.sqlite_version_info < (3, 6, 11), "Backup API not supported")
class BackupTests(unittest.TestCase):
def setUp(self):
cx = self.cx = sqlite.connect(":memory:")
cx.execute('CREATE TABLE foo (key INTEGER)')
cx.executemany('INSERT INTO foo (key) VALUES (?)', [(3,), (4,)])
cx.commit()
def tearDown(self):
self.cx.close()
def verify_backup(self, bckcx):
result = bckcx.execute("SELECT key FROM foo ORDER BY key").fetchall()
self.assertEqual(result[0][0], 3)
self.assertEqual(result[1][0], 4)
def test_bad_target_none(self):
with self.assertRaises(TypeError):
self.cx.backup(None)
def test_bad_target_filename(self):
with self.assertRaises(TypeError):
self.cx.backup('some_file_name.db')
def test_bad_target_same_connection(self):
with self.assertRaises(ValueError):
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)
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)
if sqlite.sqlite_version_info < (3, 8, 8):
self.assertEqual(str(cm.exception), 'target is in transaction')
def test_keyword_only_args(self):
with self.assertRaises(TypeError):
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, 1)
def test_simple(self):
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck)
self.verify_backup(bck)
def test_progress(self):
journal = []
def progress(status, remaining, total):
journal.append(status)
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, pages=1, progress=progress)
self.verify_backup(bck)
self.assertEqual(len(journal), 2)
self.assertEqual(journal[0], sqlite.SQLITE_OK)
self.assertEqual(journal[1], sqlite.SQLITE_DONE)
def test_progress_all_pages_at_once_1(self):
journal = []
def progress(status, remaining, total):
journal.append(remaining)
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, progress=progress)
self.verify_backup(bck)
self.assertEqual(len(journal), 1)
self.assertEqual(journal[0], 0)
def test_progress_all_pages_at_once_2(self):
journal = []
def progress(status, remaining, total):
journal.append(remaining)
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, pages=-1, progress=progress)
self.verify_backup(bck)
self.assertEqual(len(journal), 1)
self.assertEqual(journal[0], 0)
def test_non_callable_progress(self):
with self.assertRaises(TypeError) as cm:
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, pages=1, progress='bar')
self.assertEqual(str(cm.exception), 'progress argument must be a callable')
def test_modifying_progress(self):
journal = []
def progress(status, remaining, total):
if not journal:
self.cx.execute('INSERT INTO foo (key) VALUES (?)', (remaining+1000,))
self.cx.commit()
journal.append(remaining)
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, pages=1, progress=progress)
self.verify_backup(bck)
result = bck.execute("SELECT key FROM foo"
" WHERE key >= 1000"
" ORDER BY key").fetchall()
self.assertEqual(result[0][0], 1001)
self.assertEqual(len(journal), 3)
self.assertEqual(journal[0], 1)
self.assertEqual(journal[1], 1)
self.assertEqual(journal[2], 0)
def test_failing_progress(self):
def progress(status, remaining, total):
raise SystemError('nearly out of space')
with self.assertRaises(SystemError) as err:
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, progress=progress)
self.assertEqual(str(err.exception), 'nearly out of space')
def test_database_source_name(self):
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, name='main')
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, name='temp')
with self.assertRaises(sqlite.OperationalError) as cm:
with sqlite.connect(':memory:') as bck:
self.cx.backup(bck, name='non-existing')
self.assertIn(
str(cm.exception),
['SQL logic error', 'SQL logic error or missing database']
)
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:
self.cx.backup(bck, name='attached_db')
self.verify_backup(bck)
def suite():
return unittest.makeSuite(BackupTests)
if __name__ == "__main__":
unittest.main()
|