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