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