quay/test/test_morecollections.py

import unittest

from util.morecollections import (FastIndexList, StreamingDiffTracker,
                                  IndexedStreamingDiffTracker)

class FastIndexListTests(unittest.TestCase):
  def test_basic_usage(self):
    indexlist = FastIndexList()

    # Add 1
    indexlist.add(1)
    self.assertEquals([1], indexlist.values())
    self.assertEquals(0, indexlist.index(1))

    # Add 2
    indexlist.add(2)
    self.assertEquals([1, 2], indexlist.values())
    self.assertEquals(0, indexlist.index(1))
    self.assertEquals(1, indexlist.index(2))

    # Pop nothing.
    indexlist.pop_until(-1)
    self.assertEquals([1, 2], indexlist.values())
    self.assertEquals(0, indexlist.index(1))
    self.assertEquals(1, indexlist.index(2))

    # Pop 1.
    self.assertEquals([1], indexlist.pop_until(0))
    self.assertEquals([2], indexlist.values())
    self.assertIsNone(indexlist.index(1))
    self.assertEquals(0, indexlist.index(2))

    # Add 3.
    indexlist.add(3)
    self.assertEquals([2, 3], indexlist.values())
    self.assertEquals(0, indexlist.index(2))
    self.assertEquals(1, indexlist.index(3))

    # Pop 2, 3.
    self.assertEquals([2, 3], indexlist.pop_until(1))
    self.assertEquals([], indexlist.values())
    self.assertIsNone(indexlist.index(1))
    self.assertIsNone(indexlist.index(2))
    self.assertIsNone(indexlist.index(3))

  def test_popping(self):
    indexlist = FastIndexList()
    indexlist.add('hello')
    indexlist.add('world')
    indexlist.add('you')
    indexlist.add('rock')

    self.assertEquals(0, indexlist.index('hello'))
    self.assertEquals(1, indexlist.index('world'))
    self.assertEquals(2, indexlist.index('you'))
    self.assertEquals(3, indexlist.index('rock'))

    indexlist.pop_until(1)
    self.assertEquals(0, indexlist.index('you'))
    self.assertEquals(1, indexlist.index('rock'))


class IndexedStreamingDiffTrackerTests(unittest.TestCase):
  def test_basic(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 3)
    tracker.push_new([('a', 0), ('b', 1), ('c', 2)])
    tracker.push_old([('b', 1)])
    tracker.done()

    self.assertEquals(['a', 'c'], added)

  def test_multiple_done(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 3)
    tracker.push_new([('a', 0), ('b', 1), ('c', 2)])
    tracker.push_old([('b', 1)])
    tracker.done()
    tracker.done()

    self.assertEquals(['a', 'c'], added)

  def test_same_streams(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 3)
    tracker.push_new([('a', 0), ('b', 1), ('c', 2)])
    tracker.push_old([('a', 0), ('b', 1), ('c', 2)])
    tracker.done()

    self.assertEquals([], added)

  def test_only_new(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 3)
    tracker.push_new([('a', 0), ('b', 1), ('c', 2)])
    tracker.push_old([])
    tracker.done()

    self.assertEquals(['a', 'b', 'c'], added)

  def test_pagination(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 2)
    tracker.push_new([('a', 0), ('b', 1)])
    tracker.push_old([])

    tracker.push_new([('c', 2)])
    tracker.push_old([])

    tracker.done()

    self.assertEquals(['a', 'b', 'c'], added)

  def test_old_pagination_no_repeat(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 2)
    tracker.push_new([('new1', 3), ('new2', 4)])
    tracker.push_old([('old1', 1), ('old2', 2)])

    tracker.push_new([])
    tracker.push_old([('new1', 3)])

    tracker.done()

    self.assertEquals(['new2'], added)

  def test_old_pagination(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 2)
    tracker.push_new([('a', 10), ('b', 11)])
    tracker.push_old([('z', 1), ('y', 2)])

    tracker.push_new([('c', 12)])
    tracker.push_old([('a', 10)])

    tracker.done()

    self.assertEquals(['b', 'c'], added)

  def test_very_offset(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 2)
    tracker.push_new([('a', 10), ('b', 11)])
    tracker.push_old([('z', 1), ('y', 2)])

    tracker.push_new([('c', 12), ('d', 13)])
    tracker.push_old([('x', 3), ('w', 4)])

    tracker.push_new([('e', 14)])
    tracker.push_old([('a', 10), ('d', 13)])

    tracker.done()

    self.assertEquals(['b', 'c', 'e'], added)

  def test_many_old(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 2)
    tracker.push_new([('z', 26), ('hello', 100)])
    tracker.push_old([('a', 1), ('b', 2)])

    tracker.push_new([])
    tracker.push_old([('c', 1), ('d', 2)])

    tracker.push_new([])
    tracker.push_old([('e', 3), ('f', 4)])

    tracker.push_new([])
    tracker.push_old([('g', 5), ('z', 26)])

    tracker.done()

    self.assertEquals(['hello'], added)

  def test_high_old_bound(self):
    added = []

    tracker = IndexedStreamingDiffTracker(added.append, 2)
    tracker.push_new([('z', 26), ('hello', 100)])
    tracker.push_old([('end1', 999), ('end2', 1000)])

    tracker.push_new([])
    tracker.push_old([])

    tracker.done()

    self.assertEquals(['z', 'hello'], added)


