Python ftplib 模块，Error() 实例源码
我们从Python开源项目中，提取了以下33个代码示例，用于说明如何使用ftplib.Error()。
def test_stor(self):
try:
data = b'abcde12345' * 100000
self.dummy_sendfile.write(data)
self.dummy_sendfile.seek(0)
self.client.storbinary('stor ' + TESTFN, self.dummy_sendfile)
self.client.retrbinary('retr ' + TESTFN, self.dummy_recvfile.write)
self.dummy_recvfile.seek(0)
datafile = self.dummy_recvfile.read()
self.assertEqual(len(data), len(datafile))
self.assertEqual(hash(data), hash(datafile))
finally:
# We do not use os.remove() because file could still be
# locked by ftpd thread. If DELE through FTP fails try
# os.remove() as last resort.
if os.path.exists(TESTFN):
try:
self.client.delete(TESTFN)
except (ftplib.Error, EOFError, socket.error):
safe_remove(TESTFN)
def test_all_errors(self):
exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
ftplib.error_proto, ftplib.Error, IOError, EOFError)
for x in exceptions:
try:
raise x('exception not included in all_errors set')
except ftplib.all_errors:
pass
def test_all_errors(self):
exceptions = (ftplib.error_reply, EOFError)
for x in exceptions:
try:
raise x('exception not included in all_errors set')
except ftplib.all_errors:
pass
def test_line_too_long(self):
self.assertRaises(ftplib.Error, self.client.sendcmd,
'x' * self.client.maxline * 2)
def test_retrlines_too_long(self):
self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2))
received = []
self.assertRaises(ftplib.Error,
self.client.retrlines, 'retr', received.append)
def test_storlines_too_long(self):
f = StringIO.StringIO('x' * self.client.maxline * 2)
self.assertRaises(ftplib.Error, self.client.storlines, 'stor', f)
def test_all_errors(self):
exceptions = (ftplib.error_reply, EOFError)
for x in exceptions:
try:
raise x('exception not included in all_errors set')
except ftplib.all_errors:
pass
def test_line_too_long(self):
self.assertRaises(ftplib.Error,
'x' * self.client.maxline * 2)
def test_retrlines_too_long(self):
self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2))
received = []
self.assertRaises(ftplib.Error, received.append)
def test_storlines_too_long(self):
f = StringIO.StringIO('x' * self.client.maxline * 2)
self.assertRaises(ftplib.Error, f)
def test_all_errors(self):
exceptions = (ftplib.error_reply, EOFError)
for x in exceptions:
try:
raise x('exception not included in all_errors set')
except ftplib.all_errors:
pass
def test_line_too_long(self):
self.assertRaises(ftplib.Error,
'x' * self.client.maxline * 2)
def test_retrlines_too_long(self):
self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2))
received = []
self.assertRaises(ftplib.Error, received.append)
def test_storlines_too_long(self):
f = io.BytesIO(b'x' * self.client.maxline * 2)
self.assertRaises(ftplib.Error, f)
def test_all_errors(self):
exceptions = (ftplib.error_reply, EOFError)
for x in exceptions:
try:
raise x('exception not included in all_errors set')
except ftplib.all_errors:
pass
def test_line_too_long(self):
self.assertRaises(ftplib.Error,
'x' * self.client.maxline * 2)
def test_retrlines_too_long(self):
self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2))
received = []
self.assertRaises(ftplib.Error, received.append)
def test_storlines_too_long(self):
f = StringIO.StringIO('x' * self.client.maxline * 2)
self.assertRaises(ftplib.Error, f)
def test_all_errors(self):
exceptions = (ftplib.error_reply, OSError, EOFError)
for x in exceptions:
try:
raise x('exception not included in all_errors set')
except ftplib.all_errors:
pass
def test_line_too_long(self):
self.assertRaises(ftplib.Error,
'x' * self.client.maxline * 2)
def test_retrlines_too_long(self):
self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2))
received = []
self.assertRaises(ftplib.Error, received.append)
def test_storlines_too_long(self):
f = io.BytesIO(b'x' * self.client.maxline * 2)
self.assertRaises(ftplib.Error, f)
def test_all_errors(self):
exceptions = (ftplib.error_reply, EOFError)
for x in exceptions:
try:
raise x('exception not included in all_errors set')
except ftplib.all_errors:
pass
def test_stor_ascii(self):
# Test STOR in ASCII mode
def store(cmd, fp, blocksize=8192):
# like storbinary() except it sends "type a" instead of
# "type i" before starting the transfer
self.client.voidcmd('type a')
with contextlib.closing(self.client.transfercmd(cmd)) as conn:
while True:
buf = fp.read(blocksize)
if not buf:
break
conn.sendall(buf)
return self.client.voidresp()
try:
data = b'abcde12345\r\n' * 100000
self.dummy_sendfile.write(data)
self.dummy_sendfile.seek(0)
store('stor ' + TESTFN, self.dummy_recvfile.write)
expected = data.replace(b'\r\n', b(os.linesep))
self.dummy_recvfile.seek(0)
datafile = self.dummy_recvfile.read()
self.assertEqual(len(expected), len(datafile))
self.assertEqual(hash(expected), socket.error):
safe_remove(TESTFN)
def test_stor_ascii_2(self):
# Test that no extra extra carriage returns are added to the
# file in ASCII mode in case CRLF gets truncated in two chunks
# (issue 116)
def store(cmd, blocksize=8192):
# like storbinary() except it sends "type a" instead of
# "type i" before starting the transfer
self.client.voidcmd('type a')
with contextlib.closing(self.client.transfercmd(cmd)) as conn:
while True:
buf = fp.read(blocksize)
if not buf:
break
conn.sendall(buf)
return self.client.voidresp()
old_buffer = DTPHandler.ac_in_buffer_size
try:
# set a small buffer so that CRLF gets delivered in two
# separate chunks: "CRLF"," f","oo"," CR","LF"," b","ar"
DTPHandler.ac_in_buffer_size = 2
data = b'\r\n foo \r\n bar'
self.dummy_sendfile.write(data)
self.dummy_sendfile.seek(0)
store('stor ' + TESTFN, self.dummy_sendfile)
expected = data.replace(b'\r\n', b(os.linesep))
self.client.retrbinary('retr ' + TESTFN, self.dummy_recvfile.write)
self.dummy_recvfile.seek(0)
self.assertEqual(expected, self.dummy_recvfile.read())
finally:
DTPHandler.ac_in_buffer_size = old_buffer
# We do not use os.remove() because file could still be
# locked by ftpd thread. If DELE through FTP fails try
# os.remove() as last resort.
if os.path.exists(TESTFN):
try:
self.client.delete(TESTFN)
except (ftplib.Error, socket.error):
safe_remove(TESTFN)
def test_stou(self):
data = b'abcde12345' * 100000
self.dummy_sendfile.write(data)
self.dummy_sendfile.seek(0)
self.client.voidcmd('TYPE I')
# filename comes in as "1xx FILE: <filename>"
filename = self.client.sendcmd('stou').split('FILE: ')[1]
try:
with contextlib.closing(self.client.makeport()) as sock:
conn, sockaddr = sock.accept()
with contextlib.closing(conn):
conn.settimeout(TIMEOUT)
if hasattr(self.client_class, 'ssl_version'):
conn = ssl.wrap_socket(conn)
while True:
buf = self.dummy_sendfile.read(8192)
if not buf:
break
conn.sendall(buf)
# transfer finished,a 226 response is expected
self.assertEqual('226', self.client.voidresp()[:3])
self.client.retrbinary('retr ' + filename,
self.dummy_recvfile.write)
self.dummy_recvfile.seek(0)
datafile = self.dummy_recvfile.read()
self.assertEqual(len(data), hash(datafile))
finally:
# We do not use os.remove() because file could still be
# locked by ftpd thread. If DELE through FTP fails try
# os.remove() as last resort.
if os.path.exists(filename):
try:
self.client.delete(filename)
except (ftplib.Error, socket.error):
safe_remove(filename)
def test_appe(self):
try:
data1 = b'abcde12345' * 100000
self.dummy_sendfile.write(data1)
self.dummy_sendfile.seek(0)
self.client.storbinary('stor ' + TESTFN, self.dummy_sendfile)
data2 = b'fghil67890' * 100000
self.dummy_sendfile.write(data2)
self.dummy_sendfile.seek(len(data1))
self.client.storbinary('appe ' + TESTFN, self.dummy_sendfile)
self.client.retrbinary("retr " + TESTFN, self.dummy_recvfile.write)
self.dummy_recvfile.seek(0)
datafile = self.dummy_recvfile.read()
self.assertEqual(len(data1 + data2), len(datafile))
self.assertEqual(hash(data1 + data2), socket.error):
safe_remove(TESTFN)
def test_abor_during_transfer(self):
# Case 4: ABOR while a data transfer on DTP channel is in
# progress: close data channel,respond with 426,respond
# with 226.
data = b'abcde12345' * 1000000
with open(TESTFN, 'w+b') as f:
f.write(data)
try:
self.client.voidcmd('TYPE I')
with contextlib.closing(
self.client.transfercmd('retr ' + TESTFN)) as conn:
bytes_recv = 0
while bytes_recv < 65536:
chunk = conn.recv(BUFSIZE)
bytes_recv += len(chunk)
# stop transfer while it isn't finished yet
self.client.putcmd('ABOR')
# transfer isn't finished yet so ftpd should respond with 426
self.assertEqual(self.client.getline()[:3], "426")
# transfer successfully aborted,so should now respond
# with a 226
self.assertEqual('226', self.client.voidresp()[:3])
finally:
# We do not use os.remove() because file could still be
# locked by ftpd thread. If DELE through FTP fails try
# os.remove() as last resort.
try:
self.client.delete(TESTFN)
except (ftplib.Error, socket.error):
safe_remove(TESTFN)
def cmdresp(self, cmd):
"""Send a command and return response,also if the command failed."""
try:
return self.client.sendcmd(cmd)
except ftplib.Error as err:
return str(err)
def test_all_errors(self):
exceptions = (ftplib.error_reply, EOFError)
for x in exceptions:
try:
raise x('exception not included in all_errors set')
except ftplib.all_errors:
pass
def test_line_too_long(self):
self.assertRaises(ftplib.Error,
'x' * self.client.maxline * 2)
def test_retrlines_too_long(self):
self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2))
received = []
self.assertRaises(ftplib.Error, received.append)
def test_storlines_too_long(self):
f = io.BytesIO(b'x' * self.client.maxline * 2)
self.assertRaises(ftplib.Error, f)
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。