File: test_extension_telnet.py

package info (click to toggle)
python-scrapy 2.14.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 6,308 kB
  • sloc: python: 55,321; xml: 199; makefile: 25; sh: 7
file content (53 lines) | stat: -rw-r--r-- 1,826 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import pytest
from twisted.conch.telnet import ITelnetProtocol
from twisted.cred import credentials
from twisted.internet.defer import inlineCallbacks

from scrapy.extensions.telnet import TelnetConsole
from scrapy.utils.test import get_crawler


class TestTelnetExtension:
    def _get_console_and_portal(self, settings=None):
        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

    @inlineCallbacks
    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()

    @inlineCallbacks
    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()

    @inlineCallbacks
    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()