- self.store.storeValue(self.key, 'barfoo')
- self.store.expireValues(1)
- val = self.store.retrieveValues(self.key)
- self.failUnlessEqual(len(val), 1)
- self.failUnlessEqual(val[0], 'barfoo')
-
- def test_RoutingTable(self):
- class dummy:
- id = self.key
- host = "127.0.0.1"
- port = 9977
- def contents(self):
- return (self.id, self.host, self.port)
- dummy2 = dummy()
- dummy2.id = '\xaa\xbb\xcc\x0c\x00\xe7\x07\xf8~])\x8f\x9d\xe5_B\xff\x1a\xc4!'
- dummy2.host = '205.23.67.124'
- dummy2.port = 12345
- class bl:
- def __init__(self):
- self.l = []
- bl1 = bl()
- bl1.l.append(dummy())
- bl2 = bl()
- bl2.l.append(dummy2)
- buckets = [bl1, bl2]
- self.store.dumpRoutingTable(buckets)
- rt = self.store.getRoutingTable()
- self.failUnlessIn(dummy().contents(), rt)
- self.failUnlessIn(dummy2.contents(), rt)
+ os.utime(self.path, None)
+ res = self.store.isUnchanged(self.path)
+ self.failUnless(res == False)
+ os.unlink(self.path)
+ res = self.store.isUnchanged(self.path)
+ self.failUnless(res == None)
+
+ def test_expiry(self):
+ res = self.store.expiredFiles(1)
+ self.failUnlessEqual(len(res.keys()), 0)
+ sleep(2)
+ res = self.store.expiredFiles(1)
+ self.failUnlessEqual(len(res.keys()), 1)
+ self.failUnlessEqual(res.keys()[0], self.hash)
+ self.failUnlessEqual(len(res[self.hash]), 1)
+ self.failUnlessEqual(res[self.hash][0], self.urlpath)
+ res = self.store.refreshFile(self.path)
+ self.failUnless(res)
+ res = self.store.expiredFiles(1)
+ self.failUnlessEqual(len(res.keys()), 0)
+
+ def test_removeUntracked(self):
+ for dir in self.dirs:
+ path = os.path.join(dir, self.path[1:])
+ os.makedirs(os.path.dirname(path))
+ f = open(path, 'w')
+ f.write(path)
+ f.close()
+ os.utime(path, None)
+ self.store.storeFile(path, self.hash, self.urlpath)
+
+ res = self.store.removeUntrackedFiles(self.dirs)
+ self.failUnlessEqual(len(res), 1, 'Got removed paths: %r' % res)
+ self.failUnlessEqual(res[0], self.path, 'Got removed paths: %r' % res)
+ res = self.store.removeUntrackedFiles(self.dirs)
+ self.failUnlessEqual(len(res), 0, 'Got removed paths: %r' % res)
+ res = self.store.removeUntrackedFiles(self.dirs[1:])
+ self.failUnlessEqual(len(res), 1, 'Got removed paths: %r' % res)
+ self.failUnlessEqual(res[0], os.path.join(self.dirs[0], self.path[1:]), 'Got removed paths: %r' % res)
+ res = self.store.removeUntrackedFiles(self.dirs[:1])
+ self.failUnlessEqual(len(res), 2, 'Got removed paths: %r' % res)
+ self.failUnlessIn(os.path.join(self.dirs[1], self.path[1:]), res, 'Got removed paths: %r' % res)
+ self.failUnlessIn(os.path.join(self.dirs[2], self.path[1:]), res, 'Got removed paths: %r' % res)