source: test/test_file_transfer.py

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