source: test/test_file_transfer.py @ 1713:f146a1ea66aa

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