source: test/test_file_transfer.py

Last change on this file was 1935:38acf28f3905, checked in by Stefan Schwarzer <sschwarzer@…>, 5 months ago
Format docstrings and comments Reformat docstrings and comments to 80 characters. Traditionally, the margin for the code was at 79 columns, so I chose a bit less for the comments. With the switch to Black for formatting the right margin is now at nearly 90 columns, so it makes sense to also allow a greater right margin for docstrings and comments. Lines still don't get too long because the formatted text often starts only in column 5 or 9.
File size: 5.9 KB
Line 
1# Copyright (C) 2010-2020, Stefan Schwarzer <sschwarzer@sschwarzer.net>
2# and ftputil contributors (see `doc/contributors.txt`)
3# See the file LICENSE for licensing terms.
4
5import datetime
6import io
7import random
8
9import pytest
10
11import ftputil.file_transfer
12import ftputil.stat
13
14from test import test_base
15from test import scripted_session
16
17
18Call = scripted_session.Call
19
20
21class MockFile:
22    """
23    Class compatible with `LocalFile` and `RemoteFile`.
24    """
25
26    def __init__(self, mtime, mtime_precision):
27        self._mtime = mtime
28        self._mtime_precision = mtime_precision
29
30    def mtime(self):
31        return self._mtime
32
33    def mtime_precision(self):
34        return self._mtime_precision
35
36
37class TestRemoteFile:
38    def test_time_shift_subtracted_only_once(self):
39        """
40        Test whether the time shift value is subtracted from the initial server
41        timestamp only once.
42
43        This subtraction happens in `stat._Stat.parse_unix_time`, so it must
44        _not_ be done a second time in `file_transfer.RemoteFile`.
45        """
46        utcnow = datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc)
47        # 3 hours
48        time_shift = 3 * 60 * 60
49        dir_line = test_base.dir_line(
50            datetime_=utcnow + datetime.timedelta(seconds=time_shift), name="dummy_name"
51        )
52        script = [
53            Call("__init__"),
54            Call("pwd", result="/"),
55            Call("cwd", args=("/",)),
56            Call("cwd", args=("/",)),
57            Call("dir", args=("",), result=dir_line),
58            Call("cwd", args=("/",)),
59            Call("close"),
60        ]
61        with test_base.ftp_host_factory(scripted_session.factory(script)) as host:
62            host.set_time_shift(3 * 60 * 60)
63            remote_file = ftputil.file_transfer.RemoteFile(host, "dummy_name", 0o644)
64            remote_mtime = remote_file.mtime()
65        # The remote mtime should be corrected by the time shift, so the
66        # calculated UTC time is the same as for the client. The 60.0 (seconds)
67        # is the timestamp precision.
68        assert remote_mtime <= utcnow.timestamp() <= remote_mtime + 60.0
69
70
71class TestTimestampComparison:
72    def test_source_is_newer_than_target(self):
73        """
74        Test whether the source is newer than the target, i. e. if the
75        file should be transferred.
76        """
77        # Define some time units/precisions.
78        second = 1.0
79        minute = 60 * second
80        hour = 60 * minute
81        day = 24 * hour
82        unknown = ftputil.stat.UNKNOWN_PRECISION
83        # Define input arguments; modification datetimes are in seconds. Fields
84        # are source datetime/precision, target datetime/precision, expected
85        # comparison result.
86        file_data = [
87            # Non-overlapping modification datetimes/precisions
88            (1000.0, second, 900.0, second, True),
89            (900.0, second, 1000.0, second, False),
90            # Equal modification datetimes/precisions (if in doubt, transfer)
91            (1000.0, second, 1000.0, second, True),
92            # Just touching intervals
93            (1000.0, second, 1000.0 + second, minute, True),
94            (1000.0 + second, minute, 1000.0, second, True),
95            # Other overlapping intervals
96            (10000.0 - 0.5 * hour, hour, 10000.0, day, True),
97            (10000.0 + 0.5 * hour, hour, 10000.0, day, True),
98            (10000.0 + 0.2 * hour, 0.2 * hour, 10000.0, hour, True),
99            (10000.0 - 0.2 * hour, 2 * hour, 10000.0, hour, True),
100            # Unknown precision
101            (1000.0, unknown, 1000.0, second, True),
102            (1000.0, second, 1000.0, unknown, True),
103            (1000.0, unknown, 1000.0, unknown, True),
104        ]
105        for (
106            source_mtime,
107            source_mtime_precision,
108            target_mtime,
109            target_mtime_precision,
110            expected_result,
111        ) in file_data:
112            source_file = MockFile(source_mtime, source_mtime_precision)
113            target_file = MockFile(target_mtime, target_mtime_precision)
114            result = ftputil.file_transfer.source_is_newer_than_target(
115                source_file, target_file
116            )
117            assert result == expected_result
118
119
120class FailingStringIO(io.BytesIO):
121    """
122    Mock class to test whether exceptions are passed on.
123    """
124
125    # Kind of nonsense; we just want to see this exception raised.
126    expected_exception = IndexError
127
128    def read(self, count):
129        raise self.expected_exception
130
131
132class TestChunkwiseTransfer:
133    def _random_string(self, count):
134        """
135        Return a `BytesIO` object containing `count` "random" bytes.
136        """
137        ints = (random.randint(0, 255) for i in range(count))
138        return bytes(ints)
139
140    def test_chunkwise_transfer_without_remainder(self):
141        """
142        Check if we get four chunks with 256 Bytes each.
143        """
144        data = self._random_string(1024)
145        fobj = io.BytesIO(data)
146        chunks = list(ftputil.file_transfer.chunks(fobj, 256))
147        assert len(chunks) == 4
148        assert chunks[0] == data[:256]
149        assert chunks[1] == data[256:512]
150        assert chunks[2] == data[512:768]
151        assert chunks[3] == data[768:1024]
152
153    def test_chunkwise_transfer_with_remainder(self):
154        """
155        Check if we get three chunks with 256 Bytes and one with 253.
156        """
157        data = self._random_string(1021)
158        fobj = io.BytesIO(data)
159        chunks = list(ftputil.file_transfer.chunks(fobj, 256))
160        assert len(chunks) == 4
161        assert chunks[0] == data[:256]
162        assert chunks[1] == data[256:512]
163        assert chunks[2] == data[512:768]
164        assert chunks[3] == data[768:1021]
165
166    def test_chunkwise_transfer_with_exception(self):
167        """
168        Check if we see the exception raised during reading.
169        """
170        data = self._random_string(1024)
171        fobj = FailingStringIO(data)
172        iterator = ftputil.file_transfer.chunks(fobj, 256)
173        with pytest.raises(FailingStringIO.expected_exception):
174            next(iterator)
Note: See TracBrowser for help on using the repository browser.