test/int/test_bloom.py

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