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
|
# encoding: UTF-8
# Copyright (C) 2003-2016, Stefan Schwarzer <sschwarzer@sschwarzer.net>
# and ftputil contributors (see `doc/contributors.txt`)
# See the file LICENSE for licensing terms.
# Execute tests on a real FTP server (other tests use a mock server).
#
# This test writes some files and directories on the local client and
# the remote server. You'll need write access in the login directory.
# This test can take a few minutes because it has to wait to test the
# timezone calculation.
from __future__ import absolute_import
from __future__ import unicode_literals
import ftplib
import functools
import gc
import operator
import os
import time
import stat
import pytest
import ftputil.compat
import ftputil.error
import ftputil.file_transfer
import ftputil.session
import ftputil.stat_cache
import test
def utc_local_time_shift():
"""
Return the expected time shift in seconds assuming the server
uses UTC in its listings and the client uses local time.
This is needed because Pure-FTPd meanwhile seems to insist that
the displayed time for files is in UTC.
"""
utc_tuple = time.gmtime()
localtime_tuple = time.localtime()
# To calculate the correct times shift, we need to ignore the
# DST component in the localtime tuple, i. e. set it to 0.
localtime_tuple = localtime_tuple[:-1] + (0,)
time_shift_in_seconds = (time.mktime(utc_tuple) -
time.mktime(localtime_tuple))
# To be safe, round the above value to units of 3600 s (1 hour).
return round(time_shift_in_seconds / 3600.0) * 3600
# Difference between local times of server and client. If 0.0, server
# and client use the same timezone.
#EXPECTED_TIME_SHIFT = utc_local_time_shift()
# Pure-FTPd seems to have changed its mind (see docstring of
# `utc_local_time_shift`).
EXPECTED_TIME_SHIFT = 0.0
class Cleaner(object):
"""
This class helps remove directories and files which might
otherwise be left behind if a test fails in unexpected ways.
"""
def __init__(self, host):
# The test class (probably `RealFTPTest`) and the helper
# class share the same `FTPHost` object.
self._host = host
self._ftp_items = []
def add_dir(self, path):
"""Schedule a directory with path `path` for removal."""
self._ftp_items.append(("d", self._host.path.abspath(path)))
def add_file(self, path):
"""Schedule a file with path `path` for removal."""
self._ftp_items.append(("f", self._host.path.abspath(path)))
def clean(self):
"""
Remove the directories and files previously remembered.
The removal works in reverse order of the scheduling with
`add_dir` and `add_file`.
Errors due to a removal are ignored.
"""
self._host.chdir("/")
for type_, path in reversed(self._ftp_items):
try:
if type_ == "d":
# If something goes wrong in `rmtree` we might
# leave a mess behind.
self._host.rmtree(path)
elif type_ == "f":
# Minor mess if `remove` fails
self._host.remove(path)
except ftputil.error.FTPError:
pass
REMOTE_FILE_NAME = "CONTENTS"
host = self.host
# Implicitly create child host object.
with host.open(REMOTE_FILE_NAME, "rb") as file_obj1:
pass
# Monkey-patch file to simulate an FTP server timeout below.
def timed_out_pwd():
raise ftplib.error_reply("delayed 226 reply")
file_obj1._host._session.pwd = timed_out_pwd
# Try to get a file - which shouldn't be the timed-out file.
with host.open(REMOTE_FILE_NAME, "rb") as file_obj2:
assert file_obj1 is not file_obj2
# Re-use closed and not timed-out child session.
with host.open(REMOTE_FILE_NAME, "rb") as file_obj3:
pass
assert file_obj2 is file_obj3
|
