priority_refresh_queue_test.py 文件源码

def test_refresh_queue_no_job(
        self,
        priority_refresh_queue,
        fake_source_name,
        refresh_result
    ):
        assert priority_refresh_queue.peek() == {}
        with pytest.raises(EmptyQueueError) as e:
            priority_refresh_queue.pop(fake_source_name)
        assert fake_source_name in e.value.message

        priority_refresh_queue.add_refreshes_to_queue([refresh_result])
        assert priority_refresh_queue.pop(fake_source_name) == refresh_result
        with pytest.raises(EmptyQueueError) as e:
            priority_refresh_queue.pop(fake_source_name)
        assert fake_source_name in e.value.message