Address issues caught by tests

This commit is contained in:
Adrian Chaves 2026-08-08 17:25:50 +02:00
parent 7289af962b
commit f530b6358a
3 changed files with 47 additions and 20 deletions

View File

@ -97,8 +97,6 @@ def _accepted(receiver: TypingAny) -> frozenset[str] | None:
return _accepted_args[key]
except KeyError:
pass
except TypeError:
return _introspect(receiver)
names = _introspect(receiver)
_accepted_args[key] = names
return names
@ -192,10 +190,7 @@ def _positional_names(receiver: TypingAny, count: int) -> frozenset[str]:
"""Return the names of the first *count* parameters of *receiver*, which
positional arguments bind and keyword arguments therefore must not repeat.
"""
try:
parameters = list(inspect.signature(receiver).parameters.values())
except (TypeError, ValueError):
return frozenset()
parameters = list(inspect.signature(receiver).parameters.values())
return frozenset(
p.name
for p in parameters[:count]

View File

@ -5,9 +5,9 @@ import pytest
from scrapy import signals
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
from scrapy.signalmanager import SignalManager
from scrapy.statscollectors import StatsCollector
from scrapy.utils.spider import DefaultSpider
from scrapy.utils.test import get_crawler
with warnings.catch_warnings():
warnings.filterwarnings(
@ -46,11 +46,8 @@ def test_from_crawler_without_recipients_raises_notconfigured():
statsmailer.StatsMailer.from_crawler(crawler)
def test_from_crawler_with_recipients_initializes_extension(dummy_stats, monkeypatch):
crawler = MagicMock()
crawler.settings.getlist.return_value = ["test@example.com"]
crawler.stats = dummy_stats
crawler.signals = SignalManager(crawler)
def test_from_crawler_with_recipients_initializes_extension(monkeypatch):
crawler = get_crawler(settings_dict={"STATSMAILER_RCPTS": ["test@example.com"]})
mailer = MagicMock(spec=MailSender)
monkeypatch.setattr(statsmailer.MailSender, "from_crawler", lambda _: mailer)
@ -62,21 +59,18 @@ def test_from_crawler_with_recipients_initializes_extension(dummy_stats, monkeyp
assert ext.mail is mailer
def test_from_crawler_connects_spider_closed_signal(dummy_stats, monkeypatch):
crawler = MagicMock()
crawler.settings.getlist.return_value = ["test@example.com"]
crawler.stats = dummy_stats
crawler.signals = SignalManager(crawler)
def test_from_crawler_connects_spider_closed_signal(monkeypatch):
crawler = get_crawler(settings_dict={"STATSMAILER_RCPTS": ["test@example.com"]})
mailer = MagicMock(spec=MailSender)
monkeypatch.setattr(statsmailer.MailSender, "from_crawler", lambda _: mailer)
statsmailer.StatsMailer.from_crawler(crawler)
ext = statsmailer.StatsMailer.from_crawler(crawler)
connected = crawler.signals.send_catch_log(
crawler.signals.send_catch_log(
signals.spider_closed, spider=DefaultSpider(name="dummy")
)
assert connected is not None
assert ext.mail.send.call_count == 1
def test_spider_closed_sends_email(dummy_stats):

View File

@ -11,6 +11,7 @@ import pytest
from scrapy import signals
from scrapy.signalmanager import SignalManager
from scrapy.utils import _signal_registry as registry
from scrapy.utils.signal import send_catch_log
if TYPE_CHECKING:
from collections.abc import Callable
@ -67,6 +68,21 @@ class TestArgumentDelivery:
sm.send_catch_log(signal, spider="SPIDER")
assert received == {"spider": "p-SPIDER"}
def test_handler_that_cannot_be_introspected_gets_every_argument(self) -> None:
assert registry.apply(dict, spider="SPIDER") == {"spider": "SPIDER"}
def test_positional_arguments_are_not_repeated_as_keywords(self) -> None:
received: dict[str, Any] = {}
def handler(spider: Any, signal: Any = None) -> None:
received.update(spider=spider, signal=signal)
signal = object()
sender = object()
registry.connect(handler, signal, sender=sender)
send_catch_log(signal, sender, "SPIDER")
assert received == {"spider": "SPIDER", "signal": signal}
def test_callable_object_handler(self) -> None:
class Handler:
def __init__(self) -> None:
@ -185,6 +201,10 @@ class TestUnknownArgumentWarning:
with pytest.warns(UserWarning, match=r"scrapy\.signals\.spider_opened"):
sm.connect(handler, signals.spider_opened)
def test_warning_falls_back_to_the_repr_of_the_signal(self) -> None:
signal = object()
assert registry._signal_name(signal) == repr(signal)
def test_lists_every_unknown_argument(self) -> None:
def handler(cheese: Any = None, ham: Any = None) -> None:
pass
@ -243,6 +263,24 @@ class TestReceiverCache:
assert registry.receivers(signal, sm.sender) == []
class TestSignalStorage:
def test_signal_that_can_be_weakly_referenced_is_not_kept_alive(self) -> None:
class Signal:
pass
def handler(**kwargs: Any) -> None:
pass
signal = Signal()
ref = weakref.ref(signal)
sm = SignalManager(object())
sm.connect(handler, signal)
sm.send_catch_log(signal)
del signal
gc.collect()
assert ref() is None
class TestMutationDuringDispatch:
def test_handler_can_disconnect_itself(self) -> None:
calls: list[str] = []