Implemented the test cases and Pyhon classes for the basic functionality.

Author: facundobatista

Lib/heapq.py

@@ -128,7 +128,8 @@
 
 __all__ = ['heappush', 'heappop', 'heapify', 'heapreplace', 'heappushpop',
            'heappush_max', 'heappop_max', 'heapify_max', 'heapreplace_max',
-           'heappushpop_max', 'nlargest', 'nsmallest', 'merge']
+           'heappushpop_max', 'nlargest', 'nsmallest', 'merge',
+           'MinHeap', 'MaxHeap']
 
 def heappush(heap, item):
     """Push item onto heap, maintaining the heap invariant."""
@@ -592,6 +593,127 @@ def nlargest(n, iterable, key=None):
     result.sort(reverse=True)
     return [elem for (k, order, elem) in result]
 
+
+class MinHeap:
+    """Min-heap: a priority queue that serves its smallest item first.
+
+    Only the ``<`` operator is used to compare items.
+
+    If *iterable* is given, the heap is initialized from its items in linear
+    time; otherwise it starts empty.
+
+    >>> h = MinHeap([3, 1, 2])
+    >>> h.push(0)
+    >>> [h.pop() for _ in range(len(h))]
+    [0, 1, 2, 3]
+    """
+
+    def __init__(self, iterable=None):
+        if iterable is None:
+            self._queue = []
+        else:
+            self._queue = list(iterable)
+            heapify(self._queue)
+
+    def push(self, item):
+        """Push item onto the heap, maintaining the heap invariant."""
+        heappush(self._queue, item)
+
+    def pop(self):
+        """Pop and return the smallest item, maintaining the heap invariant.
+
+        Raises IndexError if the heap is empty.
+        """
+        return heappop(self._queue)
+
+    def pushpop(self, item):
+        """Push item on the heap, then pop and return the smallest item.
+
+        The combined action runs more efficiently than a push() followed by
+        a separate call to pop().
+        """
+        return heappushpop(self._queue, item)
+
+    def replace(self, item):
+        """Pop and return the smallest item, and also push the new item.
+
+        The heap size doesn't change.  Raises IndexError if the heap is
+        empty.  This is more efficient than a pop() followed by a push(),
+        and can be more appropriate when using a fixed-size heap.  Note that
+        the value returned may be larger than item.
+        """
+        return heapreplace(self._queue, item)
+
+    def __len__(self):
+        return len(self._queue)
+
+    def __bool__(self):
+        return bool(self._queue)
+
+    def __repr__(self):
+        return f'{type(self).__name__}({self._queue!r})'
+
+
+class MaxHeap:
+    """Max-heap: a priority queue that serves its largest item first.
+
+    Only the ``<`` operator is used to compare items.
+
+    If *iterable* is given, the heap is initialized from its items in linear
+    time; otherwise it starts empty.
+
+    >>> h = MaxHeap([1, 3, 2])
+    >>> h.push(4)
+    >>> [h.pop() for _ in range(len(h))]
+    [4, 3, 2, 1]
+    """
+
+    def __init__(self, iterable=None):
+        if iterable is None:
+            self._queue = []
+        else:
+            self._queue = list(iterable)
+            heapify_max(self._queue)
+
+    def push(self, item):
+        """Push item onto the heap, maintaining the heap invariant."""
+        heappush_max(self._queue, item)
+
+    def pop(self):
+        """Pop and return the largest item, maintaining the heap invariant.
+
+        Raises IndexError if the heap is empty.
+        """
+        return heappop_max(self._queue)
+
+    def pushpop(self, item):
+        """Push item on the heap, then pop and return the largest item.
+
+        The combined action runs more efficiently than a push() followed by
+        a separate call to pop().
+        """
+        return heappushpop_max(self._queue, item)
+
+    def replace(self, item):
+        """Pop and return the largest item, and also push the new item.
+
+        The heap size doesn't change.  Raises IndexError if the heap is
+        empty.  This is more efficient than a pop() followed by a push(),
+        and can be more appropriate when using a fixed-size heap.  Note that
+        the value returned may be smaller than item.
+        """
+        return heapreplace_max(self._queue, item)
+
+    def __len__(self):
+        return len(self._queue)
+
+    def __bool__(self):
+        return bool(self._queue)
+
+    def __repr__(self):
+        return f'{type(self).__name__}({self._queue!r})'
+
+
 # If available, use C implementation
 try:
     from _heapq import *

Lib/test/test_heapq.py

@@ -5,7 +5,7 @@
 import doctest
 
 from test.support import import_helper
-from unittest import TestCase, skipUnless
+from unittest import TestCase, mock, skipUnless
 from operator import itemgetter
 
 py_heapq = import_helper.import_fresh_module('heapq', blocked=['_heapq'])
@@ -637,5 +637,178 @@ class TestErrorHandlingC(TestErrorHandling, TestCase):
     module = c_heapq
 
 
