source: test/test_session.py @ 1501:cc42f6ab9d3e

Last change on this file since 1501:cc42f6ab9d3e was 1501:cc42f6ab9d3e, checked in by Stefan Schwarzer <sschwarzer@…>, 7 years ago
Encrypt data channel by default if a method `prot_p` exists. I think most users will want to secure the data channel, so rather be safe than sorry and turn the encryption on by default. If needed, the data channel encryption can be suppressed by passing `encrypt_data_channel=False`.
File size: 5.5 KB
Line 
1# Copyright (C) 2014, Stefan Schwarzer
2
3"""
4Unit tests for session factory helpers.
5"""
6
7from __future__ import unicode_literals
8
9import unittest
10
11import ftputil.session
12
13
14class MockSession(object):
15    """
16    Mock session base class to determine if all expected calls
17    have happened.
18    """
19
20    def __init__(self):
21        self.calls = []
22
23    def add_call(self, *args):
24        self.calls.append(args)
25
26    def connect(self, host, port):
27        self.add_call("connect", host, port)
28
29    def _fix_socket(self):
30        self.add_call("_fix_socket")
31
32    def set_debuglevel(self, value):
33        self.add_call("set_debuglevel", value)
34
35    def login(self, user, password):
36        self.add_call("login", user, password)
37
38    def set_pasv(self, flag):
39        self.add_call("set_pasv", flag)
40
41
42class EncryptedMockSession(MockSession):
43
44    def auth_tls(self):
45        self.add_call("auth_tls")
46
47    def prot_p(self):
48        self.add_call("prot_p")
49
50
51class TestSessionFactory(unittest.TestCase):
52    """
53    Test if session factories created by
54    `ftputil.session.session_factory` trigger the expected calls.
55    """
56
57    def test_defaults(self):
58        """Test defaults (apart from base class)."""
59        factory = \
60          ftputil.session.session_factory(base_class=MockSession)
61        session = factory("host", "user", "password")
62        self.assertEqual(session.calls, [("connect", "host", 21),
63                                         ("login", "user", "password")])
64
65    def test_different_port(self):
66        """Test setting the command channel port with `port`."""
67        factory = \
68          ftputil.session.session_factory(base_class=MockSession, port=2121)
69        session = factory("host", "user", "password")
70        self.assertEqual(session.calls, [("connect", "host", 2121),
71                                         ("login", "user", "password")])
72
73    def test_use_passive_mode(self):
74        """
75        Test explicitly setting passive/active mode with
76        `use_passive_mode`.
77        """
78        # Passive mode
79        factory = ftputil.session.session_factory(base_class=MockSession,
80                                                  use_passive_mode=True)
81        session = factory("host", "user", "password")
82        self.assertEqual(session.calls, [("connect", "host", 21),
83                                         ("login", "user", "password"),
84                                         ("set_pasv", True)])
85        # Active mode
86        factory = ftputil.session.session_factory(base_class=MockSession,
87                                                  use_passive_mode=False)
88        session = factory("host", "user", "password")
89        self.assertEqual(session.calls, [("connect", "host", 21),
90                                         ("login", "user", "password"),
91                                         ("set_pasv", False)])
92
93    def test_encrypt_data_channel(self):
94        """Test request to call `prot_p` with `encrypt_data_channel`."""
95        # With encrypted data channel (default for encrypted session).
96        factory = ftputil.session.session_factory(
97                    base_class=EncryptedMockSession)
98        session = factory("host", "user", "password")
99        self.assertEqual(session.calls, [("connect", "host", 21),
100                                         ("login", "user", "password"),
101                                         ("prot_p",)])
102        #
103        factory = ftputil.session.session_factory(
104                    base_class=EncryptedMockSession, encrypt_data_channel=True)
105        session = factory("host", "user", "password")
106        self.assertEqual(session.calls, [("connect", "host", 21),
107                                         ("login", "user", "password"),
108                                         ("prot_p",)])
109        # Without encrypted data channel.
110        factory = ftputil.session.session_factory(
111                    base_class=EncryptedMockSession, encrypt_data_channel=False)
112        session = factory("host", "user", "password")
113        self.assertEqual(session.calls, [("connect", "host", 21),
114                                         ("login", "user", "password")])
115
116    def test_debug_level(self):
117        """Test setting the debug level on the session."""
118        factory = ftputil.session.session_factory(base_class=MockSession,
119                                                  debug_level=1)
120        session = factory("host", "user", "password")
121        self.assertEqual(session.calls, [("connect", "host", 21),
122                                         ("set_debuglevel", 1),
123                                         ("login", "user", "password")])
124
125    def test_m2crypto_session(self):
126        """Test call sequence for M2Crypto session."""
127        factory = \
128          ftputil.session.session_factory(base_class=EncryptedMockSession)
129        # Return `True` to fake that this is a session deriving from
130        # `M2Crypto.ftpslib.FTP_TLS`.
131        factory._use_m2crypto_ftpslib = lambda self: True
132        # Override `_fix_socket` here, not in `MockSession`. Since
133        # the created session class _inherits_ from `MockSession`,
134        # it would override the `_fix_socket` there.
135        factory._fix_socket = lambda self: self.add_call("_fix_socket")
136        session = factory("host", "user", "password")
137        self.assertEqual(session.calls, [("connect", "host", 21),
138                                         ("auth_tls",),
139                                         ("_fix_socket",),
140                                         ("login", "user", "password"),
141                                         ("prot_p",)])
142
143
144if __name__ == "__main__":
145    unittest.main()
Note: See TracBrowser for help on using the repository browser.