# Copyright (c) 2013, Kevin Greenan (kmgreen2@gmail.com) # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # # Redistributions of source code must retain the above copyright notice, this # list of conditions and the following disclaimer. # # Redistributions in binary form must reproduce the above copyright notice, # this list of conditions and the following disclaimer in the documentation # and/or other materials provided with the distribution. THIS SOFTWARE IS # PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES # OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN # NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY # DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import os import random import resource import string import sys import tempfile import unittest from distutils.version import StrictVersion from itertools import combinations import six import pyeclib.ec_iface from pyeclib.ec_iface import ECBackendNotSupported from pyeclib.ec_iface import ECDriver from pyeclib.ec_iface import ECDriverError from pyeclib.ec_iface import ECInsufficientFragments from pyeclib.ec_iface import ECInvalidFragmentMetadata from pyeclib.ec_iface import ECInvalidParameter from pyeclib.ec_iface import PyECLib_EC_Types from pyeclib.ec_iface import ALL_EC_TYPES from pyeclib.ec_iface import VALID_EC_TYPES from pyeclib.ec_iface import LIBERASURECODE_VERSION if sys.version < '3': def b2i(b): return ord(b) else: def b2i(b): return b class TestNullDriver(unittest.TestCase): def setUp(self): self.null_driver = ECDriver( library_import_str="pyeclib.core.ECNullDriver", k=8, m=2) def tearDown(self): pass def test_null_driver(self): self.null_driver.encode('') self.null_driver.decode([]) class TestStripeDriver(unittest.TestCase): def setUp(self): self.stripe_driver = ECDriver( library_import_str="pyeclib.core.ECStripingDriver", k=8, m=0) def tearDown(self): pass class TestPyECLibDriver(unittest.TestCase): def __init__(self, *args): # Create the temp files needed for testing self.file_sizes = ["100-K"] self.files = {} self.num_iterations = 100 self._create_tmp_files() unittest.TestCase.__init__(self, *args) def _create_tmp_files(self): """ Create the temporary files needed for testing. Use the tempfile package so that the files will be automatically removed during garbage collection. """ for size_str in self.file_sizes: # Determine the size of the file to create size_desc = size_str.split("-") size = int(size_desc[0]) if size_desc[1] == 'M': size *= 1000000 elif size_desc[1] == 'K': size *= 1000 # Create the dictionary of files to test with buf = ''.join(random.choice(string.ascii_letters) for i in range(size)) if sys.version_info >= (3,): buf = buf.encode('ascii') tmp_file = tempfile.NamedTemporaryFile() tmp_file.write(buf) self.files[size_str] = tmp_file def setUp(self): # Ensure that the file offset is set to the head of the file for _, tmp_file in self.files.items(): tmp_file.seek(0, 0) def tearDown(self): pass def test_missing_required_args(self): # missing ec_type with self.assertRaises(ECDriverError) as err_context: ECDriver(k=1, m=1) self.assertEqual(str(err_context.exception), "Invalid Argument: either ec_type or " "library_import_str must be provided") for ec_type in VALID_EC_TYPES: # missing k with self.assertRaises(ECDriverError) as err_context: ECDriver(ec_type=ec_type, m=1) self.assertEqual(str(err_context.exception), "Invalid Argument: k is required") # missing m with self.assertRaises(ECDriverError) as err_context: ECDriver(ec_type=ec_type, k=1) self.assertEqual(str(err_context.exception), "Invalid Argument: m is required") def test_invalid_km_args(self): for ec_type in VALID_EC_TYPES: with self.assertRaises(ECDriverError) as err_context: # k is smaller than 1 ECDriver(ec_type=ec_type, k=-100, m=1) self.assertEqual(str(err_context.exception), "Invalid number of data fragments (k)") with self.assertRaises(ECDriverError) as err_context: # m is smaller than 1 ECDriver(ec_type=ec_type, k=1, m=-100) self.assertEqual(str(err_context.exception), "Invalid number of parity fragments (m)") def test_valid_ec_types(self): # Build list of available types and compare to VALID_EC_TYPES available_ec_types = [] for _type in ALL_EC_TYPES: try: if _type == 'shss': _k = 10 _m = 4 elif _type == 'libphazr': _k = 4 _m = 4 else: _k = 10 _m = 5 ECDriver(k=_k, m=_m, ec_type=_type, validate=True) available_ec_types.append(_type) except Exception: # ignore any errors, assume backend not available pass self.assertEqual(available_ec_types, VALID_EC_TYPES) def test_valid_algo(self): print("") for _type in ALL_EC_TYPES: # Check if this algo works if _type not in VALID_EC_TYPES: print("Skipping test for %s backend" % _type) continue try: if _type == 'shss': ECDriver(k=10, m=4, ec_type=_type) elif _type == 'libphazr': ECDriver(k=4, m=4, ec_type=_type) else: ECDriver(k=10, m=5, ec_type=_type) except ECDriverError: self.fail("%s algorithm not supported" % _type) self.assertRaises(ECBackendNotSupported, ECDriver, k=10, m=5, ec_type="invalid_algo") def get_pyeclib_testspec(self, csum="none"): pyeclib_drivers = [] _type1 = 'jerasure_rs_vand' if _type1 in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=12, m=2, ec_type=_type1, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=11, m=2, ec_type=_type1, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=10, m=2, ec_type=_type1, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=8, m=4, ec_type=_type1, chksum_type=csum)) _type2 = 'liberasurecode_rs_vand' if _type2 in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=12, m=2, ec_type=_type2, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=11, m=2, ec_type=_type2, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=10, m=2, ec_type=_type2, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=8, m=4, ec_type=_type2, chksum_type=csum)) _type3_1 = 'flat_xor_hd' if _type3_1 in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=12, m=6, ec_type=_type3_1, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=10, m=5, ec_type=_type3_1, chksum_type=csum)) _type3_2 = 'flat_xor_hd_4' if _type3_2 in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=12, m=6, ec_type=_type3_2, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=10, m=5, ec_type=_type3_2, chksum_type=csum)) _type4 = 'shss' if _type4 in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=10, m=4, ec_type=_type4, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=20, m=4, ec_type=_type4, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=11, m=7, ec_type=_type4, chksum_type=csum)) _type5 = 'isa_l_rs_vand' if _type5 in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=12, m=2, ec_type=_type5, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=11, m=2, ec_type=_type5, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=10, m=2, ec_type=_type5, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=8, m=4, ec_type=_type5, chksum_type=csum)) _type6 = 'isa_l_rs_cauchy' if _type6 in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=12, m=2, ec_type=_type6, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=11, m=2, ec_type=_type6, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=10, m=2, ec_type=_type6, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=8, m=4, ec_type=_type6, chksum_type=csum)) pyeclib_drivers.append(ECDriver(k=11, m=7, ec_type=_type6, chksum_type=csum)) _type7 = 'libphazr' if _type7 in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=4, m=4, ec_type=_type7, chksum_type=csum)) return pyeclib_drivers def test_small_encode(self): pyeclib_drivers = self.get_pyeclib_testspec() encode_strs = [b"a", b"hello", b"hellohyhi", b"yo"] for pyeclib_driver in pyeclib_drivers: for encode_str in encode_strs: encoded_fragments = pyeclib_driver.encode(encode_str) decoded_str = pyeclib_driver.decode(encoded_fragments) self.assertTrue(decoded_str == encode_str) def test_encode_invalid_params(self): pyeclib_drivers = self.get_pyeclib_testspec() encode_args = [u'\U0001F0A1', 3, object(), None, {}, []] for pyeclib_driver in pyeclib_drivers: for encode_str in encode_args: with self.assertRaises(ECInvalidParameter): pyeclib_driver.encode(encode_str) def test_attribute_error_in_the_error_handling(self): pyeclib_drivers = self.get_pyeclib_testspec() self.assertGreater(len(pyeclib_drivers), 0) pyeclib_driver = pyeclib_drivers[0] del pyeclib.ec_iface.ECInvalidParameter try: with self.assertRaises(AttributeError): # !! pyeclib_driver.encode(3) finally: pyeclib.ec_iface.ECInvalidParameter = ECInvalidParameter def test_import_error_in_the_error_handling(self): pyeclib_drivers = self.get_pyeclib_testspec() self.assertGreater(len(pyeclib_drivers), 0) pyeclib_driver = pyeclib_drivers[0] from six.moves import builtins real_import = builtins.__import__ def fake_import(*a, **kw): raise ImportError builtins.__import__ = fake_import try: with self.assertRaises(ImportError): # !! pyeclib_driver.encode(3) finally: builtins.__import__ = real_import def test_decode_reconstruct_with_fragment_iterator(self): pyeclib_drivers = self.get_pyeclib_testspec() encode_strs = [b"a", b"hello", b"hellohyhi", b"yo"] for pyeclib_driver in pyeclib_drivers: for encode_str in encode_strs: encoded_fragments = pyeclib_driver.encode(encode_str) idxs_to_remove = random.sample(range( pyeclib_driver.k + pyeclib_driver.m), 2) available_fragments = encoded_fragments[:] for i in sorted(idxs_to_remove, reverse=True): available_fragments.pop(i) frag_iter = iter(available_fragments) decoded_str = pyeclib_driver.decode(frag_iter) self.assertEqual(decoded_str, encode_str) # Since the iterator is exhausted, we can't decode again with self.assertRaises(ECDriverError) as exc_mgr: pyeclib_driver.decode(frag_iter) self.assertEqual( 'Invalid fragment payload in ECPyECLibDriver.decode', str(exc_mgr.exception)) frag_iter = iter(available_fragments) reconstructed_fragments = pyeclib_driver.reconstruct( frag_iter, idxs_to_remove) self.assertEqual(len(reconstructed_fragments), len(idxs_to_remove)) for i, data in zip(idxs_to_remove, reconstructed_fragments): self.assertEqual(data, encoded_fragments[i]) # Since the iterator is exhausted, we can't decode again with self.assertRaises(ECDriverError) as exc_mgr: pyeclib_driver.reconstruct(frag_iter, idxs_to_remove) self.assertEqual( 'Invalid fragment payload in ECPyECLibDriver.reconstruct', str(exc_mgr.exception)) def check_metadata_formatted(self, k, m, ec_type, chksum_type): if ec_type not in VALID_EC_TYPES: return filesize = 1024 * 1024 * 3 file_str = ''.join(random.choice(string.ascii_letters) for i in range(filesize)) file_bytes = file_str.encode('utf-8') pyeclib_driver = ECDriver(k=k, m=m, ec_type=ec_type, chksum_type=chksum_type) fragments = pyeclib_driver.encode(file_bytes) f = 0 for fragment in fragments: metadata = pyeclib_driver.get_metadata(fragment, 1) if 'index' in metadata: self.assertEqual(metadata['index'], f) else: self.assertTrue(False) if 'chksum_mismatch' in metadata: self.assertEqual(metadata['chksum_mismatch'], 0) else: self.assertTrue(False) if 'backend_id' in metadata: self.assertEqual(metadata['backend_id'], ec_type) else: self.assertTrue(False) if 'orig_data_size' in metadata: self.assertEqual(metadata['orig_data_size'], 3145728) else: self.assertTrue(False) if 'chksum_type' in metadata: self.assertEqual(metadata['chksum_type'], 'crc32') else: self.assertTrue(False) if 'backend_version' not in metadata: self.assertTrue(False) if 'chksum' not in metadata: self.assertTrue(False) if 'size' not in metadata: self.assertTrue(False) f += 1 def test_get_metadata_formatted(self): self.check_metadata_formatted(12, 2, "jerasure_rs_vand", "inline_crc32") self.check_metadata_formatted(12, 2, "liberasurecode_rs_vand", "inline_crc32") self.check_metadata_formatted(8, 4, "liberasurecode_rs_vand", "inline_crc32") def test_verify_fragment_inline_chksum_fail(self): pyeclib_drivers = self.get_pyeclib_testspec("inline_crc32") filesize = 1024 * 1024 * 3 file_str = ''.join(random.choice(string.ascii_letters) for i in range(filesize)) file_bytes = file_str.encode('utf-8') for pyeclib_driver in pyeclib_drivers: fragments = pyeclib_driver.encode(file_bytes) fragment_metadata_list = [] first_fragment_to_corrupt = random.randint(0, len(fragments)) num_to_corrupt = 2 fragments_to_corrupt = [ (first_fragment_to_corrupt + i) % len(fragments) for i in range(num_to_corrupt + 1)] fragments_to_corrupt.sort() i = 0 for fragment in fragments: if i in fragments_to_corrupt: corrupted_fragment = fragment[:100] +\ (str(chr((b2i(fragment[100]) + 0x1) % 128))).encode('utf-8') + fragment[101:] fragment_metadata_list.append( pyeclib_driver.get_metadata(corrupted_fragment)) else: fragment_metadata_list.append( pyeclib_driver.get_metadata(fragment)) i += 1 expected_ret_value = {"status": -206, "reason": "Bad checksum", "bad_fragments": fragments_to_corrupt} self.assertEqual( pyeclib_driver.verify_stripe_metadata(fragment_metadata_list), expected_ret_value) def test_verify_fragment_inline_chksum_succeed(self): pyeclib_drivers = self.get_pyeclib_testspec("inline_crc32") filesize = 1024 * 1024 * 3 file_str = ''.join(random.choice(string.ascii_letters) for i in range(filesize)) file_bytes = file_str.encode('utf-8') for pyeclib_driver in pyeclib_drivers: fragments = pyeclib_driver.encode(file_bytes) fragment_metadata_list = [] for fragment in fragments: fragment_metadata_list.append( pyeclib_driver.get_metadata(fragment)) expected_ret_value = {"status": 0} self.assertTrue(pyeclib_driver.verify_stripe_metadata( fragment_metadata_list) == expected_ret_value) def test_get_segment_byterange_info(self): pyeclib_drivers = self.get_pyeclib_testspec() if not pyeclib_drivers: return file_size = 1024 * 1024 segment_size = 3 * 1024 ranges = [ (0, 1), (1, 12), (10, 1000), (0, segment_size - 1), (1, segment_size + 1), (segment_size - 1, 2 * segment_size)] expected_results = {} expected_results[(0, 1)] = {0: (0, 1)} expected_results[(1, 12)] = {0: (1, 12)} expected_results[(10, 1000)] = {0: (10, 1000)} expected_results[(0, segment_size - 1)] = {0: (0, segment_size - 1)} expected_results[(1, segment_size + 1) ] = {0: (1, segment_size - 1), 1: (0, 1)} expected_results[ (segment_size - 1, 2 * segment_size)] = { 0: (segment_size - 1, segment_size - 1), 1: (0, segment_size - 1), 2: (0, 0)} results = pyeclib_drivers[0].get_segment_info_byterange( ranges, file_size, segment_size) for exp_result_key in expected_results: self.assertIn(exp_result_key, results) self.assertTrue( len(results[exp_result_key]) == len(expected_results[exp_result_key])) exp_result_map = expected_results[exp_result_key] for segment_key in exp_result_map: self.assertIn(segment_key, results[exp_result_key]) self.assertTrue( results[exp_result_key][segment_key] == expected_results[exp_result_key][segment_key]) def test_get_segment_info(self): pyeclib_drivers = self.get_pyeclib_testspec() file_sizes = [ 1024 * 1024, 2 * 1024 * 1024, 10 * 1024 * 1024, 10 * 1024 * 1024 + 7] segment_sizes = [3 * 1024, 1024 * 1024] segment_strings = {} # # Generate some test segments for each segment size. # Use 2 * segment size, because last segment may be # greater than segment_size # char_set = string.ascii_uppercase + string.digits for segment_size in segment_sizes: segment_strings[segment_size] = ''.join( random.choice(char_set) for i in range(segment_size * 2)) for pyeclib_driver in pyeclib_drivers: for file_size in file_sizes: for segment_size in segment_sizes: # # Compute the segment info # segment_info = pyeclib_driver.get_segment_info( file_size, segment_size) num_segments = segment_info['num_segments'] segment_size = segment_info['segment_size'] fragment_size = segment_info['fragment_size'] last_segment_size = segment_info['last_segment_size'] last_fragment_size = segment_info['last_fragment_size'] computed_file_size = ( (num_segments - 1) * segment_size) + last_segment_size # # Verify that the segment sizes add up # self.assertTrue(computed_file_size == file_size) encoded_fragments = pyeclib_driver.encode( (segment_strings[segment_size][ :segment_size]).encode('utf-8')) # # Verify the fragment size # self.assertTrue(fragment_size == len(encoded_fragments[0])) if last_segment_size > 0: encoded_fragments = pyeclib_driver.encode( (segment_strings[segment_size][ :last_segment_size]).encode('utf-8')) # # Verify the last fragment size, if there is one # self.assertTrue( last_fragment_size == len( encoded_fragments[0])) def test_greedy_decode_reconstruct_combination(self): # the testing spec defined at get_pyeclib_testspec() method # and if you want to test either other parameters or backends, # you can add the spec you want to test there. pyeclib_drivers = self.get_pyeclib_testspec() orig_data = os.urandom(1024 ** 2) for pyeclib_driver in pyeclib_drivers: encoded = pyeclib_driver.encode(orig_data) # make all fragment like (index, frag_data) format to feed # to combinations frags = [(i, frag) for i, frag in enumerate(encoded)] num_frags = pyeclib_driver.k + pyeclib_driver.m if pyeclib_driver.ec_type == PyECLib_EC_Types.flat_xor_hd: # flat_xord_hd is guaranteed to work with 2 or 3 failures tolerable_failures = pyeclib_driver.hd - 1 else: # ... while others can tolerate more tolerable_failures = pyeclib_driver.m for check_frags_tuples in combinations( frags, num_frags - tolerable_failures): # extract check_frags_tuples from [(index, data bytes), ...] # to [index, index, ...] and [data bytes, data bytes, ...] indexes, check_frags = zip(*check_frags_tuples) decoded = pyeclib_driver.decode(check_frags) self.assertEqual( orig_data, decoded, "assertion fail in decode %s from:%s" % (pyeclib_driver, indexes)) holes = [index for index in range(num_frags) if index not in indexes] for hole in holes: reconed = pyeclib_driver.reconstruct( check_frags, [hole])[0] self.assertEqual( frags[hole][1], reconed, "assertion fail in reconstruct %s target:%s " "from:%s" % (pyeclib_driver, hole, indexes)) def test_rs(self): pyeclib_drivers = self.get_pyeclib_testspec() for pyeclib_driver in pyeclib_drivers: for file_size in self.file_sizes: tmp_file = self.files[file_size] tmp_file.seek(0) whole_file_bytes = tmp_file.read() encode_input = whole_file_bytes orig_fragments = pyeclib_driver.encode(encode_input) for iter in range(self.num_iterations): num_missing = 2 idxs_to_remove = random.sample(range( pyeclib_driver.k + pyeclib_driver.m), num_missing) fragments = orig_fragments[:] # Reverse sort the list, so we can always # remove from the original index idxs_to_remove.sort(reverse=True) for idx in idxs_to_remove: fragments.pop(idx) # # Test decoder # decoded_string = pyeclib_driver.decode(fragments) self.assertTrue(encode_input == decoded_string) # # Test reconstructor # reconstructed_fragments = pyeclib_driver.reconstruct( fragments, idxs_to_remove) self.assertEqual(len(reconstructed_fragments), len(idxs_to_remove)) for idx, frag_data in zip(idxs_to_remove, reconstructed_fragments): self.assertEqual( frag_data, orig_fragments[idx], 'Failed to reconstruct fragment %d!' % idx) # # Test decode with integrity checks # first_fragment_to_corrupt = random.randint( 0, len(fragments)) num_to_corrupt = min(len(fragments), pyeclib_driver.m + 1) fragments_to_corrupt = [ (first_fragment_to_corrupt + i) % len(fragments) for i in range(num_to_corrupt)] if StrictVersion(LIBERASURECODE_VERSION) < \ StrictVersion('1.2.0'): # if liberasurecode is older than the version supports # fragment integrity check, skip following test continue i = 0 for fragment in fragments: if i in fragments_to_corrupt: corrupted_fragment = ( "0" * len(fragment)).encode('utf-8') fragments[i] = corrupted_fragment i += 1 self.assertRaises(ECInvalidFragmentMetadata, pyeclib_driver.decode, fragments, force_metadata_checks=True) def get_available_backend(self, k, m, ec_type, chksum_type="inline_crc32"): if ec_type[:11] == "flat_xor_hd": return ECDriver(k=k, m=m, ec_type="flat_xor_hd", chksum_type=chksum_type) elif ec_type in VALID_EC_TYPES: return ECDriver(k=k, m=m, ec_type=ec_type, chksum_type=chksum_type) else: return None def test_liberasurecode_insufficient_frags_error(self): file_size = self.file_sizes[0] tmp_file = self.files[file_size] tmp_file.seek(0) whole_file_bytes = tmp_file.read() for ec_type in ['flat_xor_hd_3', 'liberasurecode_rs_vand']: if ec_type in VALID_EC_TYPES: pyeclib_driver = self.get_available_backend( k=10, m=5, ec_type=ec_type) fragments = pyeclib_driver.encode(whole_file_bytes) self.assertRaises(ECInsufficientFragments, pyeclib_driver.reconstruct, [fragments[0]], [1, 2, 3, 4, 5, 6]) def test_min_parity_fragments_needed(self): pyeclib_drivers = [] for ec_type in ['flat_xor_hd_3', 'liberasurecode_rs_vand']: if ec_type in VALID_EC_TYPES: pyeclib_drivers.append(ECDriver(k=10, m=5, ec_type=ec_type)) self.assertTrue( pyeclib_drivers[0].min_parity_fragments_needed() == 1) def test_pyeclib_driver_repr_expression(self): pyeclib_drivers = self.get_pyeclib_testspec() for driver in pyeclib_drivers: if driver.ec_type.name == 'flat_xor_hd': name = 'flat_xor_hd_%s' % driver.hd else: name = driver.ec_type.name self.assertEqual( "ECDriver(ec_type='%s', k=%s, m=%s)" % (name, driver.k, driver.m), repr(driver)) def test_get_segment_info_memory_usage(self): for ec_driver in self.get_pyeclib_testspec(): self._test_get_segment_info_memory_usage(ec_driver) def _test_get_segment_info_memory_usage(self, ec_driver): # 1. Preapre the expected memory allocation ec_driver.get_segment_info(1024 * 1024, 1024 * 1024) loop_range = range(1000) # 2. Get current memory usage usage = resource.getrusage(resource.RUSAGE_SELF)[2] # 3. Loop to call get_segment_info for x in loop_range: ec_driver.get_segment_info(1024 * 1024, 1024 * 1024) # 4. memory usage shoudln't be increased self.assertEqual(usage, resource.getrusage(resource.RUSAGE_SELF)[2], 'Memory usage is increased unexpectedly %s - %s' % (usage, resource.getrusage(resource.RUSAGE_SELF)[2])) def test_get_metadata_memory_usage(self): for ec_driver in self.get_pyeclib_testspec(): self._test_get_metadata_memory_usage(ec_driver) def _test_get_metadata_memory_usage(self, ec_driver): # 1. Prepare the expected memory allocation encoded = ec_driver.encode(b'aaa') ec_driver.get_metadata(encoded[0], formatted=True) loop_range = range(400000) # 2. Get current memory usage baseline_usage = resource.getrusage(resource.RUSAGE_SELF)[2] # 3. Loop to call get_metadata for x in loop_range: ec_driver.get_metadata(encoded[0], formatted=True) # 4. memory usage shouldn't increase new_usage = resource.getrusage(resource.RUSAGE_SELF)[2] self.assertEqual(baseline_usage, new_usage, 'Memory usage is increased unexpectedly %s -> %s' % (baseline_usage, new_usage)) class BackendsEnabledMetaclass(type): def __new__(meta, cls_name, cls_bases, cls_dict): for ec_type in ALL_EC_TYPES: def dummy(self, ec_type=ec_type): if ec_type not in VALID_EC_TYPES: raise unittest.SkipTest if ec_type == 'shss': k = 10 m = 4 elif ec_type == 'libphazr': k = 4 m = 4 else: k = 10 m = 5 ECDriver(k=k, m=m, ec_type=ec_type) dummy.__name__ = 'test_%s_available' % ec_type cls_dict[dummy.__name__] = dummy return type.__new__(meta, cls_name, cls_bases, cls_dict) class TestBackendsEnabled(six.with_metaclass(BackendsEnabledMetaclass, unittest.TestCase)): ''' Based on TestPyECLibDriver.test_valid_algo above, but these tests should *always* either pass or skip. ''' if __name__ == '__main__': unittest.main()