Improve test coverage

scrapy/throttling.py

@@ -1054,7 +1054,9 @@ class ThrottlingScopeManager:
                 self._rampup_min_delay, self._delay * self._rampup_delay_factor
             )
             self._base_delay = min(self._base_delay, self._delay)
-        elif self._concurrency is not None:
+        else:
+            # Rampup is only enabled with a concurrency limit set.
+            assert self._concurrency is not None
             self._concurrency += 1
 
     def _maybe_reset_quota(self, now: float) -> None:

tests/test_core_downloader.py

@@ -14,7 +14,7 @@ from twisted.web import server, static
 from twisted.web.client import Agent, BrowserLikePolicyForHTTPS, readBody
 from twisted.web.client import Response as TxResponse
 
-from scrapy.core.downloader import Downloader, Slot, tls
+from scrapy.core.downloader import Downloader, Slot, _get_concurrency_delay, tls
 from scrapy.core.downloader.contextfactory import (
     _load_context_factory_from_settings,
     _ScrapyClientContextFactory,
@@ -45,6 +45,31 @@ class TestSlot:
         assert repr(slot) == "Slot(concurrency=8, delay=0.1, randomize_delay=True)"
 
 
+class TestGetConcurrencyDelay:
+    def test_default(self):
+        crawler = get_crawler()
+        concurrency, delay = _get_concurrency_delay(
+            8, DefaultSpider(), crawler.settings
+        )
+        assert (concurrency, delay) == (8, 0.0)
+
+    def test_spider_download_delay(self):
+        crawler = get_crawler()
+        spider = DefaultSpider()
+        spider.download_delay = 2.5
+        _concurrency, delay = _get_concurrency_delay(8, spider, crawler.settings)
+        assert delay == 2.5
+
+    def test_download_delay_per_slot(self):
+        crawler = get_crawler(settings_dict={"DOWNLOAD_DELAY_PER_SLOT": 3.0})
+        # DOWNLOAD_DELAY_PER_SLOT takes precedence over both DOWNLOAD_DELAY and
+        # the spider's download_delay attribute.
+        spider = DefaultSpider()
+        spider.download_delay = 2.5
+        _concurrency, delay = _get_concurrency_delay(8, spider, crawler.settings)
+        assert delay == 3.0
+
+
 @pytest.mark.requires_reactor  # this test is related to the Twisted HTTP code
 class TestContextFactoryBase:
     @async_yield_fixture

tests/test_engine.py

@@ -595,6 +595,21 @@ class TestEngineDownloadAsync:
         engine._slot.add_request.assert_called_once_with(request)
         engine._slot.remove_request.assert_called_once_with(request)
 
+    @coroutine_test
+    async def test_download_async_fetch_needs_spider(self, engine):
+        """A downloader whose fetch() requires a spider gets it passed in."""
+        engine._downloader_fetch_needs_spider = True
+        request = Request("http://example.com")
+        response = Response("http://example.com", body=b"test body")
+        engine.spider = Mock()
+        engine.downloader.fetch.return_value = defer.succeed(response)
+        engine._slot.add_request = Mock()
+        engine._slot.remove_request = Mock()
+
+        result = await self._download(engine, request)
+        assert result == response
+        engine.downloader.fetch.assert_called_once_with(request, engine.spider)
+
 
 @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
 class TestEngineDownload(TestEngineDownloadAsync):
@@ -640,6 +655,146 @@ async def test_request_scheduled_signal():
     crawler.signals.disconnect(signal_handler, signals.request_scheduled)
 
 
+class TestEngineThrottling:
+    @pytest.fixture
+    def engine(self):
+        crawler = get_crawler(MySpider)
+        engine = ExecutionEngine(crawler, lambda _: None)
+        yield engine
+        engine.downloader.close()
+
+    def test_pause_cancels_throttling_wakeup(self, engine):
+        wakeup = Mock()
+        engine._throttling_wakeup = wakeup
+        engine.pause()
+        assert engine.paused is True
+        wakeup.cancel.assert_called_once_with()
+        assert engine._throttling_wakeup is None
+        engine.unpause()
+        assert engine.paused is False
+
+    def test_maybe_arm_throttling_wakeup_arms_timer(self, engine):
+        scheduler = Mock()
+        scheduler.has_pending_requests.return_value = True
+        scheduler.next_request_delay.return_value = 5.0
+        engine._slot = Mock()
+        engine._slot.scheduler = scheduler
+        engine._maybe_arm_throttling_wakeup()
+        assert engine._throttling_wakeup is not None
+        # Cancel the scheduled reactor call so it does not leak into other tests.
+        engine._cancel_throttling_wakeup()
+
+    def test_maybe_arm_throttling_wakeup_no_delay(self, engine):
+        scheduler = Mock()
+        scheduler.has_pending_requests.return_value = True
+        scheduler.next_request_delay.return_value = None
+        engine._slot = Mock()
+        engine._slot.scheduler = scheduler
+        engine._maybe_arm_throttling_wakeup()
+        assert engine._throttling_wakeup is None
+
+    def test_warn_delayed_requests(self, engine):
+        engine._delayed_requests_warn_threshold = 1
+        engine._throttling_waiting = {Request("http://a.example")}
+        engine._slot = Mock()
+        # A scheduler without next_request_delay is not throttling-aware, so the
+        # warning recommends switching to one.
+        engine._slot.scheduler = Mock(spec=BaseScheduler)
+        with LogCapture() as log:
+            engine._maybe_warn_delayed_requests()
+            # A second call is a no-op (the warning is emitted only once).
+            engine._maybe_warn_delayed_requests()
+        assert engine._delayed_requests_warned is True
+        log_text = str(log)
+        assert "requests held back by throttling" in log_text
+        assert log_text.count("ThrottlingAwareScheduler") == 1
+
+    def test_warn_delayed_requests_throttling_aware(self, engine):
+        engine._delayed_requests_warn_threshold = 1
+        engine._throttling_waiting = {Request("http://a.example")}
+        engine._slot = Mock()
+        # A throttling-aware scheduler (one with next_request_delay) holds
+        # throttled requests itself, so no switch is recommended.
+        engine._slot.scheduler = Mock()
+        with LogCapture() as log:
+            engine._maybe_warn_delayed_requests()
+        assert "ThrottlingAwareScheduler" not in str(log)
+
+    def test_spider_is_idle_false_while_scheduling(self, engine):
+        engine._slot = Mock()
+        engine.scraper.slot = Mock()
+        engine.scraper.slot.is_idle.return_value = True
+        engine.downloader = Mock()
+        engine.downloader.active = []
+        engine._throttling_waiting = set()
+        engine._start = None
+        engine._scheduling = 1
+        # An in-flight async enqueue keeps the spider from being considered idle.
+        assert engine.spider_is_idle() is False
+
+    @coroutine_test
+    async def test_enqueue_request_async_dropped(self, engine):
+        scheduler = Mock()
+
+        async def enqueue_request_async(request):
+            return False
+
+        scheduler.enqueue_request_async = enqueue_request_async
+        engine._slot = Mock()
+        engine._slot.scheduler = scheduler
+        engine.spider = Mock()
+        dropped = []
+
+        def on_dropped(request, spider):
+            dropped.append(request)
+
+        engine.signals.connect(on_dropped, signals.request_dropped, weak=False)
+        engine._scheduling = 1
+        request = Request("http://a.example")
+        await engine._enqueue_request_async(request)
+        assert dropped == [request]
+        assert engine._scheduling == 0
+        engine._slot.nextcall.schedule.assert_called_once_with()
+
+    @coroutine_test
+    async def test_enqueue_request_async_error(self, engine):
+        scheduler = Mock()
+
+        async def enqueue_request_async(request):
+            raise RuntimeError("boom")
+
+        scheduler.enqueue_request_async = enqueue_request_async
+        engine._slot = Mock()
+        engine._slot.scheduler = scheduler
+        engine.spider = Mock()
+        engine._scheduling = 1
+        with LogCapture() as log:
+            await engine._enqueue_request_async(Request("http://a.example"))
+        assert "Error while enqueuing request" in str(log)
+        assert engine._scheduling == 0
+        engine._slot.nextcall.schedule.assert_called_once_with()
+
+    @coroutine_test
+    async def test_enqueue_request_async_slot_gone(self, engine):
+        scheduler = Mock()
+
+        async def enqueue_request_async(request):
+            # The spider is closed while the enqueue is in flight.
+            engine._slot = None
+            return True
+
+        scheduler.enqueue_request_async = enqueue_request_async
+        slot = Mock()
+        slot.scheduler = scheduler
+        engine._slot = slot
+        engine.spider = Mock()
+        engine._scheduling = 1
+        await engine._enqueue_request_async(Request("http://a.example"))
+        assert engine._scheduling == 0
+        # No reschedule is attempted once the slot is gone.
+        slot.nextcall.schedule.assert_not_called()
+
+
 class TestEngineCloseSpider:
     """Tests for exception handling coverage during close_spider_async()."""
 

tests/test_pqueues.py

@@ -448,6 +448,43 @@ class TestThrottlingAwarePriorityQueue:
         queue.pqueues[frozenset({"a.com"})] = queue.pqfactory(frozenset({"a.com"}))
         assert queue.next_request_delay() is None
 
+    @coroutine_test
+    async def test_next_request_delay_keeps_minimum(self):
+        crawler = get_crawler(
+            Spider,
+            settings_dict={
+                "THROTTLING_SCOPES": {
+                    "a.com": {"delay": 10.0},
+                    "b.com": {"delay": 1000.0},
+                },
+                "RANDOMIZE_DOWNLOAD_DELAY": False,
+            },
+        )
+        queue = self._queue(crawler)
+        # Two requests per scope so a blocked head remains after the first one
+        # (sendable, since no delay has accrued yet) is popped and reserved.
+        await self._push(queue, crawler, Request("http://a.com/1"))
+        await self._push(queue, crawler, Request("http://a.com/2"))
+        await self._push(queue, crawler, Request("http://b.com/1"))
+        await self._push(queue, crawler, Request("http://b.com/2"))
+        queue.pop()
+        queue.pop()
+        # Both scopes are now time-blocked; the smaller per-scope delay wins,
+        # so the larger one exercises the "not below the running minimum" branch.
+        delay = queue.next_request_delay()
+        assert delay == pytest.approx(10.0, abs=1.0)
+
+    @coroutine_test
+    async def test_pop_handles_drained_selected_queue(self):
+        crawler = get_crawler(Spider)
+        queue = self._queue(crawler)
+        await self._push(queue, crawler, Request("http://a.com/1"))
+        inner = next(iter(queue.pqueues.values()))
+        # peek() still reports a sendable head, but pop() yields nothing: the
+        # request-is-None guard must not try to reserve a missing request.
+        inner.pop = lambda: None
+        assert queue.pop() is None
+
     @coroutine_test
     async def test_contains(self):
         crawler = get_crawler(Spider)

tests/test_throttling.py

@@ -478,6 +478,16 @@ class TestThrottlingScopeManager:
         scope.record_done(now=0.0)
         assert not event.called
 
+    def test_fire_slot_waiters_skips_already_fired(self):
+        scope = _scope_manager(config={"id": "x", "concurrency": 1})
+        scope.record_sent(now=0.0)
+        event = scope.slot_event()
+        event.callback(None)  # fired out-of-band before the slot frees up
+        # record_done() fires the waiters; the already-fired one is skipped
+        # rather than called a second time (which would raise).
+        scope.record_done(now=0.0)
+        assert event.called
+
     def test_set_concurrency_respects_min(self):
         scope = _scope_manager(config={"id": "x", "min_concurrency": 3})
         scope.set_concurrency(1)
@@ -781,6 +791,42 @@ class TestThrottlingManagerEdges:
         # A second call is a no-op (the request was already delayed).
         await manager._apply_request_delay(request)
 
+    @coroutine_test
+    async def test_apply_request_delay_without_debug(self):
+        # Same as above but with debug logging off, so the delay is applied
+        # without emitting the debug message.
+        manager = _manager()
+        request = Request("http://example.com/a", meta={"throttling_delay": 0.01})
+        await manager._apply_request_delay(request)
+        assert request.meta["_throttling_delayed"] is True
+
+    @coroutine_test
+    async def test_wait_for_slot_discards_unfired_events(self):
+        from scrapy.utils.asyncio import call_later  # noqa: PLC0415
+
+        manager = _manager()
+        m1 = _scope_manager(config={"id": "a", "concurrency": 1})
+        m2 = _scope_manager(config={"id": "b", "concurrency": 1})
+        m1.record_sent(now=0.0)
+        m2.record_sent(now=0.0)
+        # Free m1's slot on the next tick so the wait wakes up with m1's event
+        # fired while m2's event is still pending.
+        call_later(0, m1.record_done)
+        await manager._wait_for_slot([m1, m2])
+        # The still-pending m2 event is discarded from its waiter list.
+        assert m2._slot_waiters == []
+
+    def test_scope_manager_protocol_defaults(self):
+        from scrapy.throttling import ThrottlingScopeManagerProtocol  # noqa: PLC0415
+
+        class _Concrete(ThrottlingScopeManagerProtocol):
+            pass
+
+        crawler = get_crawler()
+        # The protocol's default from_crawler()/__init__ are usable as-is.
+        instance = _Concrete.from_crawler(crawler, {"id": "example.com"})
+        assert isinstance(instance, _Concrete)
+
     @coroutine_test
     async def test_process_exception_applies_backoff(self):
         manager = _manager()

tests/test_utils_asyncio.py

@@ -173,3 +173,20 @@ class TestWaitForFirst:
         done, pending = await wait_for_first([fired], timeout=1)
         assert done == {fired}
         assert pending == set()
+
+    @coroutine_test
+    async def test_no_timeout(self):
+        # timeout=None -> no timeout deferred is created.
+        fired: Deferred[None] = Deferred()
+        fired.callback(None)
+        done, pending = await wait_for_first([fired])
+        assert done == {fired}
+        assert pending == set()
+
+    @coroutine_test
+    async def test_timeout_elapses(self):
+        # When the timeout fires first, every input deferred is still pending.
+        never: Deferred[None] = Deferred()
+        done, pending = await wait_for_first([never], timeout=0.01)
+        assert done == set()
+        assert pending == {never}