Changeset 1936:eea2502c2676


Ignore:
Timestamp:
May 24, 2020, 10:54:36 PM (15 months ago)
Author:
Stefan Schwarzer <sschwarzer@…>
Branch:
default
Message:
Move `TestTimeShift` up

Move `TestTimeShift` up before `TestUploadAndDownload`. This order
reflects the order of the method groups in `FTPHost`.
File:
1 edited

Legend:

Unmodified
Added
Removed
  • test/test_host.py

    r1935 r1936  
    256256            files = host.listdir(host.curdir)
    257257            assert files == ["bin", "dev", "etc", "pub", "usr"]
    258 
    259 
    260 class TestUploadAndDownload:
    261     """
    262     Test upload and download.
    263     """
    264 
    265     def test_download(self, tmp_path):
    266         """
    267         Test mode download.
    268         """
    269         remote_file_name = "dummy_name"
    270         remote_file_content = b"dummy_content"
    271         local_target = tmp_path / "test_target"
    272         host_script = [Call("__init__"), Call("pwd", result="/"), Call("close")]
    273         file_script = [
    274             Call("__init__"),
    275             Call("pwd", result="/"),
    276             Call("cwd", args=("/",)),
    277             Call("voidcmd", args=("TYPE I",)),
    278             Call(
    279                 "transfercmd",
    280                 args=("RETR {}".format(remote_file_name), None),
    281                 result=io.BytesIO(remote_file_content),
    282             ),
    283             Call("voidresp"),
    284             Call("close"),
    285         ]
    286         multisession_factory = scripted_session.factory(host_script, file_script)
    287         # Download
    288         with test_base.ftp_host_factory(multisession_factory) as host:
    289             host.download(remote_file_name, str(local_target))
    290         # Verify expected operations on mock socket as done in `FTPFile.close`.
    291         # We expect one `gettimeout` and two `settimeout` calls.
    292         file_session = multisession_factory.scripted_sessions[1]
    293         file_session.sock.gettimeout.assert_called_once_with()
    294         assert len(file_session.sock.settimeout.call_args_list) == 2
    295         assert file_session.sock.settimeout.call_args_list[0] == (
    296             (ftputil.file.FTPFile._close_timeout,),
    297             {},
    298         )
    299         assert file_session.sock.settimeout.call_args_list[1] == (
    300             (file_session.sock.gettimeout(),),
    301             {},
    302         )
    303         assert local_target.read_bytes() == remote_file_content
    304 
    305     def test_conditional_upload_without_upload(self, tmp_path):
    306         """
    307         If the target file is newer, no upload should happen.
    308         """
    309         local_source = tmp_path / "test_source"
    310         data = binary_data()
    311         local_source.write_bytes(data)
    312         dir_result = test_base.dir_line(
    313             mode_string="-rw-r--r--",
    314             date_=datetime.date.today() + datetime.timedelta(days=1),
    315             name="newer",
    316         )
    317         script = [
    318             Call("__init__"),
    319             Call("pwd", result="/"),
    320             Call("cwd", args=("/",)),
    321             Call("cwd", args=("/",)),
    322             Call("dir", args=("",), result=dir_result),
    323             Call("cwd", args=("/",)),
    324             Call("close"),
    325         ]
    326         # Target is newer, so don't upload.
    327         #
    328         # This not only tests the return value, but also if a transfer
    329         # happened. If an upload was tried, our test framework would complain
    330         # about a missing scripted session for the `FTPFile` host.
    331         multisession_factory = scripted_session.factory(script)
    332         with test_base.ftp_host_factory(multisession_factory) as host:
    333             flag = host.upload_if_newer(str(local_source), "/newer")
    334         assert flag is False
    335 
    336     def test_conditional_upload_with_upload(self, tmp_path):
    337         """
    338         If the target file is older or doesn't exist, the source file should be
    339         uploaded.
    340         """
    341         local_source = tmp_path / "test_source"
    342         file_content = b"dummy_content"
    343         local_source.write_bytes(file_content)
    344         remote_file_name = "dummy_name"
    345         dir_result = test_base.dir_line(
    346             mode_string="-rw-r--r--",
    347             date_=datetime.date.today() - datetime.timedelta(days=1),
    348             name="older",
    349         )
    350         host_script = [
    351             Call("__init__"),
    352             Call("pwd", result="/"),
    353             Call("cwd", args=("/",)),
    354             Call("cwd", args=("/",)),
    355             Call("dir", args=("",), result=dir_result),
    356             Call("cwd", args=("/",)),
    357             Call("close"),
    358         ]
    359         file_script = [
    360             Call("__init__"),
    361             Call("pwd", result="/"),
    362             Call("cwd", args=("/",)),
    363             Call("voidcmd", args=("TYPE I",)),
    364             Call(
    365                 "transfercmd",
    366                 args=("STOR older", None),
    367                 result=test_base.MockableBytesIO(),
    368             ),
    369             Call("voidresp"),
    370             Call("close"),
    371         ]
    372         # Target is older, so upload.
    373         multisession_factory = scripted_session.factory(host_script, file_script)
    374         with unittest.mock.patch("test.test_base.MockableBytesIO.write") as write_mock:
    375             with test_base.ftp_host_factory(multisession_factory) as host:
    376                 flag = host.upload_if_newer(str(local_source), "/older")
    377             write_mock.assert_called_with(file_content)
    378         assert flag is True
    379         # Target doesn't exist, so upload.
    380         #  Use correct file name for this test.
    381         file_script[4] = Call(
    382             "transfercmd",
    383             args=("STOR notthere", None),
    384             result=test_base.MockableBytesIO(),
    385         )
    386         multisession_factory = scripted_session.factory(host_script, file_script)
    387         with unittest.mock.patch("test.test_base.MockableBytesIO.write") as write_mock:
    388             with test_base.ftp_host_factory(multisession_factory) as host:
    389                 flag = host.upload_if_newer(str(local_source), "/notthere")
    390             write_mock.assert_called_with(file_content)
    391         assert flag is True
    392 
    393     def test_conditional_download_without_target(self, tmp_path):
    394         """
    395         Test conditional binary mode download when no target file exists.
    396         """
    397         local_target = tmp_path / "test_target"
    398         data = binary_data()
    399         # Target does not exist, so download.
    400         #  There isn't a `dir` call to compare the datetimes of the remote and
    401         #  the target file because the local `exists` call for the local target
    402         #  returns `False` and the datetime comparison therefore isn't done.
    403         host_script = [Call("__init__"), Call("pwd", result="/"), Call("close")]
    404         file_script = [
    405             Call("__init__"),
    406             Call("pwd", result="/"),
    407             Call("cwd", args=("/",)),
    408             Call("voidcmd", args=("TYPE I",)),
    409             Call("transfercmd", args=("RETR newer", None), result=io.BytesIO(data)),
    410             Call("voidresp"),
    411             Call("close"),
    412         ]
    413         multisession_factory = scripted_session.factory(host_script, file_script)
    414         with test_base.ftp_host_factory(multisession_factory) as host:
    415             flag = host.download_if_newer("/newer", str(local_target))
    416         assert flag is True
    417         assert local_target.read_bytes() == data
    418 
    419     def test_conditional_download_with_older_target(self, tmp_path):
    420         """
    421         Test conditional binary mode download with newer source file.
    422         """
    423         local_target = tmp_path / "test_target"
    424         # Make sure file exists for the timestamp comparison.
    425         local_target.touch()
    426         data = binary_data()
    427         # Target is older, so download.
    428         #  Use a date in the future. That isn't realistic, but for the purpose
    429         #  of the test it's an easy way to make sure the source file is newer
    430         #  than the target file.
    431         dir_result = test_base.dir_line(
    432             mode_string="-rw-r--r--",
    433             date_=datetime.date.today() + datetime.timedelta(days=1),
    434             name="newer",
    435         )
    436         host_script = [
    437             Call("__init__"),
    438             Call("pwd", result="/"),
    439             Call("cwd", args=("/",)),
    440             Call("cwd", args=("/",)),
    441             Call("dir", args=("",), result=dir_result),
    442             Call("cwd", args=("/",)),
    443             Call("close"),
    444         ]
    445         file_script = [
    446             Call("__init__"),
    447             Call("pwd", result="/"),
    448             Call("cwd", args=("/",)),
    449             Call("voidcmd", args=("TYPE I",)),
    450             Call("transfercmd", args=("RETR newer", None), result=io.BytesIO(data)),
    451             Call("voidresp"),
    452             Call("close"),
    453         ]
    454         multisession_factory = scripted_session.factory(host_script, file_script)
    455         with test_base.ftp_host_factory(multisession_factory) as host:
    456             flag = host.download_if_newer("/newer", str(local_target))
    457         assert flag is True
    458         assert local_target.read_bytes() == data
    459 
    460     def test_conditional_download_with_newer_target(self, tmp_path):
    461         """
    462         Test conditional binary mode download with older source file.
    463         """
    464         local_target = tmp_path / "test_target"
    465         # Make sure file exists for timestamp comparison.
    466         local_target.touch()
    467         data = binary_data()
    468         # Use date in the past, so the target file is newer and no download
    469         # happens.
    470         dir_result = test_base.dir_line(
    471             mode_string="-rw-r--r--",
    472             date_=datetime.date.today() - datetime.timedelta(days=1),
    473             name="newer",
    474         )
    475         host_script = [
    476             Call("__init__"),
    477             Call("pwd", result="/"),
    478             Call("cwd", args=("/",)),
    479             Call("cwd", args=("/",)),
    480             Call("dir", args=("",), result=dir_result),
    481             Call("cwd", args=("/",)),
    482             Call("close"),
    483         ]
    484         file_script = [
    485             Call("__init__"),
    486             Call("pwd", result="/"),
    487             Call("cwd", args=("/",)),
    488             Call("voidcmd", args=("TYPE I",)),
    489             Call("transfercmd", args=("RETR newer", None), result=io.BytesIO(data)),
    490             Call("voidresp"),
    491             Call("close"),
    492         ]
    493         multisession_factory = scripted_session.factory(host_script, file_script)
    494         with test_base.ftp_host_factory(multisession_factory) as host:
    495             flag = host.download_if_newer("/newer", str(local_target))
    496         assert flag is False
    497258
    498259
     
    689450
    690451
     452class TestUploadAndDownload:
     453    """
     454    Test upload and download.
     455    """
     456
     457    def test_download(self, tmp_path):
     458        """
     459        Test mode download.
     460        """
     461        remote_file_name = "dummy_name"
     462        remote_file_content = b"dummy_content"
     463        local_target = tmp_path / "test_target"
     464        host_script = [Call("__init__"), Call("pwd", result="/"), Call("close")]
     465        file_script = [
     466            Call("__init__"),
     467            Call("pwd", result="/"),
     468            Call("cwd", args=("/",)),
     469            Call("voidcmd", args=("TYPE I",)),
     470            Call(
     471                "transfercmd",
     472                args=("RETR {}".format(remote_file_name), None),
     473                result=io.BytesIO(remote_file_content),
     474            ),
     475            Call("voidresp"),
     476            Call("close"),
     477        ]
     478        multisession_factory = scripted_session.factory(host_script, file_script)
     479        # Download
     480        with test_base.ftp_host_factory(multisession_factory) as host:
     481            host.download(remote_file_name, str(local_target))
     482        # Verify expected operations on mock socket as done in `FTPFile.close`.
     483        # We expect one `gettimeout` and two `settimeout` calls.
     484        file_session = multisession_factory.scripted_sessions[1]
     485        file_session.sock.gettimeout.assert_called_once_with()
     486        assert len(file_session.sock.settimeout.call_args_list) == 2
     487        assert file_session.sock.settimeout.call_args_list[0] == (
     488            (ftputil.file.FTPFile._close_timeout,),
     489            {},
     490        )
     491        assert file_session.sock.settimeout.call_args_list[1] == (
     492            (file_session.sock.gettimeout(),),
     493            {},
     494        )
     495        assert local_target.read_bytes() == remote_file_content
     496
     497    def test_conditional_upload_without_upload(self, tmp_path):
     498        """
     499        If the target file is newer, no upload should happen.
     500        """
     501        local_source = tmp_path / "test_source"
     502        data = binary_data()
     503        local_source.write_bytes(data)
     504        dir_result = test_base.dir_line(
     505            mode_string="-rw-r--r--",
     506            date_=datetime.date.today() + datetime.timedelta(days=1),
     507            name="newer",
     508        )
     509        script = [
     510            Call("__init__"),
     511            Call("pwd", result="/"),
     512            Call("cwd", args=("/",)),
     513            Call("cwd", args=("/",)),
     514            Call("dir", args=("",), result=dir_result),
     515            Call("cwd", args=("/",)),
     516            Call("close"),
     517        ]
     518        # Target is newer, so don't upload.
     519        #
     520        # This not only tests the return value, but also if a transfer
     521        # happened. If an upload was tried, our test framework would complain
     522        # about a missing scripted session for the `FTPFile` host.
     523        multisession_factory = scripted_session.factory(script)
     524        with test_base.ftp_host_factory(multisession_factory) as host:
     525            flag = host.upload_if_newer(str(local_source), "/newer")
     526        assert flag is False
     527
     528    def test_conditional_upload_with_upload(self, tmp_path):
     529        """
     530        If the target file is older or doesn't exist, the source file should be
     531        uploaded.
     532        """
     533        local_source = tmp_path / "test_source"
     534        file_content = b"dummy_content"
     535        local_source.write_bytes(file_content)
     536        remote_file_name = "dummy_name"
     537        dir_result = test_base.dir_line(
     538            mode_string="-rw-r--r--",
     539            date_=datetime.date.today() - datetime.timedelta(days=1),
     540            name="older",
     541        )
     542        host_script = [
     543            Call("__init__"),
     544            Call("pwd", result="/"),
     545            Call("cwd", args=("/",)),
     546            Call("cwd", args=("/",)),
     547            Call("dir", args=("",), result=dir_result),
     548            Call("cwd", args=("/",)),
     549            Call("close"),
     550        ]
     551        file_script = [
     552            Call("__init__"),
     553            Call("pwd", result="/"),
     554            Call("cwd", args=("/",)),
     555            Call("voidcmd", args=("TYPE I",)),
     556            Call(
     557                "transfercmd",
     558                args=("STOR older", None),
     559                result=test_base.MockableBytesIO(),
     560            ),
     561            Call("voidresp"),
     562            Call("close"),
     563        ]
     564        # Target is older, so upload.
     565        multisession_factory = scripted_session.factory(host_script, file_script)
     566        with unittest.mock.patch("test.test_base.MockableBytesIO.write") as write_mock:
     567            with test_base.ftp_host_factory(multisession_factory) as host:
     568                flag = host.upload_if_newer(str(local_source), "/older")
     569            write_mock.assert_called_with(file_content)
     570        assert flag is True
     571        # Target doesn't exist, so upload.
     572        #  Use correct file name for this test.
     573        file_script[4] = Call(
     574            "transfercmd",
     575            args=("STOR notthere", None),
     576            result=test_base.MockableBytesIO(),
     577        )
     578        multisession_factory = scripted_session.factory(host_script, file_script)
     579        with unittest.mock.patch("test.test_base.MockableBytesIO.write") as write_mock:
     580            with test_base.ftp_host_factory(multisession_factory) as host:
     581                flag = host.upload_if_newer(str(local_source), "/notthere")
     582            write_mock.assert_called_with(file_content)
     583        assert flag is True
     584
     585    def test_conditional_download_without_target(self, tmp_path):
     586        """
     587        Test conditional binary mode download when no target file exists.
     588        """
     589        local_target = tmp_path / "test_target"
     590        data = binary_data()
     591        # Target does not exist, so download.
     592        #  There isn't a `dir` call to compare the datetimes of the remote and
     593        #  the target file because the local `exists` call for the local target
     594        #  returns `False` and the datetime comparison therefore isn't done.
     595        host_script = [Call("__init__"), Call("pwd", result="/"), Call("close")]
     596        file_script = [
     597            Call("__init__"),
     598            Call("pwd", result="/"),
     599            Call("cwd", args=("/",)),
     600            Call("voidcmd", args=("TYPE I",)),
     601            Call("transfercmd", args=("RETR newer", None), result=io.BytesIO(data)),
     602            Call("voidresp"),
     603            Call("close"),
     604        ]
     605        multisession_factory = scripted_session.factory(host_script, file_script)
     606        with test_base.ftp_host_factory(multisession_factory) as host:
     607            flag = host.download_if_newer("/newer", str(local_target))
     608        assert flag is True
     609        assert local_target.read_bytes() == data
     610
     611    def test_conditional_download_with_older_target(self, tmp_path):
     612        """
     613        Test conditional binary mode download with newer source file.
     614        """
     615        local_target = tmp_path / "test_target"
     616        # Make sure file exists for the timestamp comparison.
     617        local_target.touch()
     618        data = binary_data()
     619        # Target is older, so download.
     620        #  Use a date in the future. That isn't realistic, but for the purpose
     621        #  of the test it's an easy way to make sure the source file is newer
     622        #  than the target file.
     623        dir_result = test_base.dir_line(
     624            mode_string="-rw-r--r--",
     625            date_=datetime.date.today() + datetime.timedelta(days=1),
     626            name="newer",
     627        )
     628        host_script = [
     629            Call("__init__"),
     630            Call("pwd", result="/"),
     631            Call("cwd", args=("/",)),
     632            Call("cwd", args=("/",)),
     633            Call("dir", args=("",), result=dir_result),
     634            Call("cwd", args=("/",)),
     635            Call("close"),
     636        ]
     637        file_script = [
     638            Call("__init__"),
     639            Call("pwd", result="/"),
     640            Call("cwd", args=("/",)),
     641            Call("voidcmd", args=("TYPE I",)),
     642            Call("transfercmd", args=("RETR newer", None), result=io.BytesIO(data)),
     643            Call("voidresp"),
     644            Call("close"),
     645        ]
     646        multisession_factory = scripted_session.factory(host_script, file_script)
     647        with test_base.ftp_host_factory(multisession_factory) as host:
     648            flag = host.download_if_newer("/newer", str(local_target))
     649        assert flag is True
     650        assert local_target.read_bytes() == data
     651
     652    def test_conditional_download_with_newer_target(self, tmp_path):
     653        """
     654        Test conditional binary mode download with older source file.
     655        """
     656        local_target = tmp_path / "test_target"
     657        # Make sure file exists for timestamp comparison.
     658        local_target.touch()
     659        data = binary_data()
     660        # Use date in the past, so the target file is newer and no download
     661        # happens.
     662        dir_result = test_base.dir_line(
     663            mode_string="-rw-r--r--",
     664            date_=datetime.date.today() - datetime.timedelta(days=1),
     665            name="newer",
     666        )
     667        host_script = [
     668            Call("__init__"),
     669            Call("pwd", result="/"),
     670            Call("cwd", args=("/",)),
     671            Call("cwd", args=("/",)),
     672            Call("dir", args=("",), result=dir_result),
     673            Call("cwd", args=("/",)),
     674            Call("close"),
     675        ]
     676        file_script = [
     677            Call("__init__"),
     678            Call("pwd", result="/"),
     679            Call("cwd", args=("/",)),
     680            Call("voidcmd", args=("TYPE I",)),
     681            Call("transfercmd", args=("RETR newer", None), result=io.BytesIO(data)),
     682            Call("voidresp"),
     683            Call("close"),
     684        ]
     685        multisession_factory = scripted_session.factory(host_script, file_script)
     686        with test_base.ftp_host_factory(multisession_factory) as host:
     687            flag = host.download_if_newer("/newer", str(local_target))
     688        assert flag is False
     689
     690
    691691class TestAcceptEitherUnicodeOrBytes:
    692692    """
Note: See TracChangeset for help on using the changeset viewer.