File: test_file.py

package info (click to toggle)
python-ftputil 3.4-3
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, bullseye, sid
  • size: 848 kB
  • sloc: python: 3,308; makefile: 3
file content (290 lines) | stat: -rw-r--r-- 9,952 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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
# Copyright (C) 2002-2016, Stefan Schwarzer <sschwarzer@sschwarzer.net>
# and ftputil contributors (see `doc/contributors.txt`)
# See the file LICENSE for licensing terms.

from __future__ import unicode_literals

import ftplib

import pytest

import ftputil.compat
import ftputil.error

from test import mock_ftplib
from test import test_base


#
# Several customized `MockSession` classes
#
class ReadMockSession(mock_ftplib.MockSession):

    mock_file_content = b"line 1\r\nanother line\r\nyet another line"


class ReadMockSessionWithMoreNewlines(mock_ftplib.MockSession):

    mock_file_content = b"\r\n".join(map(ftputil.compat.bytes_type, range(20)))


class InaccessibleDirSession(mock_ftplib.MockSession):

    _login_dir = "/inaccessible"

    def pwd(self):
        return self._login_dir

    def cwd(self, dir):
        if dir in (self._login_dir, self._login_dir + "/"):
            raise ftplib.error_perm
        else:
            super(InaccessibleDirSession, self).cwd(dir)


class TestFileOperations(object):
    """Test operations with file-like objects."""

    def test_inaccessible_dir(self):
        """Test whether opening a file at an invalid location fails."""
        host = test_base.ftp_host_factory(
                 session_factory=InaccessibleDirSession)
        with pytest.raises(ftputil.error.FTPIOError):
            host.open("/inaccessible/new_file", "w")

    def test_caching(self):
        """Test whether `FTPFile` cache of `FTPHost` object works."""
        host = test_base.ftp_host_factory()
        assert len(host._children) == 0
        path1 = "path1"
        path2 = "path2"
        # Open one file and inspect cache.
        file1 = host.open(path1, "w")
        child1 = host._children[0]
        assert len(host._children) == 1
        assert not child1._file.closed
        # Open another file.
        file2 = host.open(path2, "w")
        child2 = host._children[1]
        assert len(host._children) == 2
        assert not child2._file.closed
        # Close first file.
        file1.close()
        assert len(host._children) == 2
        assert child1._file.closed
        assert not child2._file.closed
        # Re-open first child's file.
        file1 = host.open(path1, "w")
        child1_1 = file1._host
        # Check if it's reused.
        assert child1 is child1_1
        assert not child1._file.closed
        assert not child2._file.closed
        # Close second file.
        file2.close()
        assert child2._file.closed

    def test_write_to_directory(self):
        """Test whether attempting to write to a directory fails."""
        host = test_base.ftp_host_factory()
        with pytest.raises(ftputil.error.FTPIOError):
            host.open("/home/sschwarzer", "w")

    def test_binary_read(self):
        """Read data from a binary file."""
        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
        with host.open("some_file", "rb") as fobj:
            data = fobj.read()
        assert data == ReadMockSession.mock_file_content

    def test_binary_write(self):
        """Write binary data with `write`."""
        host = test_base.ftp_host_factory()
        data = b"\000a\001b\r\n\002c\003\n\004\r\005"
        with host.open("dummy", "wb") as output:
            output.write(data)
        child_data = mock_ftplib.content_of("dummy")
        expected_data = data
        assert child_data == expected_data

    def test_ascii_read(self):
        """Read ASCII text with plain `read`."""
        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
        with host.open("dummy", "r") as input_:
            data = input_.read(0)
            assert data == ""
            data = input_.read(3)
            assert data == "lin"
            data = input_.read(7)
            assert data == "e 1\nano"
            data = input_.read()
            assert data == "ther line\nyet another line"
            data = input_.read()
            assert data == ""

    def test_ascii_write(self):
        """Write ASCII text with `write`."""
        host = test_base.ftp_host_factory()
        data = " \nline 2\nline 3"
        with host.open("dummy", "w", newline="\r\n") as output:
            output.write(data)
        child_data = mock_ftplib.content_of("dummy")
        # This corresponds to the byte stream, so expect a `bytes` object.
        expected_data = b" \r\nline 2\r\nline 3"
        assert child_data == expected_data

    # TODO: Add tests with given encoding and possibly buffering.

    def test_ascii_writelines(self):
        """Write ASCII text with `writelines`."""
        host = test_base.ftp_host_factory()
        data = [" \n", "line 2\n", "line 3"]
        backup_data = data[:]
        output = host.open("dummy", "w", newline="\r\n")
        output.writelines(data)
        output.close()
        child_data = mock_ftplib.content_of("dummy")
        expected_data = b" \r\nline 2\r\nline 3"
        assert child_data == expected_data
        # Ensure that the original data was not modified.
        assert data == backup_data

    def test_binary_readline(self):
        """Read binary data with `readline`."""
        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
        input_ = host.open("dummy", "rb")
        data = input_.readline(3)
        assert data == b"lin"
        data = input_.readline(10)
        assert data == b"e 1\r\n"
        data = input_.readline(13)
        assert data == b"another line\r"
        data = input_.readline()
        assert data == b"\n"
        data = input_.readline()
        assert data == b"yet another line"
        data = input_.readline()
        assert data == b""
        input_.close()

    def test_ascii_readline(self):
        """Read ASCII text with `readline`."""
        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
        input_ = host.open("dummy", "r")
        data = input_.readline(3)
        assert data == "lin"
        data = input_.readline(10)
        assert data == "e 1\n"
        data = input_.readline(13)
        assert data == "another line\n"
        data = input_.readline()
        assert data == "yet another line"
        data = input_.readline()
        assert data == ""
        input_.close()

    def test_ascii_readlines(self):
        """Read ASCII text with `readlines`."""
        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
        input_ = host.open("dummy", "r")
        data = input_.read(3)
        assert data == "lin"
        data = input_.readlines()
        assert data == ["e 1\n", "another line\n", "yet another line"]
        input_.close()

    def test_binary_iterator(self):
        """
        Test the iterator interface of `FTPFile` objects (without
        newline conversion.
        """
        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
        input_ = host.open("dummy", "rb")
        input_iterator = iter(input_)
        assert next(input_iterator) == b"line 1\r\n"
        assert next(input_iterator) == b"another line\r\n"
        assert next(input_iterator) == b"yet another line"
        with pytest.raises(StopIteration):
            input_iterator.__next__()
        input_.close()

    def test_ascii_iterator(self):
        """
        Test the iterator interface of `FTPFile` objects (with newline
        conversion).
        """
        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
        input_ = host.open("dummy")
        input_iterator = iter(input_)
        assert next(input_iterator) == "line 1\n"
        assert next(input_iterator) == "another line\n"
        assert next(input_iterator) == "yet another line"
        with pytest.raises(StopIteration):
            input_iterator.__next__()
        input_.close()

    def test_read_unknown_file(self):
        """Test whether reading a file which isn't there fails."""
        host = test_base.ftp_host_factory()
        with pytest.raises(ftputil.error.FTPIOError):
            host.open("notthere", "r")


