source: test/test_file_transfer.py @ 1848:7cca1ad2711f

Last change on this file since 1848:7cca1ad2711f was 1848:7cca1ad2711f, checked in by Stefan Schwarzer <sschwarzer@…>, 21 months ago
Format source code with Black Format code in `ftputil` and `test` with Black. Don't format files in the `private` or other directories (or least not yet).
File size: 4.3 KB
Line 
1# Copyright (C) 2010-2018, Stefan Schwarzer <sschwarzer@sschwarzer.net>
2# and ftputil contributors (see `doc/contributors.txt`)
3# See the file LICENSE for licensing terms.
4
5import io
6import random
7
8import pytest
9
10import ftputil.file_transfer
11import ftputil.stat
12
13
14class MockFile:
15    """Class compatible with `LocalFile` and `RemoteFile`."""
16
17    def __init__(self, mtime, mtime_precision):
18        self._mtime = mtime
19        self._mtime_precision = mtime_precision
20
21    def mtime(self):
22        return self._mtime
23
24    def mtime_precision(self):
25        return self._mtime_precision
26
27
28class TestTimestampComparison:
29    def test_source_is_newer_than_target(self):
30        """
31        Test whether the source is newer than the target, i. e. if the
32        file should be transferred.
33        """
34        # Define some time units/precisions.
35        second = 1.0
36        minute = 60.0 * second
37        hour = 60 * minute
38        day = 24 * hour
39        unknown = ftputil.stat.UNKNOWN_PRECISION
40        # Define input arguments; modification datetimes are in seconds.
41        # Fields are source datetime/precision, target datetime/precision,
42        # expected comparison result.
43        file_data = [
44            # Non-overlapping modification datetimes/precisions
45            (1000.0, second, 900.0, second, True),
46            (900.0, second, 1000.0, second, False),
47            # Equal modification datetimes/precisions (if in doubt, transfer)
48            (1000.0, second, 1000.0, second, True),
49            # Just touching intervals
50            (1000.0, second, 1000.0 + second, minute, True),
51            (1000.0 + second, minute, 1000.0, second, True),
52            # Other overlapping intervals
53            (10000.0 - 0.5 * hour, hour, 10000.0, day, True),
54            (10000.0 + 0.5 * hour, hour, 10000.0, day, True),
55            (10000.0 + 0.2 * hour, 0.2 * hour, 10000.0, hour, True),
56            (10000.0 - 0.2 * hour, 2 * hour, 10000.0, hour, True),
57            # Unknown precision
58            (1000.0, None, 1000.0, second, True),
59            (1000.0, second, 1000.0, None, True),
60            (1000.0, None, 1000.0, None, True),
61        ]
62        for (
63            source_mtime,
64            source_mtime_precision,
65            target_mtime,
66            target_mtime_precision,
67            expected_result,
68        ) in file_data:
69            source_file = MockFile(source_mtime, source_mtime_precision)
70            target_file = MockFile(target_mtime, target_mtime_precision)
71            result = ftputil.file_transfer.source_is_newer_than_target(
72                source_file, target_file
73            )
74            assert result == expected_result
75
76
77class FailingStringIO(io.BytesIO):
78    """Mock class to test whether exceptions are passed on."""
79
80    # Kind of nonsense; we just want to see this exception raised.
81    expected_exception = IndexError
82
83    def read(self, count):
84        raise self.expected_exception
85
86
87class TestChunkwiseTransfer:
88    def _random_string(self, count):
89        """Return a `BytesIO` object containing `count` "random" bytes."""
90        ints = (random.randint(0, 255) for i in range(count))
91        return bytes(ints)
92
93    def test_chunkwise_transfer_without_remainder(self):
94        """Check if we get four chunks with 256 Bytes each."""
95        data = self._random_string(1024)
96        fobj = io.BytesIO(data)
97        chunks = list(ftputil.file_transfer.chunks(fobj, 256))
98        assert len(chunks) == 4
99        assert chunks[0] == data[:256]
100        assert chunks[1] == data[256:512]
101        assert chunks[2] == data[512:768]
102        assert chunks[3] == data[768:1024]
103
104    def test_chunkwise_transfer_with_remainder(self):
105        """Check if we get three chunks with 256 Bytes and one with 253."""
106        data = self._random_string(1021)
107        fobj = io.BytesIO(data)
108        chunks = list(ftputil.file_transfer.chunks(fobj, 256))
109        assert len(chunks) == 4
110        assert chunks[0] == data[:256]
111        assert chunks[1] == data[256:512]
112        assert chunks[2] == data[512:768]
113        assert chunks[3] == data[768:1021]
114
115    def test_chunkwise_transfer_with_exception(self):
116        """Check if we see the exception raised during reading."""
117        data = self._random_string(1024)
118        fobj = FailingStringIO(data)
119        iterator = ftputil.file_transfer.chunks(fobj, 256)
120        with pytest.raises(FailingStringIO.expected_exception):
121            next(iterator)
Note: See TracBrowser for help on using the repository browser.