| 74 | |
| 75 | @pytest.mark.parametrize("test_peek", [True, False]) |
| 76 | def test_order(self, q: queuelib.queue.BaseQueue, test_peek: bool): |
| 77 | if test_peek and not HAVE_PEEK: |
| 78 | pytest.skip("The queuelib queues do not define peek") |
| 79 | if not test_peek and HAVE_PEEK: |
| 80 | pytest.skip("The queuelib queues define peek") |
| 81 | assert len(q) == 0 |
| 82 | if test_peek: |
| 83 | assert q.peek() is None |
| 84 | assert q.pop() is None |
| 85 | req1 = Request("http://www.example.com/1") |
| 86 | req2 = Request("http://www.example.com/2") |
| 87 | req3 = Request("http://www.example.com/3") |
| 88 | q.push(req1) |
| 89 | q.push(req2) |
| 90 | q.push(req3) |
| 91 | if not test_peek: |
| 92 | with pytest.raises( |
| 93 | NotImplementedError, |
| 94 | match="The underlying queue class does not implement 'peek'", |
| 95 | ): |
| 96 | q.peek() |
| 97 | reqs = [req1, req2, req3] if self.is_fifo else [req3, req2, req1] |
| 98 | for i, req in enumerate(reqs): |
| 99 | assert len(q) == 3 - i |
| 100 | if test_peek: |
| 101 | result = q.peek() |
| 102 | assert result is not None |
| 103 | assert result.url == req.url |
| 104 | result = q.pop() |
| 105 | assert result is not None |
| 106 | assert result.url == req.url |
| 107 | assert len(q) == 0 |
| 108 | if test_peek: |
| 109 | assert q.peek() is None |
| 110 | assert q.pop() is None |
| 111 | q.close() |
| 112 | |
| 113 | |
| 114 | class TestPickleFifoDiskQueueRequest(TestRequestQueueBase): |