Changeset 1825:eeddeb17229c
- Timestamp:
- Jun 30, 2019, 8:41:03 PM (20 months ago)
- Branch:
- default
- rebase_source:
- a1957fbe57642fc2fef33bb27ab1a0b9e05d27b7
- File:
-
- 1 edited
Legend:
- Unmodified
- Added
- Removed
-
test/test_file.py
r1823 r1825 1 # Copyright (C) 2002-201 8, Stefan Schwarzer <sschwarzer@sschwarzer.net>1 # Copyright (C) 2002-2019, Stefan Schwarzer <sschwarzer@sschwarzer.net> 2 2 # and ftputil contributors (see `doc/contributors.txt`) 3 3 # See the file LICENSE for licensing terms. 4 4 5 5 import ftplib 6 import io 7 import unittest 6 8 7 9 import pytest … … 10 12 11 13 from test import mock_ftplib 14 from test import scripted_session 12 15 from test import test_base 16 17 18 Call = scripted_session.Call 13 19 14 20 … … 56 62 def test_inaccessible_dir(self): 57 63 """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): 64 83 """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 94 147 95 148 def test_write_to_directory(self): 96 149 """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") 100 170 101 171 def test_binary_read(self): 102 172 """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 107 195 108 196 def test_binary_write(self): 109 197 """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 ] 111 214 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) 117 221 118 222 def test_ascii_read(self): 119 223 """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 == "" 132 260 133 261 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") 143 286 144 287 # TODO: Add tests with given encoding and possibly buffering. … … 146 289 def test_ascii_writelines(self): 147 290 """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 ] 149 307 data = [" \n", "line 2\n", "line 3"] 150 308 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_data309 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") 157 315 # Ensure that the original data was not modified. 158 316 assert data == backup_data … … 160 318 def test_binary_readline(self): 161 319 """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"" 177 351 178 352 def test_ascii_readline(self): 179 353 """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 == "" 193 383 194 384 def test_ascii_readlines(self): 195 385 """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() 203 410 204 411 def test_binary_iterator(self): … … 207 414 newline conversion. 208 415 """ 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__() 218 441 219 442 def test_ascii_iterator(self): … … 222 445 conversion). 223 446 """ 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__() 233 472 234 473 def test_read_unknown_file(self): 235 474 """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 239 495 240 496
Note: See TracChangeset
for help on using the changeset viewer.