1 | # Copyright (C) 2014-2018, Stefan Schwarzer <sschwarzer@sschwarzer.net> |
---|
2 | # and ftputil contributors (see `doc/contributors.txt`) |
---|
3 | # See the file LICENSE for licensing terms. |
---|
4 | |
---|
5 | """ |
---|
6 | Unit tests for session factory helpers. |
---|
7 | """ |
---|
8 | |
---|
9 | import ftputil.session |
---|
10 | |
---|
11 | |
---|
12 | class MockSession: |
---|
13 | """ |
---|
14 | Mock session base class to determine if all expected calls |
---|
15 | have happened. |
---|
16 | """ |
---|
17 | |
---|
18 | def __init__(self): |
---|
19 | self.calls = [] |
---|
20 | |
---|
21 | def add_call(self, *args): |
---|
22 | self.calls.append(args) |
---|
23 | |
---|
24 | def connect(self, host, port): |
---|
25 | self.add_call("connect", host, port) |
---|
26 | |
---|
27 | def _fix_socket(self): |
---|
28 | self.add_call("_fix_socket") |
---|
29 | |
---|
30 | def set_debuglevel(self, value): |
---|
31 | self.add_call("set_debuglevel", value) |
---|
32 | |
---|
33 | def login(self, user, password): |
---|
34 | self.add_call("login", user, password) |
---|
35 | |
---|
36 | def set_pasv(self, flag): |
---|
37 | self.add_call("set_pasv", flag) |
---|
38 | |
---|
39 | |
---|
40 | class EncryptedMockSession(MockSession): |
---|
41 | |
---|
42 | def auth_tls(self): |
---|
43 | self.add_call("auth_tls") |
---|
44 | |
---|
45 | def prot_p(self): |
---|
46 | self.add_call("prot_p") |
---|
47 | |
---|
48 | |
---|
49 | class TestSessionFactory: |
---|
50 | """ |
---|
51 | Test if session factories created by |
---|
52 | `ftputil.session.session_factory` trigger the expected calls. |
---|
53 | """ |
---|
54 | |
---|
55 | def test_defaults(self): |
---|
56 | """Test defaults (apart from base class).""" |
---|
57 | factory = \ |
---|
58 | ftputil.session.session_factory(base_class=MockSession) |
---|
59 | session = factory("host", "user", "password") |
---|
60 | assert (session.calls == |
---|
61 | [("connect", "host", 21), ("login", "user", "password")]) |
---|
62 | |
---|
63 | def test_different_port(self): |
---|
64 | """Test setting the command channel port with `port`.""" |
---|
65 | factory = \ |
---|
66 | ftputil.session.session_factory(base_class=MockSession, port=2121) |
---|
67 | session = factory("host", "user", "password") |
---|
68 | assert (session.calls == |
---|
69 | [("connect", "host", 2121), ("login", "user", "password")]) |
---|
70 | |
---|
71 | def test_use_passive_mode(self): |
---|
72 | """ |
---|
73 | Test explicitly setting passive/active mode with |
---|
74 | `use_passive_mode`. |
---|
75 | """ |
---|
76 | # Passive mode |
---|
77 | factory = ftputil.session.session_factory(base_class=MockSession, |
---|
78 | use_passive_mode=True) |
---|
79 | session = factory("host", "user", "password") |
---|
80 | assert session.calls == [("connect", "host", 21), |
---|
81 | ("login", "user", "password"), |
---|
82 | ("set_pasv", True)] |
---|
83 | # Active mode |
---|
84 | factory = ftputil.session.session_factory(base_class=MockSession, |
---|
85 | use_passive_mode=False) |
---|
86 | session = factory("host", "user", "password") |
---|
87 | assert session.calls == [("connect", "host", 21), |
---|
88 | ("login", "user", "password"), |
---|
89 | ("set_pasv", False)] |
---|
90 | |
---|
91 | def test_encrypt_data_channel(self): |
---|
92 | """Test request to call `prot_p` with `encrypt_data_channel`.""" |
---|
93 | # With encrypted data channel (default for encrypted session). |
---|
94 | factory = ftputil.session.session_factory( |
---|
95 | base_class=EncryptedMockSession) |
---|
96 | session = factory("host", "user", "password") |
---|
97 | assert session.calls == [("connect", "host", 21), |
---|
98 | ("login", "user", "password"), |
---|
99 | ("prot_p",)] |
---|
100 | # |
---|
101 | factory = ftputil.session.session_factory( |
---|
102 | base_class=EncryptedMockSession, encrypt_data_channel=True) |
---|
103 | session = factory("host", "user", "password") |
---|
104 | assert session.calls == [("connect", "host", 21), |
---|
105 | ("login", "user", "password"), |
---|
106 | ("prot_p",)] |
---|
107 | # Without encrypted data channel. |
---|
108 | factory = ftputil.session.session_factory( |
---|
109 | base_class=EncryptedMockSession, encrypt_data_channel=False) |
---|
110 | session = factory("host", "user", "password") |
---|
111 | assert session.calls == [("connect", "host", 21), |
---|
112 | ("login", "user", "password")] |
---|
113 | |
---|
114 | def test_debug_level(self): |
---|
115 | """Test setting the debug level on the session.""" |
---|
116 | factory = ftputil.session.session_factory(base_class=MockSession, |
---|
117 | debug_level=1) |
---|
118 | session = factory("host", "user", "password") |
---|
119 | assert session.calls == [("connect", "host", 21), |
---|
120 | ("set_debuglevel", 1), |
---|
121 | ("login", "user", "password")] |
---|
122 | |
---|
123 | def test_m2crypto_session(self): |
---|
124 | """Test call sequence for M2Crypto session.""" |
---|
125 | factory = \ |
---|
126 | ftputil.session.session_factory(base_class=EncryptedMockSession) |
---|
127 | # Return `True` to fake that this is a session deriving from |
---|
128 | # `M2Crypto.ftpslib.FTP_TLS`. |
---|
129 | factory._use_m2crypto_ftpslib = lambda self: True |
---|
130 | # Override `_fix_socket` here, not in `MockSession`. Since |
---|
131 | # the created session class _inherits_ from `MockSession`, |
---|
132 | # it would override the `_fix_socket` there. |
---|
133 | factory._fix_socket = lambda self: self.add_call("_fix_socket") |
---|
134 | session = factory("host", "user", "password") |
---|
135 | assert session.calls == [("connect", "host", 21), |
---|
136 | ("auth_tls",), |
---|
137 | ("_fix_socket",), |
---|
138 | ("login", "user", "password"), |
---|
139 | ("prot_p",)] |
---|