mirror of https://github.com/scrapy/scrapy.git
Set TELNETCONSOLE_ENABLED=False in get_crawler().
This commit is contained in:
parent
c9f952c258
commit
7af7c4094a
|
|
@ -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 {}),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Reference in New Issue