class StreamingDiffTrackerTests(unittest.TestCase):
  def test_basic(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 3)
    tracker.push_new(['a', 'b', 'c'])
    tracker.push_old(['b'])
    tracker.done()

    self.assertEquals(['a', 'c'], added)

  def test_same_streams(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 3)
    tracker.push_new(['a', 'b', 'c'])
    tracker.push_old(['a', 'b', 'c'])
    tracker.done()

    self.assertEquals([], added)

  def test_some_new(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 5)
    tracker.push_new(['a', 'b', 'c', 'd', 'e'])
    tracker.push_old(['a', 'b', 'c'])
    tracker.done()

    self.assertEquals(['d', 'e'], added)

  def test_offset_new(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 5)
    tracker.push_new(['b', 'c', 'd', 'e'])
    tracker.push_old(['a', 'b', 'c'])
    tracker.done()

    self.assertEquals(['d', 'e'], added)

  def test_multiple_calls(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 3)
    tracker.push_new(['a', 'b', 'c'])
    tracker.push_old(['b', 'd', 'e'])

    tracker.push_new(['f', 'g', 'h'])
    tracker.push_old(['g', 'h'])
    tracker.done()

    self.assertEquals(['a', 'c', 'f'], added)

  def test_empty_old(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 3)
    tracker.push_new(['a', 'b', 'c'])
    tracker.push_old([])

    tracker.push_new(['f', 'g', 'h'])
    tracker.push_old([])
    tracker.done()

    self.assertEquals(['a', 'b', 'c', 'f', 'g', 'h'], added)

  def test_more_old(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 2)
    tracker.push_new(['c', 'd'])
    tracker.push_old(['a', 'b'])

    tracker.push_new([])
    tracker.push_old(['c'])
    tracker.done()

    self.assertEquals(['d'], added)

  def test_more_new(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 4)
    tracker.push_new(['a', 'b', 'c', 'd'])
    tracker.push_old(['r'])

    tracker.push_new(['e', 'f', 'r', 'z'])
    tracker.push_old([])
    tracker.done()

    self.assertEquals(['a', 'b', 'c', 'd', 'e', 'f', 'z'], added)

  def test_more_new2(self):
    added = []

    tracker = StreamingDiffTracker(added.append, 4)
    tracker.push_new(['a', 'b', 'c', 'd'])
    tracker.push_old(['r'])

    tracker.push_new(['e', 'f', 'g', 'h'])
    tracker.push_old([])

    tracker.push_new(['i', 'j', 'r', 'z'])
    tracker.push_old([])
    tracker.done()

    self.assertEquals(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'z'], added)


if __name__ == '__main__':
  unittest.main()