3 from StringIO import StringIO
4 from twisted.trial import unittest
6 from allmydata.util import fileutil
7 from allmydata.util.stringutils import listdir_unicode, get_filesystem_encoding, unicode_platform
8 from allmydata.util.assertutil import precondition
9 from allmydata.scripts import backupdb
11 class BackupDB(unittest.TestCase):
12 def create_or_skip(self, dbfile):
14 bdb = backupdb.get_backupdb(dbfile, stderr=stderr)
16 if "I was unable to import a python sqlite library" in stderr.getvalue():
17 raise unittest.SkipTest("sqlite unavailable, skipping test")
20 def skip_if_cannot_represent_filename(self, u):
21 precondition(isinstance(u, unicode))
23 enc = get_filesystem_encoding()
24 if not unicode_platform():
27 except UnicodeEncodeError:
28 raise unittest.SkipTest("A non-ASCII filename could not be encoded on this platform.")
31 self.basedir = basedir = os.path.join("backupdb", "create")
32 fileutil.make_dirs(basedir)
33 dbfile = os.path.join(basedir, "dbfile")
34 bdb = self.create_or_skip(dbfile)
36 self.failUnlessEqual(bdb.VERSION, 2)
38 def test_upgrade_v1_v2(self):
39 self.basedir = basedir = os.path.join("backupdb", "upgrade_v1_v2")
40 fileutil.make_dirs(basedir)
41 dbfile = os.path.join(basedir, "dbfile")
43 created = backupdb.get_backupdb(dbfile, stderr=stderr,
44 create_version=(backupdb.SCHEMA_v1, 1),
47 if "I was unable to import a python sqlite library" in stderr.getvalue():
48 raise unittest.SkipTest("sqlite unavailable, skipping test")
49 self.fail("unable to create v1 backupdb")
50 # now we should have a v1 database on disk
51 bdb = self.create_or_skip(dbfile)
53 self.failUnlessEqual(bdb.VERSION, 2)
56 self.basedir = basedir = os.path.join("backupdb", "fail")
57 fileutil.make_dirs(basedir)
59 # put a non-DB file in the way
60 not_a_db = ("I do not look like a sqlite database\n" +
61 "I'M NOT" * 1000) # OS-X sqlite-2.3.2 takes some convincing
62 self.writeto("not-a-database", not_a_db)
64 bdb = backupdb.get_backupdb(os.path.join(basedir, "not-a-database"),
66 self.failUnlessEqual(bdb, None)
67 stderr = stderr_f.getvalue()
68 if "I was unable to import a python sqlite library" in stderr:
71 self.failUnless("backupdb file is unusable" in stderr, stderr)
72 self.failUnless("file is encrypted or is not a database" in stderr,
75 # put a directory in the way, to exercise a different error path
76 where = os.path.join(basedir, "roadblock-dir")
77 fileutil.make_dirs(where)
79 bdb = backupdb.get_backupdb(where, stderr_f)
80 self.failUnlessEqual(bdb, None)
81 stderr = stderr_f.getvalue()
82 if "I was unable to import a python sqlite library" in stderr:
85 self.failUnless(("Unable to create/open backupdb file %s" % where)
87 self.failUnless("unable to open database file" in stderr, stderr)
90 def writeto(self, filename, data):
91 fn = os.path.join(self.basedir, unicode(filename))
92 parentdir = os.path.dirname(fn)
93 fileutil.make_dirs(parentdir)
94 fileutil.write(fn, data)
98 self.basedir = basedir = os.path.join("backupdb", "check")
99 fileutil.make_dirs(basedir)
100 dbfile = os.path.join(basedir, "dbfile")
101 bdb = self.create_or_skip(dbfile)
104 foo_fn = self.writeto("foo.txt", "foo.txt")
105 blah_fn = self.writeto("bar/blah.txt", "blah.txt")
107 r = bdb.check_file(foo_fn)
108 self.failUnlessEqual(r.was_uploaded(), False)
109 r.did_upload("foo-cap")
111 r = bdb.check_file(blah_fn)
112 self.failUnlessEqual(r.was_uploaded(), False)
113 r.did_upload("blah-cap")
115 r = bdb.check_file(foo_fn)
116 self.failUnlessEqual(r.was_uploaded(), "foo-cap")
117 self.failUnlessEqual(type(r.was_uploaded()), str)
118 self.failUnlessEqual(r.should_check(), False)
120 time.sleep(1.0) # make sure the timestamp changes
121 self.writeto("foo.txt", "NEW")
123 r = bdb.check_file(foo_fn)
124 self.failUnlessEqual(r.was_uploaded(), False)
125 r.did_upload("new-cap")
127 r = bdb.check_file(foo_fn)
128 self.failUnlessEqual(r.was_uploaded(), "new-cap")
129 self.failUnlessEqual(r.should_check(), False)
130 # if we spontaneously decide to upload it anyways, nothing should
132 r.did_upload("new-cap")
134 r = bdb.check_file(foo_fn, use_timestamps=False)
135 self.failUnlessEqual(r.was_uploaded(), False)
136 r.did_upload("new-cap")
138 r = bdb.check_file(foo_fn)
139 self.failUnlessEqual(r.was_uploaded(), "new-cap")
140 self.failUnlessEqual(r.should_check(), False)
142 bdb.NO_CHECK_BEFORE = 0
143 bdb.ALWAYS_CHECK_AFTER = 0.1
145 r = bdb.check_file(blah_fn)
146 self.failUnlessEqual(r.was_uploaded(), "blah-cap")
147 self.failUnlessEqual(r.should_check(), True)
148 r.did_check_healthy("results") # we know they're ignored for now
150 bdb.NO_CHECK_BEFORE = 200
151 bdb.ALWAYS_CHECK_AFTER = 400
153 r = bdb.check_file(blah_fn)
154 self.failUnlessEqual(r.was_uploaded(), "blah-cap")
155 self.failUnlessEqual(r.should_check(), False)
157 os.unlink(os.path.join(basedir, "foo.txt"))
158 fileutil.make_dirs(os.path.join(basedir, "foo.txt")) # file becomes dir
159 r = bdb.check_file(foo_fn)
160 self.failUnlessEqual(r.was_uploaded(), False)
162 def test_wrong_version(self):
163 self.basedir = basedir = os.path.join("backupdb", "wrong_version")
164 fileutil.make_dirs(basedir)
166 where = os.path.join(basedir, "tooold.db")
167 bdb = self.create_or_skip(where)
168 # reach into the DB and make it old
169 bdb.cursor.execute("UPDATE version SET version=0")
170 bdb.connection.commit()
172 # now the next time we open the database, it should be an unusable
174 stderr_f = StringIO()
175 bdb = backupdb.get_backupdb(where, stderr_f)
176 self.failUnlessEqual(bdb, None)
177 stderr = stderr_f.getvalue()
178 self.failUnlessEqual(stderr.strip(),
179 "Unable to handle backupdb version 0")
181 def test_directory(self):
182 self.basedir = basedir = os.path.join("backupdb", "directory")
183 fileutil.make_dirs(basedir)
184 dbfile = os.path.join(basedir, "dbfile")
185 bdb = self.create_or_skip(dbfile)
188 contents = {u"file1": "URI:CHK:blah1",
189 u"file2": "URI:CHK:blah2",
190 u"dir1": "URI:DIR2-CHK:baz2"}
191 r = bdb.check_directory(contents)
192 self.failUnless(isinstance(r, backupdb.DirectoryResult))
193 self.failIf(r.was_created())
194 dircap = "URI:DIR2-CHK:foo1"
197 r = bdb.check_directory(contents)
198 self.failUnless(r.was_created())
199 self.failUnlessEqual(r.was_created(), dircap)
200 self.failUnlessEqual(r.should_check(), False)
202 # if we spontaneously decide to upload it anyways, nothing should
205 r = bdb.check_directory(contents)
206 self.failUnless(r.was_created())
207 self.failUnlessEqual(r.was_created(), dircap)
208 self.failUnlessEqual(type(r.was_created()), str)
209 self.failUnlessEqual(r.should_check(), False)
211 bdb.NO_CHECK_BEFORE = 0
212 bdb.ALWAYS_CHECK_AFTER = 0.1
215 r = bdb.check_directory(contents)
216 self.failUnless(r.was_created())
217 self.failUnlessEqual(r.was_created(), dircap)
218 self.failUnlessEqual(r.should_check(), True)
219 r.did_check_healthy("results")
221 bdb.NO_CHECK_BEFORE = 200
222 bdb.ALWAYS_CHECK_AFTER = 400
224 r = bdb.check_directory(contents)
225 self.failUnless(r.was_created())
226 self.failUnlessEqual(r.was_created(), dircap)
227 self.failUnlessEqual(r.should_check(), False)
230 contents2 = {u"file1": "URI:CHK:blah1",
231 u"dir1": "URI:DIR2-CHK:baz2"}
232 r = bdb.check_directory(contents2)
233 self.failIf(r.was_created())
235 contents3 = {u"file1": "URI:CHK:blah1",
236 u"file2": "URI:CHK:blah3",
237 u"dir1": "URI:DIR2-CHK:baz2"}
238 r = bdb.check_directory(contents3)
239 self.failIf(r.was_created())
241 def test_unicode(self):
242 self.skip_if_cannot_represent_filename(u"f\u00f6\u00f6.txt")
243 self.skip_if_cannot_represent_filename(u"b\u00e5r.txt")
245 self.basedir = basedir = os.path.join("backupdb", "unicode")
246 fileutil.make_dirs(basedir)
247 dbfile = os.path.join(basedir, "dbfile")
248 bdb = self.create_or_skip(dbfile)
251 self.writeto(u"f\u00f6\u00f6.txt", "foo.txt")
252 files = [fn for fn in listdir_unicode(unicode(basedir)) if fn.endswith(".txt")]
253 self.failUnlessEqual(len(files), 1)
254 foo_fn = os.path.join(basedir, files[0])
255 #print foo_fn, type(foo_fn)
257 r = bdb.check_file(foo_fn)
258 self.failUnlessEqual(r.was_uploaded(), False)
259 r.did_upload("foo-cap")
261 r = bdb.check_file(foo_fn)
262 self.failUnlessEqual(r.was_uploaded(), "foo-cap")
263 self.failUnlessEqual(r.should_check(), False)
265 bar_fn = self.writeto(u"b\u00e5r.txt", "bar.txt")
266 #print bar_fn, type(bar_fn)
268 r = bdb.check_file(bar_fn)
269 self.failUnlessEqual(r.was_uploaded(), False)
270 r.did_upload("bar-cap")
272 r = bdb.check_file(bar_fn)
273 self.failUnlessEqual(r.was_uploaded(), "bar-cap")
274 self.failUnlessEqual(r.should_check(), False)