Changeset 1803:555e10d7d302


Ignore:
Timestamp:
Jun 10, 2019, 10:55:46 PM (3 months ago)
Author:
Stefan Schwarzer <sschwarzer@…>
Branch:
default
amend_source:
7713836e79fd036c03ea0e414f8db5b32d331c1d
Message:
Use `ScriptedSession` in `test_synchronize_times`

We no longer use the `TimeShiftFTPHost` class here; instead put
`_Path` helper class in the test method. However, `TimeShiftFTPHost`
is still used in the following test method
`test_synchronize_times_for_server_in_east`, so this needs some
refactoring later.
File:
1 edited

Legend:

Unmodified
Added
Removed
  • test/test_host.py

    r1801 r1803  
    649649    def test_synchronize_times(self):
    650650        """Test time synchronization with server."""
    651         host = test_base.ftp_host_factory(ftp_host_class=TimeShiftFTPHost,
    652                                           session_factory=TimeShiftMockSession)
     651        # Helper mock class that frees us from setting up complicated
     652        # session scripts for the remote calls.
     653        class _Path:
     654            def split(self, path):
     655                return posixpath.split(path)
     656            def set_mtime(self, mtime):
     657                self._mtime = mtime
     658            def getmtime(self, file_name):
     659                return self._mtime
     660            def join(self, *args):
     661                return posixpath.join(*args)
     662            def normpath(self, path):
     663                return posixpath.normpath(path)
     664            def isabs(self, path):
     665                return posixpath.isabs(path)
     666            def abspath(self, path):
     667                return "/_ftputil_sync_"
     668            # Needed for `isdir` in `FTPHost.remove`
     669            def isfile(self, path):
     670                return True
     671        #
     672        Call = scripted_session.Call
     673        host_script = [
     674          Call("__init__"),
     675          Call(method_name="pwd", result="/"),
     676          Call(method_name="cwd", result=None, args=("/",)),
     677          Call(method_name="cwd", result=None, args=("/",)),
     678          Call(method_name="delete", result=None, args=("_ftputil_sync_",)),
     679          Call(method_name="cwd", result=None, args=("/",)),
     680          Call(method_name="close"),
     681        ]
     682        file_script = [
     683          Call("__init__"),
     684          Call(method_name="pwd", result="/"),
     685          Call(method_name="cwd", result=None, args=("/",)),
     686          Call(method_name="voidcmd", result=None, args=("TYPE I",)),
     687          Call(method_name="transfercmd", result=io.BytesIO(),
     688               args=("STOR _ftputil_sync_", None)),
     689          Call(method_name="voidresp", result=None, args=()),
     690          Call(method_name="close"),
     691        ]
    653692        # Valid time shifts
    654693        test_data = [
     
    659698        ]
    660699        for measured_time_shift, expected_time_shift in test_data:
    661             host.path.set_mtime(time.time() + measured_time_shift)
    662             host.synchronize_times()
    663             assert host.time_shift() == expected_time_shift
     700            print("=== measured_time_shift:", measured_time_shift)
     701            print("=== expected_time_shift:", expected_time_shift)
     702            # Use a new `BytesIO` object to avoid exception
     703            # `ValueError: I/O operation on closed file`.
     704            file_script[4] = Call(method_name="transfercmd", result=io.BytesIO(),
     705                                  args=("STOR _ftputil_sync_", None))
     706            multisession_factory = scripted_session.factory(host_script,
     707                                                            file_script)
     708            with test_base.ftp_host_factory(multisession_factory) as host:
     709                host.path = _Path()
     710                host.path.set_mtime(time.time() + measured_time_shift)
     711                host.synchronize_times()
     712                assert host.time_shift() == expected_time_shift
    664713        # Invalid time shifts
    665714        measured_time_shifts = [60*60+8*60, 45*60-6*60]
    666715        for measured_time_shift in measured_time_shifts:
    667             host.path.set_mtime(time.time() + measured_time_shift)
    668             with pytest.raises(ftputil.error.TimeShiftError):
    669                 host.synchronize_times()
     716            print("=== measured_time_shift:", measured_time_shift)
     717            # Use a new `BytesIO` object to avoid exception
     718            # `ValueError: I/O operation on closed file`.
     719            file_script[4] = Call(method_name="transfercmd", result=io.BytesIO(),
     720                                  args=("STOR _ftputil_sync_", None))
     721            multisession_factory = scripted_session.factory(host_script,
     722                                                            file_script)
     723            with test_base.ftp_host_factory(multisession_factory) as host:
     724                host.path = _Path()
     725                host.path.set_mtime(time.time() + measured_time_shift)
     726                with pytest.raises(ftputil.error.TimeShiftError):
     727                    host.synchronize_times()
    670728
    671729    def test_synchronize_times_for_server_in_east(self):
Note: See TracChangeset for help on using the changeset viewer.