lib/bup/t/tindex.py

   1 import os
   2 import time
   3 from bup import index, metadata
   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     unlink('index.meta.tmp')
  25     ms = index.MetaStoreWriter('index.meta.tmp');
  26     tmax = (time.time() - 1) * 10**9
  27     w = index.Writer('index.tmp', ms, tmax)
  28     w.add('/var/tmp/sporky', fs, 0)
  29     w.add('/etc/passwd', fs, 0)
  30     w.add('/etc/', ds, 0)
  31     w.add('/', ds, 0)
  32     ms.close()
  33     w.close()
  34
  35
  36 def dump(m):
  37     for e in list(m):
  38         print '%s%s %s' % (e.is_valid() and ' ' or 'M',
  39                            e.is_fake() and 'F' or ' ',
  40                            e.name)
  41
  42 def fake_validate(*l):
  43     for i in l:
  44         for e in i:
  45             e.validate(0100644, index.FAKE_SHA)
  46             e.repack()
  47
  48 def eget(l, ename):
  49     for e in l:
  50         if e.name == ename:
  51             return e
  52
  53 @wvtest
  54 def index_negative_timestamps():
  55     # Makes 'foo' exist
  56     f = file('foo', 'wb')
  57     f.close()
  58
  59     # Dec 31, 1969
  60     os.utime("foo", (-86400, -86400))
  61     ns_per_sec = 10**9
  62     tstart = time.time() * ns_per_sec
  63     tmax = tstart - ns_per_sec
  64     e = index.BlankNewEntry("foo", 0, tmax)
  65     e.from_stat(xstat.stat("foo"), 0, tstart)
  66     assert len(e.packed())
  67     WVPASS()
  68
  69     # Jun 10, 1893
  70     os.utime("foo", (-0x80000000, -0x80000000))
  71     e = index.BlankNewEntry("foo", 0, tmax)
  72     e.from_stat(xstat.stat("foo"), 0, tstart)
  73     assert len(e.packed())
  74     WVPASS()
  75
  76     unlink('foo')
  77
  78
  79 @wvtest
  80 def index_dirty():
  81     unlink('index.meta.tmp')
  82     unlink('index2.meta.tmp')
  83     unlink('index3.meta.tmp')
  84     default_meta = metadata.Metadata()
  85     ms1 = index.MetaStoreWriter('index.meta.tmp')
  86     ms2 = index.MetaStoreWriter('index2.meta.tmp')
  87     ms3 = index.MetaStoreWriter('index3.meta.tmp')
  88     meta_ofs1 = ms1.store(default_meta)
  89     meta_ofs2 = ms2.store(default_meta)
  90     meta_ofs3 = ms3.store(default_meta)
  91     unlink('index.tmp')
  92     unlink('index2.tmp')
  93     unlink('index3.tmp')
  94
  95     ds = xstat.stat('.')
  96     fs = xstat.stat('tindex.py')
  97     tmax = (time.time() - 1) * 10**9
  98
  99     w1 = index.Writer('index.tmp', ms1, tmax)
 100     w1.add('/a/b/x', fs, meta_ofs1)
 101     w1.add('/a/b/c', fs, meta_ofs1)
 102     w1.add('/a/b/', ds, meta_ofs1)
 103     w1.add('/a/', ds, meta_ofs1)
 104     #w1.close()
 105     WVPASS()
 106
 107     w2 = index.Writer('index2.tmp', ms2, tmax)
 108     w2.add('/a/b/n/2', fs, meta_ofs2)
 109     #w2.close()
 110     WVPASS()
 111
 112     w3 = index.Writer('index3.tmp', ms3, tmax)
 113     w3.add('/a/c/n/3', fs, meta_ofs3)
 114     #w3.close()
 115     WVPASS()
 116
 117     r1 = w1.new_reader()
 118     r2 = w2.new_reader()
 119     r3 = w3.new_reader()
 120     WVPASS()
 121
 122     r1all = [e.name for e in r1]
 123     WVPASSEQ(r1all,
 124              ['/a/b/x', '/a/b/c', '/a/b/', '/a/', '/'])
 125     r2all = [e.name for e in r2]
 126     WVPASSEQ(r2all,
 127              ['/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
 128     r3all = [e.name for e in r3]
 129     WVPASSEQ(r3all,
 130              ['/a/c/n/3', '/a/c/n/', '/a/c/', '/a/', '/'])
 131     all = [e.name for e in index.merge(r2, r1, r3)]
 132     WVPASSEQ(all,
 133              ['/a/c/n/3', '/a/c/n/', '/a/c/',
 134               '/a/b/x', '/a/b/n/2', '/a/b/n/', '/a/b/c',
 135               '/a/b/', '/a/', '/'])
 136     fake_validate(r1)
 137     dump(r1)
 138
 139     print [hex(e.flags) for e in r1]
 140     WVPASSEQ([e.name for e in r1 if e.is_valid()], r1all)
 141     WVPASSEQ([e.name for e in r1 if not e.is_valid()], [])
 142     WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
 143              ['/a/c/n/3', '/a/c/n/', '/a/c/',
 144               '/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
 145
 146     expect_invalid = ['/'] + r2all + r3all
 147     expect_real = (set(r1all) - set(r2all) - set(r3all)) \
 148                     | set(['/a/b/n/2', '/a/c/n/3'])
 149     dump(index.merge(r2, r1, r3))
 150     for e in index.merge(r2, r1, r3):
 151         print e.name, hex(e.flags), e.ctime
 152         eiv = e.name in expect_invalid
 153         er  = e.name in expect_real
 154         WVPASSEQ(eiv, not e.is_valid())
 155         WVPASSEQ(er, e.is_real())
 156     fake_validate(r2, r3)
 157     dump(index.merge(r2, r1, r3))
 158     WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()], [])
 159
 160     e = eget(index.merge(r2, r1, r3), '/a/b/c')
 161     e.invalidate()
 162     e.repack()
 163     dump(index.merge(r2, r1, r3))
 164     WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
 165              ['/a/b/c', '/a/b/', '/a/', '/'])