Test specific cases for None slot handling with different priority queues.
(
mockserver: MockServer, priority_queue_class: str
)
| 138 | ) |
| 139 | @coroutine_test |
| 140 | async def test_none_slot_with_priority_queue( |
| 141 | mockserver: MockServer, priority_queue_class: str |
| 142 | ) -> None: |
| 143 | """Test specific cases for None slot handling with different priority queues.""" |
| 144 | crawler = get_crawler( |
| 145 | DownloaderSlotsSettingsTestSpider, |
| 146 | settings_dict={"SCHEDULER_PRIORITY_QUEUE": priority_queue_class}, |
| 147 | ) |
| 148 | await crawler.crawl_async(mockserver=mockserver) |
| 149 | assert isinstance(crawler.spider, DownloaderSlotsSettingsTestSpider) |
| 150 | |
| 151 | assert hasattr(crawler.spider, "times") |
| 152 | assert None not in crawler.spider.times |
| 153 | assert crawler.spider.default_slot in crawler.spider.times |
| 154 | assert len(crawler.spider.times[crawler.spider.default_slot]) == 2 |
| 155 | |
| 156 | assert crawler.stats |
| 157 | stats = crawler.stats |
| 158 | assert stats.get_value("spider_exceptions", 0) == 0 |
| 159 | assert stats.get_value("downloader/exception_count", 0) == 0 |
nothing calls this directly
no test coverage detected