1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
|
# Copyright (C) 2008-2019, Stefan Schwarzer <sschwarzer@sschwarzer.net>
# and ftputil contributors (see `doc/contributors.txt`)
# See the file LICENSE for licensing terms.
import ftplib
import io
import pytest
import ftputil.error
import test.scripted_session as scripted_session
from test import test_base
Call = scripted_session.Call
# Exception raised by client code, i. e. code using ftputil. Used to test the
# behavior in case of client exceptions.
class ClientCodeException(Exception):
pass
#
# Test cases
#
class TestHostContextManager:
def test_normal_operation(self):
"""
If an `FTPHost` instance is created, it should be closed by the host
context manager.
"""
script = [
Call(method_name="__init__"),
Call(method_name="pwd", result="/"),
Call(method_name="close"),
]
with test_base.ftp_host_factory(scripted_session.factory(script)) as host:
assert host.closed is False
assert host.closed is True
def test_ftputil_exception(self):
"""
If an `ftplib.FTP` method raises an exception, it should be caught by
the host context manager and the host object should be closed.
"""
script = [
# Since `__init__` raises an exception, `pwd` isn't called.
# However, `close` is called via the context manager.
Call(method_name="__init__", result=ftplib.error_perm),
Call(method_name="close"),
]
with pytest.raises(ftputil.error.FTPOSError):
with test_base.ftp_host_factory(scripted_session.factory(script)) as host:
pass
# We arrived here, that's fine. Because the `FTPHost` object wasn't
# successfully constructed, the assignment to `host` shouldn't have
# happened.
assert "host" not in locals()
def test_client_code_exception(self):
"""
If client code raises an exception in the context manager block, the
host object should be closed by the context manager.
"""
script = [
Call(method_name="__init__"),
Call(method_name="pwd", result="/"),
Call(method_name="close"),
]
try:
with test_base.ftp_host_factory(scripted_session.factory(script)) as host:
assert host.closed is False
raise ClientCodeException()
except ClientCodeException:
assert host.closed is True
else:
pytest.fail("`ClientCodeException` not raised")
class TestFileContextManager:
def test_normal_operation(self):
"""
If an `FTPFile` object is created in the `FTPFile` context manager, the
context manager should close the file at the end of the `with` block.
"""
host_script = [
Call(method_name="__init__"),
Call(method_name="pwd", result="/"),
Call(method_name="close"),
]
file_script = [
Call(method_name="__init__"),
Call(method_name="pwd", result="/"),
Call(method_name="cwd", result=None, args=("/",)),
Call(method_name="voidcmd", result=None, args=("TYPE I",)),
Call(
method_name="transfercmd",
result=io.StringIO("line 1\nline 2"),
args=("RETR dummy", None),
),
Call(method_name="voidresp", result=None),
Call(method_name="close"),
]
multisession_factory = scripted_session.factory(host_script, file_script)
with test_base.ftp_host_factory(multisession_factory) as host:
with host.open("dummy", "r") as fobj:
assert fobj.closed is False
data = fobj.readline()
assert data == "line 1\n"
assert fobj.closed is False
assert fobj.closed is True
def test_ftputil_exception(self):
"""
If an `ftplib.FTP` method raises an exception, the `FTPFile` context
manager should try to close the file.
"""
host_script = [
Call(method_name="__init__"),
Call(method_name="pwd", result="/"),
Call(method_name="close"),
]
file_script = [
Call(method_name="__init__"),
Call(method_name="pwd", result="/"),
Call(method_name="cwd", result=None, args=("/",)),
Call(method_name="voidcmd", result=None, args=("TYPE I",)),
# Raise exception. `voidresp` therefore won't be called, but
# `close` will be called by the context manager.
Call(
method_name="transfercmd",
result=ftplib.error_perm,
args=("STOR inaccessible", None),
),
# Call(method_name="voidresp", result=None),
Call(method_name="close"),
]
multisession_factory = scripted_session.factory(host_script, file_script)
with test_base.ftp_host_factory(multisession_factory) as host:
with pytest.raises(ftputil.error.FTPIOError):
# This should fail.
with host.open("/inaccessible", "w") as fobj:
pass
# The file construction shouldn't have succeeded, so `fobj` should
# be absent from the local namespace.
assert "fobj" not in locals()
def test_client_code_exception(self):
"""
If client code raises an exception in the `FTPFile` context manager
block, the file object should be closed by the context manager.
"""
host_script = [
Call(method_name="__init__"),
Call(method_name="pwd", result="/"),
Call(method_name="close"),
]
file_script = [
Call(method_name="__init__"),
Call(method_name="pwd", result="/"),
Call(method_name="cwd", result=None, args=("/",)),
Call(method_name="voidcmd", result=None, args=("TYPE I",)),
Call(
method_name="transfercmd",
result=io.BytesIO(b""),
args=("RETR dummy", None),
),
Call(method_name="voidresp", result=None),
Call(method_name="close"),
]
multisession_factory = scripted_session.factory(host_script, file_script)
with test_base.ftp_host_factory(multisession_factory) as host:
try:
with host.open("dummy", "r") as fobj:
assert fobj.closed is False
raise ClientCodeException()
except ClientCodeException:
assert fobj.closed is True
else:
pytest.fail("`ClientCodeException` not raised")
|
