Reworked PriorityQueue and Added Tests (#1025)

* Reworked PriorityQueue spec

Modified:
- Priority Queue methods:
    queue[elem] now returns the first value of elem stored in queue
    elem in queue now correctly returns whether a copy of element is present regardless of the function value. Apparently the bug was introduced while trying to meet heapq spec
    del queue[elem] deletes the first instance of elem in queue correctly
- Algorithms
    Same change in best_first_graph_search in romania_problem.py and search.py to make them compatible with the new spec
- Tests
    Introduced 3 tests in test_utils.py to comprehensively test PriorityQueue's new spec

* Reworked PriorityQueue spec

Modified:
- Priority Queue methods:
    queue[elem] now returns the first value of elem stored in queue
    elem in queue now correctly returns whether a copy of element is present regardless of the function value. Apparently the bug was introduced while trying to meet heapq spec
    del queue[elem] deletes the first instance of elem in queue correctly
- Algorithms
    Same change in best_first_graph_search in romania_problem.py and search.py to make them compatible with the new spec
- Tests
    Introduced 3 tests in test_utils.py to comprehensively test PriorityQueue's new spec

Author: rajatjain1997

.gitignore

@@ -44,6 +44,7 @@ nosetests.xml
 coverage.xml
 *,cover
 .hypothesis/
+*.pytest_cache
 
 # Translations
 *.mo

gui/romania_problem.py

@@ -538,9 +538,8 @@ def best_first_graph_search(problem, f):
             if child.state not in explored and child not in frontier:
                 frontier.append(child)
             elif child in frontier:
-                incumbent = frontier[child]
-                if f(child) < f(incumbent):
-                    del frontier[incumbent]
+                if f(child) < frontier[child]:
+                    del frontier[child]
                     frontier.append(child)
         display_frontier(frontier)
     if counter % 3 == 2 and counter >= 0:

search.py

@@ -275,9 +275,8 @@ def best_first_graph_search(problem, f):
             if child.state not in explored and child not in frontier:
                 frontier.append(child)
             elif child in frontier:
-                incumbent = frontier[child]
-                if f(child) < f(incumbent):
-                    del frontier[incumbent]
+                if f(child) < frontier[child]:
+                    del frontier[child]
                     frontier.append(child)
     return None
 

tests/.pytest_cache/v/cache/lastfailed

@@ -273,6 +273,43 @@ def test_expr():
     assert (expr('GP(x, z) <== P(x, y) & P(y, z)')
             == Expr('<==', GP(x, z), P(x, y) & P(y, z)))
 
+def test_min_priorityqueue():
+    queue = PriorityQueue(f=lambda x: x[1])
+    queue.append((1,100))
+    queue.append((2,30))
+    queue.append((3,50))
+    assert queue.pop() == (2,30)
+    assert len(queue) == 2
+    assert queue[(3,50)] == 50
+    assert (1,100) in queue
+    del queue[(1,100)]
+    assert (1,100) not in queue
+    queue.extend([(1,100), (4,10)])
+    assert queue.pop() == (4,10)
+    assert len(queue) == 2
+
+def test_max_priorityqueue():
+    queue = PriorityQueue(order='max', f=lambda x: x[1])
+    queue.append((1,100))
+    queue.append((2,30))
+    queue.append((3,50))
+    assert queue.pop() == (1,100)
+
+def test_priorityqueue_with_objects():
+    class Test:
+        def __init__(self, a, b):
+            self.a = a
+            self.b = b
+        def __eq__(self, other):
+            return self.a==other.a
+
+    queue = PriorityQueue(f=lambda x: x.b)
+    queue.append(Test(1,100))
+    other = Test(1,10)
+    assert queue[other]==100
+    assert other in queue
+    del queue[other]
+    assert len(queue)==0
 
 if __name__ == '__main__':
     pytest.main()

tests/test_utils.py

@@ -773,18 +773,24 @@ def __len__(self):
         """Return current capacity of PriorityQueue."""
         return len(self.heap)
 
-    def __contains__(self, item):
-        """Return True if item in PriorityQueue."""
-        return (self.f(item), item) in self.heap
+    def __contains__(self, key):
+        """Return True if the key is in PriorityQueue."""
+        return any([item == key for _, item in self.heap])
 
     def __getitem__(self, key):
-        for _, item in self.heap:
+        """Returns the first value associated with key in PriorityQueue.
+        Raises KeyError if key is not present."""
+        for value, item in self.heap:
             if item == key:
-                return item
+                return value
+        raise KeyError(str(key) + " is not in the priority queue")
 
     def __delitem__(self, key):
         """Delete the first occurrence of key."""
-        self.heap.remove((self.f(key), key))
+        try:
+            del self.heap[[item == key for _, item in self.heap].index(True)]
+        except ValueError:
+            raise KeyError(str(key) + " is not in the priority queue")
         heapq.heapify(self.heap)