lib/bup/t/tbloom.py

   1
   2 import errno, platform, subprocess, tempfile
   3
   4 from bup import bloom
   5 from bup.helpers import mkdirp
   6
   7 from wvtest import *
   8
   9
  10 bup_tmp = os.path.realpath('../../../t/tmp')
  11 mkdirp(bup_tmp)
  12
  13 @wvtest
  14 def test_bloom():
  15     initial_failures = wvfailure_count()
  16     tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tbloom-')
  17     hashes = [os.urandom(20) for i in range(100)]
  18     class Idx:
  19         pass
  20     ix = Idx()
  21     ix.name='dummy.idx'
  22     ix.shatable = ''.join(hashes)
  23     for k in (4, 5):
  24         b = bloom.create(tmpdir + '/pybuptest.bloom', expected=100, k=k)
  25         b.add_idx(ix)
  26         WVPASSLT(b.pfalse_positive(), .1)
  27         b.close()
  28         b = bloom.ShaBloom(tmpdir + '/pybuptest.bloom')
  29         all_present = True
  30         for h in hashes:
  31             all_present &= b.exists(h)
  32         WVPASS(all_present)
  33         false_positives = 0
  34         for h in [os.urandom(20) for i in range(1000)]:
  35             if b.exists(h):
  36                 false_positives += 1
  37         WVPASSLT(false_positives, 5)
  38         os.unlink(tmpdir + '/pybuptest.bloom')
  39
  40     tf = tempfile.TemporaryFile()
  41     b = bloom.create('bup.bloom', f=tf, expected=100)
  42     WVPASSEQ(b.rwfile, tf)
  43     WVPASSEQ(b.k, 5)
  44
  45     # Test large (~1GiB) filter.  This may fail on s390 (31-bit
  46     # architecture), and anywhere else where the address space is
  47     # sufficiently limited.
  48     tf = tempfile.TemporaryFile()
  49     skip_test = False
  50     try:
  51         b = bloom.create('bup.bloom', f=tf, expected=2**28, delaywrite=False)
  52     except EnvironmentError as ex:
  53         (ptr_width, linkage) = platform.architecture()
  54         if ptr_width == '32bit' and ex.errno == errno.ENOMEM:
  55             WVMSG('skipping large bloom filter test (mmap probably failed) '
  56                   + str(ex))
  57             skip_test = True
  58         else:
  59             raise
  60     if not skip_test:
  61         WVPASSEQ(b.k, 4)
  62     if wvfailure_count() == initial_failures:
  63         subprocess.call(['rm', '-rf', tmpdir])