Changeset 1825:eeddeb17229c


Ignore:
Timestamp:
Jun 30, 2019, 8:41:03 PM (4 months ago)
Author:
Stefan Schwarzer <sschwarzer@…>
Branch:
default
rebase_source:
a1957fbe57642fc2fef33bb27ab1a0b9e05d27b7
Message:
Use `ScriptedSession` in `TestFileOperations`
File:
1 edited

Legend:

Unmodified
Added
Removed

  • test/test_file.py

    r1823 r1825  
    1 # Copyright (C) 2002-2018, Stefan Schwarzer <sschwarzer@sschwarzer.net>
     1# Copyright (C) 2002-2019, Stefan Schwarzer <sschwarzer@sschwarzer.net>
    22# and ftputil contributors (see `doc/contributors.txt`)
    33# See the file LICENSE for licensing terms.
    44
    55import ftplib
     6import io
     7import unittest
    68
    79import pytest
    … …  
    1012
    1113from test import mock_ftplib
     14from test import scripted_session
    1215from test import test_base
     16
     17
     18Call = scripted_session.Call
    1319
    1420
    … …  
    5662    def test_inaccessible_dir(self):
    5763        """Test whether opening a file at an invalid location fails."""
    58         host = test_base.ftp_host_factory(
    59                  session_factory=InaccessibleDirSession)
    60         with pytest.raises(ftputil.error.FTPIOError):
    61             host.open("/inaccessible/new_file", "w")
    62 
    63     def test_caching(self):
     64        host_script = [
     65          Call("__init__"),
     66          Call("pwd", result="/"),
     67          Call("close")
     68        ]
     69        file_script = [
     70          Call("__init__"),
     71          Call("pwd", result="/"),
     72          Call("cwd", args=("/",)),
     73          Call("voidcmd", args=("TYPE I",)),
     74          Call("transfercmd", args=("STOR inaccessible", None), result=ftplib.error_perm),
     75          Call("close")
     76        ]
     77        multisession_factory = scripted_session.factory(host_script, file_script)
     78        with test_base.ftp_host_factory(multisession_factory) as host:
     79            with pytest.raises(ftputil.error.FTPIOError):
     80                host.open("/inaccessible", "w")
     81
     82    def test_caching_of_children(self):
    6483        """Test whether `FTPFile` cache of `FTPHost` object works."""
    65         host = test_base.ftp_host_factory()
    66         assert len(host._children) == 0
    67         path1 = "path1"
    68         path2 = "path2"
    69         # Open one file and inspect cache.
    70         file1 = host.open(path1, "w")
    71         child1 = host._children[0]
    72         assert len(host._children) == 1
    73         assert not child1._file.closed
    74         # Open another file.
    75         file2 = host.open(path2, "w")
    76         child2 = host._children[1]
    77         assert len(host._children) == 2
    78         assert not child2._file.closed
    79         # Close first file.
    80         file1.close()
    81         assert len(host._children) == 2
    82         assert child1._file.closed
    83         assert not child2._file.closed
    84         # Re-open first child's file.
    85         file1 = host.open(path1, "w")
    86         child1_1 = file1._host
    87         # Check if it's reused.
    88         assert child1 is child1_1
    89         assert not child1._file.closed
    90         assert not child2._file.closed
    91         # Close second file.
    92         file2.close()
    93         assert child2._file.closed
     84        host_script = [
     85          Call("__init__"),
     86          Call("pwd", result="/"),
     87          Call("close"),
     88        ]
     89        file1_script = [
     90          Call("__init__"),
     91          Call("pwd", result="/"),
     92          Call("cwd", args=("/",)),
     93          Call("voidcmd", args=("TYPE I",)),
     94          Call("transfercmd", args=("STOR path1", None), result=io.StringIO("")),
     95          Call("voidresp"),
     96          # Open a file again while reusing the child object and with it its
     97          # `_session` attribute (the `ftplib.FTP` object).
     98          Call("pwd", result="/"),
     99          Call("cwd", args=("/",)),
     100          Call("voidcmd", args=("TYPE I",)),
     101          Call("transfercmd", args=("STOR path1", None), result=io.StringIO("")),
     102          Call("voidresp"),
     103          Call("close")
     104        ]
     105        file2_script = [
     106          Call("__init__"),
     107          Call("pwd", result="/"),
     108          Call("cwd", args=("/",)),
     109          Call("voidcmd", args=("TYPE I",)),
     110          Call("transfercmd",
     111               result=io.StringIO(""),
     112               args=("STOR path2", None)),
     113          Call("voidresp"),
     114          Call("close")
     115        ]
     116        multisession_factory = scripted_session.factory(host_script,
     117                                                        file1_script, file2_script)
     118        with test_base.ftp_host_factory(multisession_factory) as host:
     119            assert len(host._children) == 0
     120            path1 = "path1"
     121            path2 = "path2"
     122            # Open one file and inspect cache of available children.
     123            file1 = host.open(path1, "w")
     124            child1 = host._children[0]
     125            assert len(host._children) == 1
     126            assert not child1._file.closed
     127            # Open another file.
     128            file2 = host.open(path2, "w")
     129            child2 = host._children[1]
     130            assert len(host._children) == 2
     131            assert not child2._file.closed
     132            # Close first file.
     133            file1.close()
     134            assert len(host._children) == 2
     135            assert child1._file.closed
     136            assert not child2._file.closed
     137            # Re-open first child's file.
     138            file1 = host.open(path1, "w")
     139            child1_1 = file1._host
     140            # Check if it's reused.
     141            assert child1 is child1_1
     142            assert not child1._file.closed
     143            assert not child2._file.closed
     144            # Close second file.
     145            file2.close()
     146            assert child2._file.closed
    94147
    95148    def test_write_to_directory(self):
    96149        """Test whether attempting to write to a directory fails."""
    97         host = test_base.ftp_host_factory()
    98         with pytest.raises(ftputil.error.FTPIOError):
    99             host.open("/home/sschwarzer", "w")
     150        host_script = [
     151          Call("__init__"),
     152          Call("pwd", result="/"),
     153          Call("close")
     154        ]
     155        file_script = [
     156          Call("__init__"),
     157          Call("pwd", result="/"),
     158          Call("cwd", args=("/",)),
     159          Call("voidcmd", args=("TYPE I",)),
     160          Call("transfercmd",
     161               args=("STOR some_directory", None),
     162               result=ftplib.error_perm),
     163          # Because of the exception, `voidresp` isn't called.
     164          Call("close")
     165        ]
     166        multisession_factory = scripted_session.factory(host_script, file_script)
     167        with test_base.ftp_host_factory(multisession_factory) as host:
     168            with pytest.raises(ftputil.error.FTPIOError):
     169                host.open("/some_directory", "w")
    100170
    101171    def test_binary_read(self):
    102172        """Read data from a binary file."""
    103         host = test_base.ftp_host_factory(session_factory=ReadMockSession)
    104         with host.open("some_file", "rb") as fobj:
    105             data = fobj.read()
    106         assert data == ReadMockSession.mock_file_content
     173        host_script = [
     174          Call("__init__"),
     175          Call("pwd", result="/"),
     176          Call("close")
     177        ]
     178        file_content = b"some\ntest"
     179        file_script = [
     180          Call("__init__"),
     181          Call("pwd", result="/"),
     182          Call("cwd", args=("/",)),
     183          Call("voidcmd", args=("TYPE I",)),
     184          Call("transfercmd",
     185               args=("RETR some_file", None),
     186               result=io.BytesIO(file_content)),
     187          Call("voidresp"),
     188          Call("close")
     189        ]
     190        multisession_factory = scripted_session.factory(host_script, file_script)
     191        with test_base.ftp_host_factory(multisession_factory) as host:
     192            with host.open("some_file", "rb") as fobj:
     193                data = fobj.read()
     194            assert data == file_content
    107195
    108196    def test_binary_write(self):
    109197        """Write binary data with `write`."""
    110         host = test_base.ftp_host_factory()
     198        host_script = [
     199          Call("__init__"),
     200          Call("pwd", result="/"),
     201          Call("close")
     202        ]
     203        file_script = [
     204          Call("__init__"),
     205          Call("pwd", result="/"),
     206          Call("cwd", args=("/",)),
     207          Call("voidcmd", args=("TYPE I",)),
     208          Call("transfercmd",
     209               args=("STOR some_file", None),
     210               result=test_base.MockableBytesIO(b"")),
     211          Call("voidresp"),
     212          Call("close")
     213        ]
    111214        data = b"\000a\001b\r\n\002c\003\n\004\r\005"
    112         with host.open("dummy", "wb") as output:
    113             output.write(data)
    114         child_data = mock_ftplib.content_of("dummy")
    115         expected_data = data
    116         assert child_data == expected_data
     215        multisession_factory = scripted_session.factory(host_script, file_script)
     216        with unittest.mock.patch("test.test_base.MockableBytesIO.write") as write_mock:
     217            with test_base.ftp_host_factory(multisession_factory) as host:
     218                with host.open("some_file", "wb") as output:
     219                    output.write(data)
     220            write_mock.assert_called_with(data)
    117221
    118222    def test_ascii_read(self):
    119223        """Read ASCII text with plain `read`."""
    120         host = test_base.ftp_host_factory(session_factory=ReadMockSession)
    121         with host.open("dummy", "r") as input_:
    122             data = input_.read(0)
    123             assert data == ""
    124             data = input_.read(3)
    125             assert data == "lin"
    126             data = input_.read(7)
    127             assert data == "e 1\nano"
    128             data = input_.read()
    129             assert data == "ther line\nyet another line"
    130             data = input_.read()
    131             assert data == ""
     224        host_script = [
     225          Call("__init__"),
     226          Call("pwd", result="/",),
     227          Call("close")
     228        ]
     229        file_script = [
     230          Call("__init__"),
     231          Call("pwd", result="/",),
     232          Call("cwd", args=("/",)),
     233          Call("voidcmd", args=('TYPE I',)),
     234          Call("transfercmd",
     235               args=('RETR dummy', None),
     236               # Since the file is eventually opened as a text file
     237               # (wrapped in a `TextIOWrapper`), line endings should
     238               # be converted.
     239               result=io.BytesIO(b"line 1\r\nanother line\nyet another line")),
     240          Call("voidresp"),
     241          Call("close")
     242        ]
     243        multisession_factory = scripted_session.factory(host_script, file_script)
     244        with test_base.ftp_host_factory(multisession_factory) as host:
     245            with host.open("dummy", "r") as input_:
     246                data = input_.read(0)
     247                assert data == ""
     248                data = input_.read(3)
     249                assert data == "lin"
     250                # Specifically check the behavior around the line ending
     251                # character.
     252                data = input_.read(3)
     253                assert data == "e 1"
     254                data = input_.read(1)
     255                assert data == "\n"
     256                data = input_.read()
     257                assert data == "another line\nyet another line"
     258                data = input_.read()
     259                assert data == ""
    132260
    133261    def test_ascii_write(self):
    134         """Write ASCII text with `write`."""
    135         host = test_base.ftp_host_factory()
    136         data = " \nline 2\nline 3"
    137         with host.open("dummy", "w", newline="\r\n") as output:
    138             output.write(data)
    139         child_data = mock_ftplib.content_of("dummy")
    140         # This corresponds to the byte stream, so expect a `bytes` object.
    141         expected_data = b" \r\nline 2\r\nline 3"
    142         assert child_data == expected_data
     262        """Write text with `write`."""
     263        host_script = [
     264          Call("__init__"),
     265          Call("pwd", result="/"),
     266          Call("close")
     267        ]
     268        file_script = [
     269          Call("__init__"),
     270          Call("pwd", result="/"),
     271          Call("cwd", args=("/",)),
     272          Call("voidcmd", args=('TYPE I',)),
     273          Call("transfercmd",
     274               args=('STOR dummy', None),
     275               result=test_base.MockableBytesIO()),
     276          Call("voidresp"),
     277          Call("close")
     278        ]
     279        input_data = " \nline 2\nline 3"
     280        multisession_factory = scripted_session.factory(host_script, file_script)
     281        with unittest.mock.patch("test.test_base.MockableBytesIO.write") as write_mock:
     282            with test_base.ftp_host_factory(multisession_factory) as host:
     283                with host.open("dummy", "w", newline="\r\n") as output:
     284                    output.write(input_data)
     285        write_mock.assert_called_with(b" \r\nline 2\r\nline 3")
    143286
    144287    # TODO: Add tests with given encoding and possibly buffering.
    … …  
    146289    def test_ascii_writelines(self):
    147290        """Write ASCII text with `writelines`."""
    148         host = test_base.ftp_host_factory()
     291        host_script = [
     292          Call("__init__"),
     293          Call("pwd", result="/"),
     294          Call("close")
     295        ]
     296        file_script = [
     297          Call("__init__"),
     298          Call("pwd", result="/"),
     299          Call("cwd", args=("/",)),
     300          Call("voidcmd", args=('TYPE I',)),
     301          Call("transfercmd",
     302               args=('STOR dummy', None),
     303               result=test_base.MockableBytesIO()),
     304          Call("voidresp"),
     305          Call("close")
     306        ]
    149307        data = [" \n", "line 2\n", "line 3"]
    150308        backup_data = data[:]
    151         output = host.open("dummy", "w", newline="\r\n")
    152         output.writelines(data)
    153         output.close()
    154         child_data = mock_ftplib.content_of("dummy")
    155         expected_data = b" \r\nline 2\r\nline 3"
    156         assert child_data == expected_data
     309        multisession_factory = scripted_session.factory(host_script, file_script)
     310        with unittest.mock.patch("test.test_base.MockableBytesIO.write") as write_mock:
     311            with test_base.ftp_host_factory(multisession_factory) as host:
     312                with host.open("dummy", "w", newline="\r\n") as output:
     313                    output.writelines(data)
     314        write_mock.assert_called_with(b" \r\nline 2\r\nline 3")
    157315        # Ensure that the original data was not modified.
    158316        assert data == backup_data
    … …  
    160318    def test_binary_readline(self):
    161319        """Read binary data with `readline`."""
    162         host = test_base.ftp_host_factory(session_factory=ReadMockSession)
    163         input_ = host.open("dummy", "rb")
    164         data = input_.readline(3)
    165         assert data == b"lin"
    166         data = input_.readline(10)
    167         assert data == b"e 1\r\n"
    168         data = input_.readline(13)
    169         assert data == b"another line\r"
    170         data = input_.readline()
    171         assert data == b"\n"
    172         data = input_.readline()
    173         assert data == b"yet another line"
    174         data = input_.readline()
    175         assert data == b""
    176         input_.close()
     320        host_script = [
     321          Call("__init__"),
     322          Call("pwd", result="/"),
     323          Call("close")
     324        ]
     325        file_script = [
     326          Call("__init__"),
     327          Call("pwd", result="/"),
     328          Call("cwd", args=("/",)),
     329          Call("voidcmd", args=('TYPE I',)),
     330          Call("transfercmd",
     331               args=('RETR dummy', None),
     332               result=io.BytesIO(b"line 1\r\nanother line\r\nyet another line")),
     333          Call("voidresp"),
     334          Call("close")
     335        ]
     336        multisession_factory = scripted_session.factory(host_script, file_script)
     337        with test_base.ftp_host_factory(multisession_factory) as host:
     338            with host.open("dummy", "rb") as input_:
     339                data = input_.readline(3)
     340                assert data == b"lin"
     341                data = input_.readline(10)
     342                assert data == b"e 1\r\n"
     343                data = input_.readline(13)
     344                assert data == b"another line\r"
     345                data = input_.readline()
     346                assert data == b"\n"
     347                data = input_.readline()
     348                assert data == b"yet another line"
     349                data = input_.readline()
     350                assert data == b""
    177351
    178352    def test_ascii_readline(self):
    179353        """Read ASCII text with `readline`."""
    180         host = test_base.ftp_host_factory(session_factory=ReadMockSession)
    181         input_ = host.open("dummy", "r")
    182         data = input_.readline(3)
    183         assert data == "lin"
    184         data = input_.readline(10)
    185         assert data == "e 1\n"
    186         data = input_.readline(13)
    187         assert data == "another line\n"
    188         data = input_.readline()
    189         assert data == "yet another line"
    190         data = input_.readline()
    191         assert data == ""
    192         input_.close()
     354        host_script = [
     355          Call("__init__"),
     356          Call("pwd", result="/"),
     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("transfercmd",
     365               args=('RETR dummy', None),
     366               result=io.BytesIO(b"line 1\r\nanother line\r\nyet another line")),
     367          Call("voidresp"),
     368          Call("close")
     369        ]
     370        multisession_factory = scripted_session.factory(host_script, file_script)
     371        with test_base.ftp_host_factory(multisession_factory) as host:
     372            with host.open("dummy", "r") as input_:
     373                data = input_.readline(3)
     374                assert data == "lin"
     375                data = input_.readline(10)
     376                assert data == "e 1\n"
     377                data = input_.readline(13)
     378                assert data == "another line\n"
     379                data = input_.readline()
     380                assert data == "yet another line"
     381                data = input_.readline()
     382                assert data == ""
    193383
    194384    def test_ascii_readlines(self):
    195385        """Read ASCII text with `readlines`."""
    196         host = test_base.ftp_host_factory(session_factory=ReadMockSession)
    197         input_ = host.open("dummy", "r")
    198         data = input_.read(3)
    199         assert data == "lin"
    200         data = input_.readlines()
    201         assert data == ["e 1\n", "another line\n", "yet another line"]
    202         input_.close()
     386        host_script = [
     387          Call("__init__"),
     388          Call("pwd", result="/"),
     389          Call("close")
     390        ]
     391        file_script = [
     392          Call("__init__"),
     393          Call("pwd", result="/"),
     394          Call("cwd", args=("/",)),
     395          Call("voidcmd", args=('TYPE I',)),
     396          Call("transfercmd",
     397               args=('RETR dummy', None),
     398               result=io.BytesIO(b"line 1\r\nanother line\r\nyet another line")),
     399          Call("voidresp"),
     400          Call("close")
     401        ]
     402        multisession_factory = scripted_session.factory(host_script, file_script)
     403        with test_base.ftp_host_factory(multisession_factory) as host:
     404            with host.open("dummy", "r") as input_:
     405                data = input_.read(3)
     406                assert data == "lin"
     407                data = input_.readlines()
     408                assert data == ["e 1\n", "another line\n", "yet another line"]
     409                input_.close()
    203410
    204411    def test_binary_iterator(self):
    … …  
    207414        newline conversion.
    208415        """
    209         host = test_base.ftp_host_factory(session_factory=ReadMockSession)
    210         input_ = host.open("dummy", "rb")
    211         input_iterator = iter(input_)
    212         assert next(input_iterator) == b"line 1\r\n"
    213         assert next(input_iterator) == b"another line\r\n"
    214         assert next(input_iterator) == b"yet another line"
    215         with pytest.raises(StopIteration):
    216             input_iterator.__next__()
    217         input_.close()
     416        host_script = [
     417          Call("__init__"),
     418          Call("pwd", result="/"),
     419          Call("close")
     420        ]
     421        file_script = [
     422          Call("__init__"),
     423          Call("pwd", result="/"),
     424          Call("cwd", args=("/",)),
     425          Call("voidcmd", args=('TYPE I',)),
     426          Call("transfercmd",
     427               args=('RETR dummy', None),
     428               result=io.BytesIO(b"line 1\r\nanother line\r\nyet another line")),
     429          Call("voidresp"),
     430          Call("close")
     431        ]
     432        multisession_factory = scripted_session.factory(host_script, file_script)
     433        with test_base.ftp_host_factory(multisession_factory) as host:
     434            with host.open("dummy", "rb") as input_:
     435                input_iterator = iter(input_)
     436                assert next(input_iterator) == b"line 1\r\n"
     437                assert next(input_iterator) == b"another line\r\n"
     438                assert next(input_iterator) == b"yet another line"
     439                with pytest.raises(StopIteration):
     440                    input_iterator.__next__()
    218441
    219442    def test_ascii_iterator(self):
    … …  
    222445        conversion).
    223446        """
    224         host = test_base.ftp_host_factory(session_factory=ReadMockSession)
    225         input_ = host.open("dummy")
    226         input_iterator = iter(input_)
    227         assert next(input_iterator) == "line 1\n"
    228         assert next(input_iterator) == "another line\n"
    229         assert next(input_iterator) == "yet another line"
    230         with pytest.raises(StopIteration):
    231             input_iterator.__next__()
    232         input_.close()
     447        host_script = [
     448          Call("__init__"),
     449          Call("pwd", result="/"),
     450          Call("close")
     451        ]
     452        file_script = [
     453          Call("__init__"),
     454          Call("pwd", result="/"),
     455          Call("cwd", args=("/",)),
     456          Call("voidcmd", args=('TYPE I',)),
     457          Call("transfercmd",
     458               args=('RETR dummy', None),
     459               result=io.BytesIO(b"line 1\r\nanother line\r\nyet another line")),
     460          Call("voidresp"),
     461          Call("close")
     462        ]
     463        multisession_factory = scripted_session.factory(host_script, file_script)
     464        with test_base.ftp_host_factory(multisession_factory) as host:
     465            with host.open("dummy", "r") as input_:
     466                input_iterator = iter(input_)
     467                assert next(input_iterator) == "line 1\n"
     468                assert next(input_iterator) == "another line\n"
     469                assert next(input_iterator) == "yet another line"
     470                with pytest.raises(StopIteration):
     471                    input_iterator.__next__()
    233472
    234473    def test_read_unknown_file(self):
    235474        """Test whether reading a file which isn't there fails."""
    236         host = test_base.ftp_host_factory()
    237         with pytest.raises(ftputil.error.FTPIOError):
    238             host.open("notthere", "r")
     475        host_script = [
     476          Call("__init__"),
     477          Call("pwd", result="/"),
     478          Call("close")
     479        ]
     480        file_script = [
     481          Call("__init__"),
     482          Call("pwd", result="/"),
     483          Call("cwd", args=("/",)),
     484          Call("voidcmd", args=('TYPE I',)),
     485          Call("transfercmd",
     486               args=('RETR notthere', None),
     487               result=ftplib.error_perm),
     488          Call("close")
     489        ]
     490        multisession_factory = scripted_session.factory(host_script, file_script)
     491        with test_base.ftp_host_factory(multisession_factory) as host:
     492            with pytest.raises(ftputil.error.FTPIOError):
     493                with host.open("notthere", "r") as input_:
     494                    pass
    239495
    240496
Note: See TracChangeset for help on using the changeset viewer.