lib/bup/t/tindex.py

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