lbry-rocksdb/rocksdb/tests/test_options.py
2014-04-27 19:20:30 +02:00

71 lines
2.1 KiB
Python

import unittest
import rocksdb
class TestFilterPolicy(rocksdb.interfaces.FilterPolicy):
def create_filter(self, keys):
return b'nix'
def key_may_match(self, key, fil):
return True
def name(self):
return b'testfilter'
class TestMergeOperator(rocksdb.interfaces.MergeOperator):
def full_merge(self, *args, **kwargs):
return (False, None)
def partial_merge(self, *args, **kwargs):
return (False, None)
def name(self):
return b'testmergeop'
class TestOptions(unittest.TestCase):
def test_simple(self):
opts = rocksdb.Options()
self.assertEqual(True, opts.paranoid_checks)
opts.paranoid_checks = False
self.assertEqual(False, opts.paranoid_checks)
self.assertIsNone(opts.filter_policy)
ob = TestFilterPolicy()
opts.filter_policy = ob
self.assertEqual(opts.filter_policy, ob)
self.assertIsNone(opts.merge_operator)
ob = TestMergeOperator()
opts.merge_operator = ob
self.assertEqual(opts.merge_operator, ob)
self.assertIsInstance(
opts.comparator,
rocksdb.BytewiseComparator)
self.assertEqual('snappy_compression', opts.compression)
opts.compression = rocksdb.CompressionType.no_compression
self.assertEqual('no_compression', opts.compression)
self.assertEqual(opts.block_size, 4096)
self.assertIsNone(opts.block_cache)
ob = rocksdb.LRUCache(100)
opts.block_cache = ob
self.assertEqual(ob, opts.block_cache)
def test_unicode_path(self):
name = b'/tmp/M\xc3\xbcnchen'.decode('utf8')
opts = rocksdb.Options()
opts.db_log_dir = name
opts.wal_dir = name
self.assertEqual(name, opts.db_log_dir)
self.assertEqual(name, opts.wal_dir)
def test_table_factory(self):
opts = rocksdb.Options()
self.assertIsNone(opts.table_factory)
opts.table_factory = rocksdb.BlockBasedTableFactory()
opts.table_factory = rocksdb.PlainTableFactory()
opts.table_factory = rocksdb.TotalOrderPlainTableFactory()