lib/bup/t/tgit.py

   1 import struct, os, tempfile, time
   2 from bup import git
   3 from bup.helpers import *
   4 from wvtest import *
   5
   6
   7 @wvtest
   8 def testmangle():
   9     afile  = 0100644
  10     afile2 = 0100770
  11     alink  = 0120000
  12     adir   = 0040000
  13     adir2  = 0040777
  14     WVPASSEQ(git.mangle_name("a", adir2, adir), "a")
  15     WVPASSEQ(git.mangle_name(".bup", adir2, adir), ".bup.bupl")
  16     WVPASSEQ(git.mangle_name("a.bupa", adir2, adir), "a.bupa.bupl")
  17     WVPASSEQ(git.mangle_name("b.bup", alink, alink), "b.bup.bupl")
  18     WVPASSEQ(git.mangle_name("b.bu", alink, alink), "b.bu")
  19     WVPASSEQ(git.mangle_name("f", afile, afile2), "f")
  20     WVPASSEQ(git.mangle_name("f.bup", afile, afile2), "f.bup.bupl")
  21     WVPASSEQ(git.mangle_name("f.bup", afile, adir), "f.bup.bup")
  22     WVPASSEQ(git.mangle_name("f", afile, adir), "f.bup")
  23
  24     WVPASSEQ(git.demangle_name("f.bup"), ("f", git.BUP_CHUNKED))
  25     WVPASSEQ(git.demangle_name("f.bupl"), ("f", git.BUP_NORMAL))
  26     WVPASSEQ(git.demangle_name("f.bup.bupl"), ("f.bup", git.BUP_NORMAL))
  27
  28     # for safety, we ignore .bup? suffixes we don't recognize.  Future
  29     # versions might implement a .bup[a-z] extension as something other
  30     # than BUP_NORMAL.
  31     WVPASSEQ(git.demangle_name("f.bupa"), ("f.bupa", git.BUP_NORMAL))
  32
  33
  34 @wvtest
  35 def testencode():
  36     s = 'hello world'
  37     looseb = ''.join(git._encode_looseobj('blob', s))
  38     looset = ''.join(git._encode_looseobj('tree', s))
  39     loosec = ''.join(git._encode_looseobj('commit', s))
  40     packb = ''.join(git._encode_packobj('blob', s))
  41     packt = ''.join(git._encode_packobj('tree', s))
  42     packc = ''.join(git._encode_packobj('commit', s))
  43     WVPASSEQ(git._decode_looseobj(looseb), ('blob', s))
  44     WVPASSEQ(git._decode_looseobj(looset), ('tree', s))
  45     WVPASSEQ(git._decode_looseobj(loosec), ('commit', s))
  46     WVPASSEQ(git._decode_packobj(packb), ('blob', s))
  47     WVPASSEQ(git._decode_packobj(packt), ('tree', s))
  48     WVPASSEQ(git._decode_packobj(packc), ('commit', s))
  49
  50
  51 @wvtest
  52 def testpacks():
  53     subprocess.call(['rm','-rf', 'pybuptest.tmp'])
  54     git.init_repo('pybuptest.tmp')
  55     git.verbose = 1
  56
  57     w = git.PackWriter()
  58     w.new_blob(os.urandom(100))
  59     w.new_blob(os.urandom(100))
  60     w.abort()
  61
  62     w = git.PackWriter()
  63     hashes = []
  64     nobj = 1000
  65     for i in range(nobj):
  66         hashes.append(w.new_blob(str(i)))
  67     log('\n')
  68     nameprefix = w.close()
  69     print repr(nameprefix)
  70     WVPASS(os.path.exists(nameprefix + '.pack'))
  71     WVPASS(os.path.exists(nameprefix + '.idx'))
  72
  73     r = git.open_idx(nameprefix + '.idx')
  74     print repr(r.fanout)
  75
  76     for i in range(nobj):
  77         WVPASS(r.find_offset(hashes[i]) > 0)
  78     WVPASS(r.exists(hashes[99]))
  79     WVFAIL(r.exists('\0'*20))
  80
  81     pi = iter(r)
  82     for h in sorted(hashes):
  83         WVPASSEQ(str(pi.next()).encode('hex'), h.encode('hex'))
  84
  85     WVFAIL(r.find_offset('\0'*20))
  86
  87     r = git.PackIdxList('pybuptest.tmp/objects/pack')
  88     WVPASS(r.exists(hashes[5]))
  89     WVPASS(r.exists(hashes[6]))
  90     WVFAIL(r.exists('\0'*20))
  91
  92
  93 @wvtest
  94 def test_pack_name_lookup():
  95     os.environ['BUP_MAIN_EXE'] = bupmain = '../../../bup'
  96     os.environ['BUP_DIR'] = bupdir = 'pybuptest.tmp'
  97     subprocess.call(['rm','-rf', bupdir])
  98     git.init_repo(bupdir)
  99     git.verbose = 1
 100
 101     idxnames = []
 102
 103     w = git.PackWriter()
 104     hashes = []
 105     for i in range(2):
 106         hashes.append(w.new_blob(str(i)))
 107     log('\n')
 108     idxnames.append(w.close() + '.idx')
 109
 110     w = git.PackWriter()
 111     for i in range(2,4):
 112         hashes.append(w.new_blob(str(i)))
 113     log('\n')
 114     idxnames.append(w.close() + '.idx')
 115
 116     idxnames = [os.path.basename(ix) for ix in idxnames]
 117
 118     def verify(r):
 119         for i in range(2):
 120             WVPASSEQ(r.exists(hashes[i], want_source=True), idxnames[0])
 121         for i in range(2,4):
 122             WVPASSEQ(r.exists(hashes[i], want_source=True), idxnames[1])
 123
 124     r = git.PackIdxList('pybuptest.tmp/objects/pack')
 125     WVPASSEQ(len(r.packs), 2)
 126     verify(r)
 127     del r
 128
 129     subprocess.call([bupmain, 'midx', '-f'])
 130
 131     r = git.PackIdxList('pybuptest.tmp/objects/pack')
 132     WVPASSEQ(len(r.packs), 1)
 133     verify(r)
 134
 135
 136 @wvtest
 137 def test_long_index():
 138     w = git.PackWriter()
 139     obj_bin = struct.pack('!IIIII',
 140             0x00112233, 0x44556677, 0x88990011, 0x22334455, 0x66778899)
 141     obj2_bin = struct.pack('!IIIII',
 142             0x11223344, 0x55667788, 0x99001122, 0x33445566, 0x77889900)
 143     obj3_bin = struct.pack('!IIIII',
 144             0x22334455, 0x66778899, 0x00112233, 0x44556677, 0x88990011)
 145     pack_bin = struct.pack('!IIIII',
 146             0x99887766, 0x55443322, 0x11009988, 0x77665544, 0x33221100)
 147     idx = list(list() for i in xrange(256))
 148     idx[0].append((obj_bin, 1, 0xfffffffff))
 149     idx[0x11].append((obj2_bin, 2, 0xffffffffff))
 150     idx[0x22].append((obj3_bin, 3, 0xff))
 151     (fd,name) = tempfile.mkstemp(suffix='.idx', dir=git.repo('objects'))
 152     f = os.fdopen(fd, 'w+b')
 153     r = w._write_pack_idx_v2(f, idx, pack_bin)
 154     f.seek(0)
 155     i = git.PackIdxV2(name, f)
 156     WVPASSEQ(i.find_offset(obj_bin), 0xfffffffff)
 157     WVPASSEQ(i.find_offset(obj2_bin), 0xffffffffff)
 158     WVPASSEQ(i.find_offset(obj3_bin), 0xff)
 159     f.close()
 160     os.remove(name)
 161
 162 @wvtest
 163 def test_bloom():
 164     hashes = [os.urandom(20) for i in range(100)]
 165     class Idx:
 166         pass
 167     ix = Idx()
 168     ix.name='dummy.idx'
 169     ix.shatable = ''.join(hashes)
 170     for k in (4, 5):
 171         b = git.ShaBloom.create('pybuptest.bloom', expected=100, k=k)
 172         b.add_idx(ix)
 173         WVPASSLT(b.pfalse_positive(), .1)
 174         b.close()
 175         b = git.ShaBloom('pybuptest.bloom')
 176         all_present = True
 177         for h in hashes:
 178             all_present &= b.exists(h)
 179         WVPASS(all_present)
 180         false_positives = 0
 181         for h in [os.urandom(20) for i in range(1000)]:
 182             if b.exists(h):
 183                 false_positives += 1
 184         WVPASSLT(false_positives, 5)
 185         os.unlink('pybuptest.bloom')
 186
 187     tf = tempfile.TemporaryFile()
 188     b = git.ShaBloom.create('bup.bloom', f=tf, expected=100)
 189     WVPASSEQ(b.rwfile, tf)
 190     WVPASSEQ(b.k, 5)
 191     tf = tempfile.TemporaryFile()
 192     b = git.ShaBloom.create('bup.bloom', f=tf, expected=2**28,
 193                             delaywrite=False)
 194     WVPASSEQ(b.k, 4)