class TestAvailableChild(object):

    def _failing_pwd(self, exception_class):
        """
        Return a function that will be used instead of the
        `session.pwd` and will raise the exception
        `exception_to_raise`.
        """
        def new_pwd():
            raise exception_class("")
        return new_pwd

    def _test_with_pwd_error(self, exception_class):
        """
        Test if reusing a child session fails because of
        `child_host._session.pwd` raising an exception of type
        `exception_class`.
        """
        host = test_base.ftp_host_factory()
        # Implicitly create a child session.
        with host.open("/home/older") as _:
            pass
        assert len(host._children) == 1
        # Make sure reusing the previous child session will fail.
        host._children[0]._session.pwd = self._failing_pwd(exception_class)
        # Try to create a new file. Since `pwd` now raises an
        # exception, a new child session should be created.
        with host.open("home/older") as _:
            pass
        assert len(host._children) == 2

    def test_pwd_with_error_temp(self):
        """
        Test if an `error_temp` in `_session.pwd` skips the child
        session.
        """
        self._test_with_pwd_error(ftplib.error_temp)

    def test_pwd_with_error_reply(self):
        """
        Test if an `error_reply` in `_session.pwd` skips the child
        session.
        """
        self._test_with_pwd_error(ftplib.error_reply)

    def test_pwd_with_OSError(self):
        """
        Test if an `OSError` in `_session.pwd` skips the child
        session.
        """
        self._test_with_pwd_error(OSError)

    def test_pwd_with_EOFError(self):
        """
        Test if an `EOFError` in `_session.pwd` skips the child
        session.
        """
        self._test_with_pwd_error(EOFError)