Method test_dequeue_priorities

tests/test_scheduler.py:177–188

(self, jobdir: Path | None)

    @coroutine_test
    async def test_dequeue_priorities(self, jobdir: Path | None) -> None:
        async with self.create_scheduler(jobdir) as scheduler:
            for url, priority in _PRIORITIES:
                scheduler.enqueue_request(Request(url, priority=priority))

            priorities = []
            while scheduler.has_pending_requests():
                request = scheduler.next_request()
                assert request is not None
                priorities.append(request.priority)

        assert priorities == sorted([x[1] for x in _PRIORITIES], key=lambda x: -x)


class TestSchedulerOnDiskBase(TestSchedulerBase):