lib/bup/t/tindex.py

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