diff --git a/scrapy/utils/test.py b/scrapy/utils/test.py index b4e20c3c6..90ed70262 100644 --- a/scrapy/utils/test.py +++ b/scrapy/utils/test.py @@ -69,6 +69,7 @@ def get_crawler( # When needed, useful settings can be added here, e.g. ones that prevent # deprecation warnings. settings: dict[str, Any] = { + "TELNETCONSOLE_ENABLED": False, **get_reactor_settings(), **(settings_dict or {}), } diff --git a/tests/test_extension_telnet.py b/tests/test_extension_telnet.py index 3f9135867..b2a2f4f54 100644 --- a/tests/test_extension_telnet.py +++ b/tests/test_extension_telnet.py @@ -1,61 +1,82 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + import pytest from twisted.conch.telnet import ITelnetProtocol from twisted.cred import credentials from scrapy.extensions.telnet import TelnetConsole +from scrapy.utils.defer import maybe_deferred_to_future from scrapy.utils.test import get_crawler -from tests.utils.decorators import inline_callbacks_test +from tests.utils.decorators import coroutine_test + +if TYPE_CHECKING: + from scrapy.crawler import Crawler pytestmark = pytest.mark.requires_reactor # TelnetConsole requires a reactor -class TestTelnetExtension: - def _get_console_and_portal(self, settings=None): - crawler = get_crawler(settings_dict=settings) - console = TelnetConsole(crawler) +def _get_crawler(settings_dict: dict[str, Any] | None = None) -> Crawler: + settings = { + "TELNETCONSOLE_ENABLED": True, + **(settings_dict or {}), + } + return get_crawler(settings_dict=settings) - # This function has some side effects we don't need for this test - console._get_telnet_vars = dict +def _get_console_and_portal( + settings: dict[str, Any] | None = None, +) -> tuple[TelnetConsole, Any]: + crawler = _get_crawler(settings_dict=settings) + console = TelnetConsole(crawler) + + # This function has some side effects we don't need for this test + console._get_telnet_vars = dict + + console.start_listening() + protocol = console.protocol() + portal = protocol.protocolArgs[0] + + return console, portal + + +@coroutine_test +async def test_bad_credentials() -> None: + console, portal = _get_console_and_portal() + creds = credentials.UsernamePassword(b"username", b"password") + d = portal.login(creds, None, ITelnetProtocol) + with pytest.raises(ValueError, match="Invalid credentials"): + await maybe_deferred_to_future(d) + console.stop_listening() + + +@coroutine_test +async def test_good_credentials() -> None: + console, portal = _get_console_and_portal() + creds = credentials.UsernamePassword( + console.username.encode("utf8"), console.password.encode("utf8") + ) + d = portal.login(creds, None, ITelnetProtocol) + await maybe_deferred_to_future(d) + console.stop_listening() + + +@coroutine_test +async def test_custom_credentials() -> None: + settings = { + "TELNETCONSOLE_USERNAME": "user", + "TELNETCONSOLE_PASSWORD": "pass", + } + console, portal = _get_console_and_portal(settings=settings) + creds = credentials.UsernamePassword(b"user", b"pass") + d = portal.login(creds, None, ITelnetProtocol) + await maybe_deferred_to_future(d) + console.stop_listening() + + +def test_invalid_reversed_portrange() -> None: + settings = {"TELNETCONSOLE_PORT": [2, 1]} + console = TelnetConsole(_get_crawler(settings_dict=settings)) + with pytest.raises(ValueError, match=r"invalid portrange: \[2, 1\]"): console.start_listening() - protocol = console.protocol() - portal = protocol.protocolArgs[0] - - return console, portal - - @inline_callbacks_test - def test_bad_credentials(self): - console, portal = self._get_console_and_portal() - creds = credentials.UsernamePassword(b"username", b"password") - d = portal.login(creds, None, ITelnetProtocol) - with pytest.raises(ValueError, match="Invalid credentials"): - yield d - console.stop_listening() - - @inline_callbacks_test - def test_good_credentials(self): - console, portal = self._get_console_and_portal() - creds = credentials.UsernamePassword( - console.username.encode("utf8"), console.password.encode("utf8") - ) - d = portal.login(creds, None, ITelnetProtocol) - yield d - console.stop_listening() - - @inline_callbacks_test - def test_custom_credentials(self): - settings = { - "TELNETCONSOLE_USERNAME": "user", - "TELNETCONSOLE_PASSWORD": "pass", - } - console, portal = self._get_console_and_portal(settings=settings) - creds = credentials.UsernamePassword(b"user", b"pass") - d = portal.login(creds, None, ITelnetProtocol) - yield d - console.stop_listening() - - def test_invalid_reversed_portrange(self): - settings = {"TELNETCONSOLE_PORT": [2, 1]} - console = TelnetConsole(get_crawler(settings_dict=settings)) - with pytest.raises(ValueError, match=r"invalid portrange: \[2, 1\]"): - console.start_listening()