+#==============================================================================
+# Object-oriented interface: MinHeap and MaxHeap.
+#
+# The classes add no new heap algorithm; they are thin wrappers that delegate
+# to the module-level functions.  They are tested in two ways:
+#
+# * Functional "sanity check" tests, run against both the pure-Python and the
+#   C classes (like the rest of this file).  They are end-to-end, so a
+#   mis-wired method (say, a MaxHeap that pushed with the min-heap routine)
+#   changes the observable output and is caught even for the C class.
+#
+# * Delegation tests that patch the module-level functions and check that each
+#   method forwards to the right one.  These only apply to the pure-Python
+#   class, since the C class calls the underlying routines directly and cannot
+#   be intercepted; its wiring is covered by the functional tests instead.
+
+
+class HeapClassSanityTest:
+    """Functional sanity checks shared by the MinHeap/MaxHeap test cases.
+
+    Subclasses set:
+
+    * *cls* -- the heap class under test (e.g. ``py_heapq.MinHeap``)
+    * *ordered* -- a small list already satisfying the heap invariant
+    * *reverse* -- ``True`` for max-heaps (the order ``pop()`` yields items in)
+    """
+
+    def in_heap_order(self, data):
+        return sorted(data, reverse=self.reverse)
+
+    def test_construct_empty(self):
+        h = self.cls()
+        self.assertEqual(len(h), 0)
+        self.assertFalse(h)
+
+    def test_construct_from_iterable(self):
+        data = [random.randrange(1000) for _ in range(100)]
+        # Accept any iterable, not just a sequence.
+        h = self.cls(iter(data))
+        self.assertEqual(len(h), len(data))
+        drained = [h.pop() for _ in range(len(data))]
+        self.assertEqual(drained, self.in_heap_order(data))
+
+    def test_push_then_pop_sorts(self):
+        data = [random.randrange(1000) for _ in range(100)]
+        h = self.cls()
+        for item in data:
+            self.assertIsNone(h.push(item))
+        self.assertEqual(len(h), len(data))
+        drained = [h.pop() for _ in range(len(data))]
+        self.assertEqual(drained, self.in_heap_order(data))
+
+    def test_len_and_bool(self):
+        h = self.cls()
+        self.assertEqual(len(h), 0)
+        self.assertFalse(h)
+        h.push(1)
+        self.assertEqual(len(h), 1)
+        self.assertTrue(h)
+
+    def test_pop_empty_raises(self):
+        h = self.cls()
+        self.assertRaises(IndexError, h.pop)
+
+    def test_pushpop(self):
+        h = self.cls(self.ordered)
+        result = h.pushpop(2)
+        expected = self.in_heap_order(list(self.ordered) + [2])[0]
+        self.assertEqual(result, expected)
+        self.assertEqual(len(h), len(self.ordered))
+
+    def test_replace(self):
+        h = self.cls(self.ordered)
+        expected = self.in_heap_order(self.ordered)[0]
+        new_item = -1000 if self.reverse else 1000
+        self.assertEqual(h.replace(new_item), expected)
+        self.assertEqual(len(h), len(self.ordered))
+
+    def test_repr(self):
+        h = self.cls(self.ordered)
+        self.assertEqual(repr(h), f'{self.cls.__name__}({self.ordered!r})')
+
+
+class TestMinHeapPython(HeapClassSanityTest, TestCase):
+    cls = py_heapq.MinHeap
+    ordered = [1, 2, 3]
+    reverse = False
+
+
+@skipUnless(c_heapq, 'requires _heapq')
+class TestMinHeapC(HeapClassSanityTest, TestCase):
+    cls = c_heapq.MinHeap if c_heapq else None
+    ordered = [1, 2, 3]
+    reverse = False
+
+
+class TestMaxHeapPython(HeapClassSanityTest, TestCase):
+    cls = py_heapq.MaxHeap
+    ordered = [3, 2, 1]
+    reverse = True
+
+
+@skipUnless(c_heapq, 'requires _heapq')
+class TestMaxHeapC(HeapClassSanityTest, TestCase):
+    cls = c_heapq.MaxHeap if c_heapq else None
+    ordered = [3, 2, 1]
+    reverse = True
+
+
+class HeapClassDelegationTest:
+    """Check that each method forwards to the right module-level function.
+
+    Delegation only makes sense for the pure-Python class, so these tests
+    always patch ``py_heapq``.
+
+    Subclasses set *cls* and the names of the functions each method is
+    expected to forward to (*heapify_func*, *push_func*, *pop_func*,
+    *pushpop_func* and *replace_func*).
+    """
+
+    def test_init_delegates_to_heapify(self):
+        with mock.patch.object(py_heapq, self.heapify_func) as m:
+            h = self.cls([3, 1, 2])
+        m.assert_called_once_with(h._queue)
+
+    def test_init_empty_does_not_heapify(self):
+        with mock.patch.object(py_heapq, self.heapify_func) as m:
+            self.cls()
+        m.assert_not_called()
+
+    def test_push_delegates(self):
+        h = self.cls()
+        with mock.patch.object(py_heapq, self.push_func) as m:
+            h.push(42)
+        m.assert_called_once_with(h._queue, 42)
+
+    def test_pop_delegates(self):
+        h = self.cls([1, 2, 3])
+        with mock.patch.object(py_heapq, self.pop_func) as m:
+            h.pop()
+        m.assert_called_once_with(h._queue)
+
+    def test_pushpop_delegates(self):
+        h = self.cls([1, 2, 3])
+        with mock.patch.object(py_heapq, self.pushpop_func) as m:
+            h.pushpop(42)
+        m.assert_called_once_with(h._queue, 42)
+
+    def test_replace_delegates(self):
+        h = self.cls([1, 2, 3])
+        with mock.patch.object(py_heapq, self.replace_func) as m:
+            h.replace(42)
+        m.assert_called_once_with(h._queue, 42)
+
+
+class TestMinHeapDelegation(HeapClassDelegationTest, TestCase):
+    cls = py_heapq.MinHeap
+    heapify_func = 'heapify'
+    push_func = 'heappush'
+    pop_func = 'heappop'
+    pushpop_func = 'heappushpop'
+    replace_func = 'heapreplace'
+
+
+class TestMaxHeapDelegation(HeapClassDelegationTest, TestCase):
+    cls = py_heapq.MaxHeap
+    heapify_func = 'heapify_max'
+    push_func = 'heappush_max'
+    pop_func = 'heappop_max'
+    pushpop_func = 'heappushpop_max'
+    replace_func = 'heapreplace_max'
+
+
 if __name__ == "__main__":
     unittest.main()