lib/bup/t/tindex.py

   1 import os
   2 import time
   3 from bup import index
   4 from bup.helpers import *
   5 import bup.xstat as xstat
   6 from wvtest import *
   7
   8 @wvtest
   9 def index_basic():
  10     cd = os.path.realpath('../../../t')
  11     WVPASS(cd)
  12     sd = os.path.realpath(cd + '/sampledata')
  13     WVPASSEQ(index.realpath(cd + '/sampledata'), cd + '/sampledata')
  14     WVPASSEQ(os.path.realpath(cd + '/sampledata/x'), sd + '/x')
  15     WVPASSEQ(os.path.realpath(cd + '/sampledata/etc'), os.path.realpath('/etc'))
  16     WVPASSEQ(index.realpath(cd + '/sampledata/etc'), sd + '/etc')
  17
  18
  19 @wvtest
  20 def index_writer():
  21     unlink('index.tmp')
  22     ds = xstat.stat('.')
  23     fs = xstat.stat('tindex.py')
  24     w = index.Writer('index.tmp', time.time() - 1)
  25     w.add('/var/tmp/sporky', fs)
  26     w.add('/etc/passwd', fs)
  27     w.add('/etc/', ds)
  28     w.add('/', ds)
  29     w.close()
  30
  31
  32 def dump(m):
  33     for e in list(m):
  34         print '%s%s %s' % (e.is_valid() and ' ' or 'M',
  35                            e.is_fake() and 'F' or ' ',
  36                            e.name)
  37
  38 def fake_validate(*l):
  39     for i in l:
  40         for e in i:
  41             e.validate(0100644, index.FAKE_SHA)
  42             e.repack()
  43
  44 def eget(l, ename):
  45     for e in l:
  46         if e.name == ename:
  47             return e
  48
  49 @wvtest
  50 def index_negative_timestamps():
  51     # Makes 'foo' exist
  52     f = file('foo', 'wb')
  53     f.close()
  54
  55     # Dec 31, 1969
  56     os.utime("foo", (-86400, -86400))
  57     now = time.time()
  58     e = index.BlankNewEntry("foo", now - 1)
  59     e.from_stat(xstat.stat("foo"), now)
  60     assert len(e.packed())
  61     WVPASS()
  62
  63     # Jun 10, 1893
  64     os.utime("foo", (-0x80000000, -0x80000000))
  65     e = index.BlankNewEntry("foo", now - 1)
  66     e.from_stat(xstat.stat("foo"), now)
  67     assert len(e.packed())
  68     WVPASS()
  69
  70     unlink('foo')
  71
  72
  73 @wvtest
  74 def index_dirty():
  75     unlink('index.tmp')
  76     unlink('index2.tmp')
  77     ds = xstat.stat('.')
  78     fs = xstat.stat('tindex.py')
  79     tmax = time.time() - 1
  80
  81     w1 = index.Writer('index.tmp', tmax)
  82     w1.add('/a/b/x', fs)
  83     w1.add('/a/b/c', fs)
  84     w1.add('/a/b/', ds)
  85     w1.add('/a/', ds)
  86     #w1.close()
  87     WVPASS()
  88
  89     w2 = index.Writer('index2.tmp', tmax)
  90     w2.add('/a/b/n/2', fs)
  91     #w2.close()
  92     WVPASS()
  93
  94     w3 = index.Writer('index3.tmp', tmax)
  95     w3.add('/a/c/n/3', fs)
  96     #w3.close()
  97     WVPASS()
  98
  99     r1 = w1.new_reader()
 100     r2 = w2.new_reader()
 101     r3 = w3.new_reader()
 102     WVPASS()
 103
 104     r1all = [e.name for e in r1]
 105     WVPASSEQ(r1all,
 106              ['/a/b/x', '/a/b/c', '/a/b/', '/a/', '/'])
 107     r2all = [e.name for e in r2]
 108     WVPASSEQ(r2all,
 109              ['/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
 110     r3all = [e.name for e in r3]
 111     WVPASSEQ(r3all,
 112              ['/a/c/n/3', '/a/c/n/', '/a/c/', '/a/', '/'])
 113     all = [e.name for e in index.merge(r2, r1, r3)]
 114     WVPASSEQ(all,
 115              ['/a/c/n/3', '/a/c/n/', '/a/c/',
 116               '/a/b/x', '/a/b/n/2', '/a/b/n/', '/a/b/c',
 117               '/a/b/', '/a/', '/'])
 118     fake_validate(r1)
 119     dump(r1)
 120
 121     print [hex(e.flags) for e in r1]
 122     WVPASSEQ([e.name for e in r1 if e.is_valid()], r1all)
 123     WVPASSEQ([e.name for e in r1 if not e.is_valid()], [])
 124     WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
 125              ['/a/c/n/3', '/a/c/n/', '/a/c/',
 126               '/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
 127
 128     expect_invalid = ['/'] + r2all + r3all
 129     expect_real = (set(r1all) - set(r2all) - set(r3all)) \
 130                     | set(['/a/b/n/2', '/a/c/n/3'])
 131     dump(index.merge(r2, r1, r3))
 132     for e in index.merge(r2, r1, r3):
 133         print e.name, hex(e.flags), e.ctime
 134         eiv = e.name in expect_invalid
 135         er  = e.name in expect_real
 136         WVPASSEQ(eiv, not e.is_valid())
 137         WVPASSEQ(er, e.is_real())
 138     fake_validate(r2, r3)
 139     dump(index.merge(r2, r1, r3))
 140     WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()], [])
 141
 142     e = eget(index.merge(r2, r1, r3), '/a/b/c')
 143     e.invalidate()
 144     e.repack()
 145     dump(index.merge(r2, r1, r3))
 146     WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
 147              ['/a/b/c', '/a/b/', '/a/', '/'])