• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import sqlite3 as sqlite
2import unittest
3
4
5class BackupTests(unittest.TestCase):
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
163def suite():
164    return unittest.TestLoader().loadTestsFromTestCase(BackupTests)
165
166if __name__ == "__main__":
167    unittest.main()
168