source: test/test_file.py @ 1721:3557f65ded13

Last change on this file since 1721:3557f65ded13 was 1721:3557f65ded13, checked in by Stefan Schwarzer <sschwarzer@…>, 7 months ago
Remove `compat.py` This module was for Python 2/3 compatibility, similar to the `six` package. Since the next version of ftputil will only support Python 3, hardcode the types from `compat.py` in the code that used to use the `compat` module. There may still be redundant code that isn't needed when running the tests under Python 3.
File size: 9.6 KB
Line 
1# Copyright (C) 2002-2018, Stefan Schwarzer <sschwarzer@sschwarzer.net>
2# and ftputil contributors (see `doc/contributors.txt`)
3# See the file LICENSE for licensing terms.
4
5import ftplib
6
7import pytest
8
9import ftputil.error
10
11from test import mock_ftplib
12from test import test_base
13
14
15#
16# Several customized `MockSession` classes
17#
18class ReadMockSession(mock_ftplib.MockSession):
19
20    mock_file_content = b"line 1\r\nanother line\r\nyet another line"
21
22
23class ReadMockSessionWithMoreNewlines(mock_ftplib.MockSession):
24
25    mock_file_content = b"\r\n".join(map(bytes, range(20)))
26
27
28class InaccessibleDirSession(mock_ftplib.MockSession):
29
30    _login_dir = "/inaccessible"
31
32    def pwd(self):
33        return self._login_dir
34
35    def cwd(self, dir):
36        if dir in (self._login_dir, self._login_dir + "/"):
37            raise ftplib.error_perm
38        else:
39            super(InaccessibleDirSession, self).cwd(dir)
40
41
42class TestFileOperations:
43    """Test operations with file-like objects."""
44
45    def test_inaccessible_dir(self):
46        """Test whether opening a file at an invalid location fails."""
47        host = test_base.ftp_host_factory(
48                 session_factory=InaccessibleDirSession)
49        with pytest.raises(ftputil.error.FTPIOError):
50            host.open("/inaccessible/new_file", "w")
51
52    def test_caching(self):
53        """Test whether `FTPFile` cache of `FTPHost` object works."""
54        host = test_base.ftp_host_factory()
55        assert len(host._children) == 0
56        path1 = "path1"
57        path2 = "path2"
58        # Open one file and inspect cache.
59        file1 = host.open(path1, "w")
60        child1 = host._children[0]
61        assert len(host._children) == 1
62        assert not child1._file.closed
63        # Open another file.
64        file2 = host.open(path2, "w")
65        child2 = host._children[1]
66        assert len(host._children) == 2
67        assert not child2._file.closed
68        # Close first file.
69        file1.close()
70        assert len(host._children) == 2
71        assert child1._file.closed
72        assert not child2._file.closed
73        # Re-open first child's file.
74        file1 = host.open(path1, "w")
75        child1_1 = file1._host
76        # Check if it's reused.
77        assert child1 is child1_1
78        assert not child1._file.closed
79        assert not child2._file.closed
80        # Close second file.
81        file2.close()
82        assert child2._file.closed
83
84    def test_write_to_directory(self):
85        """Test whether attempting to write to a directory fails."""
86        host = test_base.ftp_host_factory()
87        with pytest.raises(ftputil.error.FTPIOError):
88            host.open("/home/sschwarzer", "w")
89
90    def test_binary_read(self):
91        """Read data from a binary file."""
92        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
93        with host.open("some_file", "rb") as fobj:
94            data = fobj.read()
95        assert data == ReadMockSession.mock_file_content
96
97    def test_binary_write(self):
98        """Write binary data with `write`."""
99        host = test_base.ftp_host_factory()
100        data = b"\000a\001b\r\n\002c\003\n\004\r\005"
101        with host.open("dummy", "wb") as output:
102            output.write(data)
103        child_data = mock_ftplib.content_of("dummy")
104        expected_data = data
105        assert child_data == expected_data
106
107    def test_ascii_read(self):
108        """Read ASCII text with plain `read`."""
109        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
110        with host.open("dummy", "r") as input_:
111            data = input_.read(0)
112            assert data == ""
113            data = input_.read(3)
114            assert data == "lin"
115            data = input_.read(7)
116            assert data == "e 1\nano"
117            data = input_.read()
118            assert data == "ther line\nyet another line"
119            data = input_.read()
120            assert data == ""
121
122    def test_ascii_write(self):
123        """Write ASCII text with `write`."""
124        host = test_base.ftp_host_factory()
125        data = " \nline 2\nline 3"
126        with host.open("dummy", "w", newline="\r\n") as output:
127            output.write(data)
128        child_data = mock_ftplib.content_of("dummy")
129        # This corresponds to the byte stream, so expect a `bytes` object.
130        expected_data = b" \r\nline 2\r\nline 3"
131        assert child_data == expected_data
132
133    # TODO: Add tests with given encoding and possibly buffering.
134
135    def test_ascii_writelines(self):
136        """Write ASCII text with `writelines`."""
137        host = test_base.ftp_host_factory()
138        data = [" \n", "line 2\n", "line 3"]
139        backup_data = data[:]
140        output = host.open("dummy", "w", newline="\r\n")
141        output.writelines(data)
142        output.close()
143        child_data = mock_ftplib.content_of("dummy")
144        expected_data = b" \r\nline 2\r\nline 3"
145        assert child_data == expected_data
146        # Ensure that the original data was not modified.
147        assert data == backup_data
148
149    def test_binary_readline(self):
150        """Read binary data with `readline`."""
151        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
152        input_ = host.open("dummy", "rb")
153        data = input_.readline(3)
154        assert data == b"lin"
155        data = input_.readline(10)
156        assert data == b"e 1\r\n"
157        data = input_.readline(13)
158        assert data == b"another line\r"
159        data = input_.readline()
160        assert data == b"\n"
161        data = input_.readline()
162        assert data == b"yet another line"
163        data = input_.readline()
164        assert data == b""
165        input_.close()
166
167    def test_ascii_readline(self):
168        """Read ASCII text with `readline`."""
169        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
170        input_ = host.open("dummy", "r")
171        data = input_.readline(3)
172        assert data == "lin"
173        data = input_.readline(10)
174        assert data == "e 1\n"
175        data = input_.readline(13)
176        assert data == "another line\n"
177        data = input_.readline()
178        assert data == "yet another line"
179        data = input_.readline()
180        assert data == ""
181        input_.close()
182
183    def test_ascii_readlines(self):
184        """Read ASCII text with `readlines`."""
185        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
186        input_ = host.open("dummy", "r")
187        data = input_.read(3)
188        assert data == "lin"
189        data = input_.readlines()
190        assert data == ["e 1\n", "another line\n", "yet another line"]
191        input_.close()
192
193    def test_binary_iterator(self):
194        """
195        Test the iterator interface of `FTPFile` objects (without
196        newline conversion.
197        """
198        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
199        input_ = host.open("dummy", "rb")
200        input_iterator = iter(input_)
201        assert next(input_iterator) == b"line 1\r\n"
202        assert next(input_iterator) == b"another line\r\n"
203        assert next(input_iterator) == b"yet another line"
204        with pytest.raises(StopIteration):
205            input_iterator.__next__()
206        input_.close()
207
208    def test_ascii_iterator(self):
209        """
210        Test the iterator interface of `FTPFile` objects (with newline
211        conversion).
212        """
213        host = test_base.ftp_host_factory(session_factory=ReadMockSession)
214        input_ = host.open("dummy")
215        input_iterator = iter(input_)
216        assert next(input_iterator) == "line 1\n"
217        assert next(input_iterator) == "another line\n"
218        assert next(input_iterator) == "yet another line"
219        with pytest.raises(StopIteration):
220            input_iterator.__next__()
221        input_.close()
222
223    def test_read_unknown_file(self):
224        """Test whether reading a file which isn't there fails."""
225        host = test_base.ftp_host_factory()
226        with pytest.raises(ftputil.error.FTPIOError):
227            host.open("notthere", "r")
228
229
230class TestAvailableChild:
231
232    def _failing_pwd(self, exception_class):
233        """
234        Return a function that will be used instead of the
235        `session.pwd` and will raise the exception
236        `exception_to_raise`.
237        """
238        def new_pwd():
239            raise exception_class("")
240        return new_pwd
241
242    def _test_with_pwd_error(self, exception_class):
243        """
244        Test if reusing a child session fails because of
245        `child_host._session.pwd` raising an exception of type
246        `exception_class`.
247        """
248        host = test_base.ftp_host_factory()
249        # Implicitly create a child session.
250        with host.open("/home/older") as _:
251            pass
252        assert len(host._children) == 1
253        # Make sure reusing the previous child session will fail.
254        host._children[0]._session.pwd = self._failing_pwd(exception_class)
255        # Try to create a new file. Since `pwd` now raises an
256        # exception, a new child session should be created.
257        with host.open("home/older") as _:
258            pass
259        assert len(host._children) == 2
260
261    def test_pwd_with_error_temp(self):
262        """
263        Test if an `error_temp` in `_session.pwd` skips the child
264        session.
265        """
266        self._test_with_pwd_error(ftplib.error_temp)
267
268    def test_pwd_with_error_reply(self):
269        """
270        Test if an `error_reply` in `_session.pwd` skips the child
271        session.
272        """
273        self._test_with_pwd_error(ftplib.error_reply)
274
275    def test_pwd_with_OSError(self):
276        """
277        Test if an `OSError` in `_session.pwd` skips the child
278        session.
279        """
280        self._test_with_pwd_error(OSError)
281
282    def test_pwd_with_EOFError(self):
283        """
284        Test if an `EOFError` in `_session.pwd` skips the child
285        session.
286        """
287        self._test_with_pwd_error(EOFError)
Note: See TracBrowser for help on using the repository browser.