import unittest import sys import relabeling from sortedcontainers import SortedListWithKey from six.moves import zip as izip, xrange # Shortcut to keep code more concise. r = relabeling def skipfloats(x, n): for i in xrange(n): x = relabeling.nextfloat(x) return x class Item(object): """ Tests use Item for items of the sorted lists we maintain. """ def __init__(self, value, key): self.value = value self.key = key def __repr__(self): return "Item(v=%s,k=%s)" % (self.value, self.key) class ItemList(object): def __init__(self, val_key_pairs): self._slist = SortedListWithKey(key=lambda item: item.key) self._slist.update(Item(v, k) for (v, k) in val_key_pairs) self.num_update_events = 0 self.num_updated_keys = 0 def get_values(self): return [item.value for item in self._slist] def get_list(self): return self._slist def find_value(self, value): return next((item for item in self._slist if item.value == value), None) def avg_updated_keys(self): return float(self.num_updated_keys) / len(self._slist) def next(self, item): return self._slist[self._slist.index(item) + 1] def prev(self, item): return self._slist[self._slist.index(item) - 1] def insert_items(self, val_key_pairs, prepare_inserts=r.prepare_inserts): keys = [k for (v, k) in val_key_pairs] adjustments, new_keys = prepare_inserts(self._slist, keys) if adjustments: self.num_update_events += 1 self.num_updated_keys += len(adjustments) # Updating items is a bit tricky: we have to do it without violating order (just changing # key of an existing item easily might), so we remove items first. And we can only rely on # indices if we scan items in a backwards order. items = [self._slist.pop(index) for (index, key) in reversed(adjustments)] items.reverse() for (index, key), item in izip(adjustments, items): item.key = key self._slist.update(items) # Now add the new items. self._slist.update(Item(val, new_key) for (val, _), new_key in izip(val_key_pairs, new_keys)) # For testing, pass along the return value from prepare_inserts. return adjustments, new_keys class TestRelabeling(unittest.TestCase): def test_nextfloat(self): def verify_nextfloat(x): nx = r.nextfloat(x) self.assertNotEqual(nx, x) self.assertGreater(nx, x) self.assertEqual(r.prevfloat(nx), x) average = (nx + x) / 2 self.assertTrue(average == nx or average == x) verify_nextfloat(1) verify_nextfloat(-1) verify_nextfloat(417) verify_nextfloat(-417) verify_nextfloat(12312422) verify_nextfloat(-12312422) verify_nextfloat(0.1234) verify_nextfloat(-0.1234) verify_nextfloat(0.00005) verify_nextfloat(-0.00005) verify_nextfloat(0.0) verify_nextfloat(r.nextfloat(0.0)) verify_nextfloat(sys.float_info.min) verify_nextfloat(-sys.float_info.min) def test_prevfloat(self): def verify_prevfloat(x): nx = r.prevfloat(x) self.assertNotEqual(nx, x) self.assertLess(nx, x) self.assertEqual(r.nextfloat(nx), x) average = (nx + x) / 2 self.assertTrue(average == nx or average == x) verify_prevfloat(1) verify_prevfloat(-1) verify_prevfloat(417) verify_prevfloat(-417) verify_prevfloat(12312422) verify_prevfloat(-12312422) verify_prevfloat(0.1234) verify_prevfloat(-0.1234) verify_prevfloat(0.00005) verify_prevfloat(-0.00005) verify_prevfloat(r.nextfloat(0.0)) verify_prevfloat(sys.float_info.min) verify_prevfloat(-sys.float_info.min) def test_range_around_float(self): def verify_range(bits, begin, end): self.assertEqual(r.range_around_float(begin, bits), (begin, end)) self.assertEqual(r.range_around_float((end + begin) / 2, bits), (begin, end)) delta = r.nextfloat(begin) - begin if begin + delta < end: self.assertEqual(r.range_around_float(begin + delta, bits), (begin, end)) if end - delta >= begin: self.assertEqual(r.range_around_float(end - delta, bits), (begin, end)) def verify_small_range_at(begin): verify_range(0, begin, skipfloats(begin, 1)) verify_range(1, begin, skipfloats(begin, 2)) verify_range(4, begin, skipfloats(begin, 16)) verify_range(10, begin, skipfloats(begin, 1024)) verify_small_range_at(1.0) verify_small_range_at(0.5) verify_small_range_at(0.25) verify_small_range_at(0.75) verify_small_range_at(17.0) verify_range(52, 1.0, 2.0) self.assertEqual(r.range_around_float(1.4, 52), (1.0, 2.0)) verify_range(52, 0.5, 1.0) self.assertEqual(r.range_around_float(0.75, 52), (0.5, 1.0)) self.assertEqual(r.range_around_float(17, 48), (17.0, 18.0)) self.assertEqual(r.range_around_float(17, 49), (16.0, 18.0)) self.assertEqual(r.range_around_float(17, 50), (16.0, 20.0)) self.assertEqual(r.range_around_float(17, 51), (16.0, 24.0)) self.assertEqual(r.range_around_float(17, 52), (16.0, 32.0)) verify_range(51, 0.25, 0.375) self.assertEqual(r.range_around_float(0.27, 51), (0.25, 0.375)) self.assertEqual(r.range_around_float(0.30, 51), (0.25, 0.375)) self.assertEqual(r.range_around_float(0.37, 51), (0.25, 0.375)) verify_range(51, 0.50, 0.75) verify_range(51, 0.75, 1.0) verify_range(52, 0.25, 0.5) # Range around 0 isn't quite right, and possibly can't be. But we test that it's at least # something meaningful. self.assertEqual(r.range_around_float(0.00, 52), (0.00, 0.5)) self.assertEqual(r.range_around_float(0.25, 52), (0.25, 0.5)) self.assertEqual(r.range_around_float(0.00, 50), (0.00, 0.125)) self.assertEqual(r.range_around_float(0.10, 50), (0.09375, 0.109375)) self.assertEqual(r.range_around_float(0.0, 53), (0.00, 1)) self.assertEqual(r.range_around_float(0.5, 53), (0.00, 1)) self.assertEqual(r.range_around_float(0, 0), (0.0, skipfloats(0.5, 1) - 0.5)) self.assertEqual(r.range_around_float(0, 1), (0.0, skipfloats(0.5, 2) - 0.5)) self.assertEqual(r.range_around_float(0, 4), (0.0, skipfloats(0.5, 16) - 0.5)) self.assertEqual(r.range_around_float(0, 10), (0.0, skipfloats(0.5, 1024) - 0.5)) def test_all_distinct(self): # Just like r.get_range, but includes endpoints. def full_range(start, end, count): return [start] + r.get_range(start, end, count) + [end] self.assertTrue(r.all_distinct(range(1000))) self.assertTrue(r.all_distinct([])) self.assertTrue(r.all_distinct([1.0])) self.assertFalse(r.all_distinct([1.0, 1.0])) self.assertTrue(r.all_distinct(full_range(0, 1, 1000))) self.assertFalse(r.all_distinct(full_range(1.0, r.nextfloat(1.0), 1))) self.assertFalse(r.all_distinct(full_range(1.0, skipfloats(1.0, 10), 10))) self.assertTrue(r.all_distinct(full_range(1.0, skipfloats(1.0, 11), 10))) self.assertTrue(r.all_distinct(full_range(0.1, skipfloats(0.1, 100), 99))) self.assertFalse(r.all_distinct(full_range(0.1, skipfloats(0.1, 100), 100))) def test_get_range(self): self.assertEqual(r.get_range(0.0, 2.0, 3), [0.5, 1, 1.5]) self.assertEqual(r.get_range(1, 17, 7), [3,5,7,9,11,13,15]) self.assertEqual(r.get_range(-1, 1.5, 4), [-0.5, 0, 0.5, 1]) def test_prepare_inserts_simple(self): slist = SortedListWithKey(key=lambda i: i.key) self.assertEqual(r.prepare_inserts(slist, [4.0]), ([], [1.0])) self.assertEqual(r.prepare_inserts(slist, [0.0]), ([], [1.0])) self.assertEqual(r.prepare_inserts(slist, [4.0, 4.0, 5, 6]), ([], [1.0, 2.0, 3.0, 4.0])) self.assertEqual(r.prepare_inserts(slist, [4, 5, 6, 5, 4]), ([], [1,3,5,4,2])) slist.update(Item(v, k) for (v, k) in zip(['a','b','c'], [3.0, 4.0, 5.0])) self.assertEqual(r.prepare_inserts(slist, [0.0]), ([], [1.5])) values = 'defgijkl' to_update, to_add = r.prepare_inserts(slist, [3,3,4,5,6,4,6,4]) self.assertEqual(to_add, [1., 2., 3.25, 4.5, 6., 3.5, 7., 3.75]) self.assertEqual(to_update, []) slist.update(Item(v, k) for (v, k) in zip(values, to_add)) self.assertEqual([i.value for i in slist], list('deafjlbgcik')) def test_with_invalid(self): slist = SortedListWithKey(key=lambda i: i.key) slist.add(Item('a', 0)) self.assertEqual(r.prepare_inserts(slist, [0.0]), ([(0, 2.0)], [1.0])) self.assertEqual(r.prepare_inserts(slist, [1.0]), ([], [1.0])) slist = SortedListWithKey(key=lambda i: i.key) slist.update(Item(v, k) for (v, k) in zip('abcdef', [0, 0, 0, 1, 1, 1])) # We expect the whole range to be renumbered. self.assertEqual(r.prepare_inserts(slist, [0.0, 0.0]), ([(0, 3.0), (1, 4.0), (2, 5.0), (3, 6.0), (4, 7.0), (5, 8.0)], [1.0, 2.0])) # We also expect a renumbering if there are negative or infinite values. slist = SortedListWithKey(key=lambda i: i.key) slist.add(Item('a', float('inf'))) self.assertEqual(r.prepare_inserts(slist, [0.0]), ([(0, 2.0)], [1.0])) self.assertEqual(r.prepare_inserts(slist, [float('inf')]), ([(0, 2.0)], [1.0])) slist = SortedListWithKey(key=lambda i: i.key) slist.add(Item('a', -17.0)) self.assertEqual(r.prepare_inserts(slist, [0.0]), ([(0, 1.0)], [2.0])) self.assertEqual(r.prepare_inserts(slist, [float('-inf')]), ([(0, 2.0)], [1.0])) def test_with_dups(self): slist = SortedListWithKey(key=lambda i: i.key) slist.update(Item(v, k) for (v, k) in zip('abcdef', [1, 1, 1, 2, 2, 2])) self.assertEqual(r.prepare_inserts(slist, [0.0]), ([], [0.5])) def test_renumber_endpoints1(self): self._do_test_renumber_ends([]) def test_renumber_endpoints2(self): self._do_test_renumber_ends(list(zip("abcd", [40,50,60,70]))) def _do_test_renumber_ends(self, initial): # Test insertions that happen together on the left and on the right. slist = ItemList(initial) for i in xrange(2000): slist.insert_items([(i, float('-inf')), (-i, float('inf'))]) self.assertEqual(slist.get_values(), rev_range(2000) + [v for v,k in initial] + list(xrange(0, -2000, -1))) #print slist.num_update_events, slist.num_updated_keys self.assertLess(slist.avg_updated_keys(), 3) self.assertLess(slist.num_update_events, 80) def test_renumber_left(self): slist = ItemList(zip("abcd", [4,5,6,7])) ins_item = slist.find_value('c') for i in xrange(1000): slist.insert_items([(i, ins_item.key)]) # Check the end result self.assertEqual(slist.get_values(), ['a', 'b'] + list(xrange(1000)) + ['c', 'd']) self.assertAlmostEqual(slist.avg_updated_keys(), 3.5, delta=1) self.assertLess(slist.num_update_events, 40) def test_renumber_right(self): slist = ItemList(zip("abcd", [4,5,6,7])) ins_item = slist.find_value('b') for i in xrange(1000): slist.insert_items([(i, r.nextfloat(ins_item.key))]) # Check the end result self.assertEqual(slist.get_values(), ['a', 'b'] + rev_range(1000) + ['c', 'd']) self.assertAlmostEqual(slist.avg_updated_keys(), 3.5, delta=1) self.assertLess(slist.num_update_events, 40) def test_renumber_left_dumb(self): # Here we use the "dumb" approach, and see that in our test case it's significantly worse. # (The badness increases with the number of insertions, but we'll keep numbers small to keep # the test fast.) slist = ItemList(zip("abcd", [4,5,6,7])) ins_item = slist.find_value('c') for i in xrange(1000): slist.insert_items([(i, ins_item.key)], prepare_inserts=r.prepare_inserts_dumb) self.assertEqual(slist.get_values(), ['a', 'b'] + list(xrange(1000)) + ['c', 'd']) self.assertGreater(slist.avg_updated_keys(), 8) def test_renumber_right_dumb(self): slist = ItemList(zip("abcd", [4,5,6,7])) ins_item = slist.find_value('b') for i in xrange(1000): slist.insert_items([(i, r.nextfloat(ins_item.key))], prepare_inserts=r.prepare_inserts_dumb) self.assertEqual(slist.get_values(), ['a', 'b'] + rev_range(1000) + ['c', 'd']) self.assertGreater(slist.avg_updated_keys(), 8) def test_renumber_multiple(self): # In this test, we make multiple difficult insertions at each step: to the left and to the # right of each value. This should involve some adjustments that get affected by subsequent # adjustments during the same prepare_inserts() call. slist = ItemList(zip("abcd", [4,5,6,7])) # We insert items on either side of each of the original items (a, b, c, d). ins_items = list(slist.get_list()) N = 250 for i in xrange(N): slist.insert_items([("%sr%s" % (x.value, i), r.nextfloat(x.key)) for x in ins_items] + [("%sl%s" % (x.value, i), x.key) for x in ins_items] + # After the first insertion, also insert items next on either side of the # neighbors of the original a, b, c, d items. ([("%sR%s" % (x.value, i), r.nextfloat(slist.next(x).key)) for x in ins_items] + [("%sL%s" % (x.value, i), slist.prev(x).key) for x in ins_items] if i > 0 else [])) # The list should grow like this: # a, b, c, d # al0, a, ar0, ... (same for b, c, d) # aL1, al0, al1, a, ar1, ar0, aR1, ... # aL1, al0, aL2, al1, al2, a, ar2, ar1, aR2, ar0, aR1, ... def left_half(val): half = list(xrange(2*N - 1)) half[0::2] = ['%sL%d' % (val, i) for i in xrange(1, N + 1)] half[1::2] = ['%sl%d' % (val, i) for i in xrange(0, N - 1)] half[-1] = '%sl%d' % (val, N - 1) return half def right_half(val): # Best described as the reverse of left_half return [v.replace('l', 'r').replace('L', 'R') for v in reversed(left_half(val))] # The list we expect to see is of the form [aL1, al1, aL2, al2, ... aL1000, al1000, a, # ar1000, aR1000, ..., aR1], # followed by the same sequence for b, c, and d. self.assertEqual(slist.get_values(), sum([left_half(v) + [v] + right_half(v) for v in ('a', 'b', 'c', 'd')], [])) self.assertAlmostEqual(slist.avg_updated_keys(), 2.5, delta=1) self.assertLess(slist.num_update_events, 40) def rev_range(n): return list(reversed(list(xrange(n)))) if __name__ == "__main__": unittest.main()