ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-04-07 10:56:49 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

workqueue: more tests, coverage now at 63.4%, yay

Browse Source
This commit is contained in:
Brian Warner 2007-01-09 01:36:12 -07:00
parent b641f6cbc7
commit d61c0a6ef6
1 changed files with 37 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
src/allmydata/test/test_workqueue.py
View File

@ -15,6 +15,37 @@ class FakeWorkQueue(workqueue.WorkQueue):
self.dispatched_steps.append((steptype, lines))
return defer.succeed(None)
class Reuse(unittest.TestCase):
def wq(self, testname):
return FakeWorkQueue("test_workqueue/Reuse/%s/workqueue" % testname)
def testOne(self):
wq = self.wq("testOne")
# steps must be retained from one session to the next
wq.add_upload_chk("source_filename", "box1")
wq.add_unlink_uri("someuri")
# files in the tmpdir are not: these are either in the process of
# being added or in the process of being removed.
tmpfile = os.path.join(wq.tmpdir, "foo")
f = open(tmpfile, "w")
f.write("foo")
f.close()
# files created with create_tempfile *are* retained, however
f, filename = wq.create_tempfile()
filename = os.path.join(wq.filesdir, filename)
f.write("bar")
f.close()
del wq
wq2 = self.wq("testOne")
steps = wq2.get_all_steps()
self.failUnlessEqual(steps[0], ("upload_chk",
["source_filename", "box1"]))
self.failUnlessEqual(steps[1], ("unlink_uri", ["someuri"]))
self.failIf(os.path.exists(tmpfile))
self.failUnless(os.path.exists(filename))
class Items(unittest.TestCase):
def wq(self, testname):
return FakeWorkQueue("test_workqueue/Items/%s/workqueue" % testname)
@ -31,6 +62,12 @@ class Items(unittest.TestCase):
f2.close()
self.failUnlessEqual(data, data2)
def testBox(self):
wq = self.wq("testBox")
boxname = wq.create_boxname()
wq.write_to_box(boxname, "contents of box")
self.failUnlessEqual(wq.read_from_box(boxname), "contents of box")
def testCHK(self):
wq = self.wq("testCHK")
wq.add_upload_chk("source_filename", "box1")