source: test/test_file_transfer.py @ 1663:40280d912474

Last change on this file since 1663:40280d912474 was 1663:40280d912474, checked in by Stefan Schwarzer <sschwarzer@…>, 3 years ago
Remove `unittest` dependency - Remove `unittest` imports - Let test case classes inherit from `object` instead of `unittest.TestCase` - Use `setup_method` and `teardown_method` instead of `unittest`'s `setUp` and `tearDown` - Use `pytest.mark.skipif` instead of `unittest.skipIf` - Use `pytest.mark.slow_test` instead of decorator in `test/__init__.py`. Exclude slow tests with py.test -m "not slow_test" test - Replace some leftover `assert*` calls with `assert` statements - Replace `assert False, ...` with `pytest.fail` ticket: 98
File size: 4.4 KB
Line 
1# Copyright (C) 2010-2016, Stefan Schwarzer <sschwarzer@sschwarzer.net>
2# and ftputil contributors (see `doc/contributors.txt`)
3# See the file LICENSE for licensing terms.
4
5from __future__ import unicode_literals
6
7import io
8import random
9
10import pytest
11
12import ftputil.compat
13import ftputil.file_transfer
14import ftputil.stat
15
16
17class MockFile(object):
18    """Class compatible with `LocalFile` and `RemoteFile`."""
19
20    def __init__(self, mtime, mtime_precision):
21        self._mtime = mtime
22        self._mtime_precision = mtime_precision
23
24    def mtime(self):
25        return self._mtime
26
27    def mtime_precision(self):
28        return self._mtime_precision
29
30
31class TestTimestampComparison(object):
32
33    def test_source_is_newer_than_target(self):
34        """
35        Test whether the source is newer than the target, i. e. if the
36        file should be transferred.
37        """
38        # Define some time units/precisions.
39        second = 1.0
40        minute = 60.0 * second
41        hour = 60 * minute
42        day = 24 * hour
43        unknown = ftputil.stat.UNKNOWN_PRECISION
44        # Define input arguments; modification datetimes are in seconds.
45        # Fields are source datetime/precision, target datetime/precision,
46        # expected comparison result.
47        file_data = [
48          # Non-overlapping modification datetimes/precisions
49          (1000.0, second,  900.0,  second,  True),
50          (900.0,  second,  1000.0, second,  False),
51          # Equal modification datetimes/precisions (if in doubt, transfer)
52          (1000.0, second,  1000.0, second,  True),
53          # Just touching intervals
54          (1000.0,        second,  1000.0+second, minute,  True),
55          (1000.0+second, minute,  1000.0,        second,  True),
56          # Other overlapping intervals
57          (10000.0-0.5*hour, hour,      10000.0, day,   True),
58          (10000.0+0.5*hour, hour,      10000.0, day,   True),
59          (10000.0+0.2*hour, 0.2*hour,  10000.0, hour,  True),
60          (10000.0-0.2*hour, 2*hour,    10000.0, hour,  True),
61          # Unknown precision
62          (1000.0, None,    1000.0, second,  True),
63          (1000.0, second,  1000.0, None,    True),
64          (1000.0, None,    1000.0, None,    True),
65        ]
66        for (source_mtime, source_mtime_precision,
67             target_mtime, target_mtime_precision,
68             expected_result) 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            assert result == expected_result
74
75
76class FailingStringIO(io.BytesIO):
77    """Mock class to test whether exceptions are passed on."""
78
79    # Kind of nonsense; we just want to see this exception raised.
80    expected_exception = IndexError
81
82    def read(self, count):
83        raise self.expected_exception
84
85
86class TestChunkwiseTransfer(object):
87
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 ftputil.compat.bytes_from_ints(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.