mirror of
https://github.com/theupdateframework/python-tuf
synced 2026-05-24 10:08:28 +00:00
Fix test_util.py test case failure in py2.7.
This commit is contained in:
parent
797bab5ddc
commit
21bbbedbb8
3 changed files with 26 additions and 75 deletions
|
|
@ -210,69 +210,6 @@ def tearDown(self):
|
|||
unittest_toolbox.Modified_TestCase.tearDown(self)
|
||||
|
||||
|
||||
"""
|
||||
def test_without_tuf_mode_1(self):
|
||||
# Simulate a slow retrieval attack.
|
||||
# 'mode_1': When download begins,the server blocks the download
|
||||
# for a long time by doing nothing before it sends the first byte of data.
|
||||
|
||||
# Retrieve 'file1.txt' provided by the pre-generated repository.
|
||||
url_prefix = self.repository_mirrors['mirror1']['url_prefix']
|
||||
url_file = os.path.join(url_prefix, 'targets', 'file1.txt')
|
||||
client_filepath = os.path.join(self.client_directory, 'file1.txt')
|
||||
|
||||
# Generate the fileinfo of 'file.txt' to compare it to what is downloaded.
|
||||
# The download should complete, albeit slowly (the slow retrieval server
|
||||
# sets a limit on the delay.)
|
||||
filepath = os.path.join(self.repository_directory, 'targets', 'file1.txt')
|
||||
length, hashes = tuf.util.get_file_details(filepath)
|
||||
fileinfo = tuf.formats.make_fileinfo(length, hashes)
|
||||
|
||||
try:
|
||||
server_process = self._start_slow_server('mode_1')
|
||||
six.moves.urllib.request.urlretrieve(url_file, client_filepath)
|
||||
|
||||
# Verify the expected file size and hash of the downloaded file.
|
||||
length, hashes = tuf.util.get_file_details(client_filepath)
|
||||
download_fileinfo = tuf.formats.make_fileinfo(length, hashes)
|
||||
self.assertEqual(fileinfo, download_fileinfo)
|
||||
|
||||
finally:
|
||||
# Terminate the slow retrieval (mode 1) server.
|
||||
self._stop_slow_server(server_process)
|
||||
|
||||
|
||||
|
||||
def test_without_tuf_mode_2(self):
|
||||
# Simulate a slow retrieval attack.
|
||||
# 'mode_1': When download begins, the server blocks the download for a long
|
||||
# time by doing nothing before it sends the first byte of data.
|
||||
|
||||
url_prefix = self.repository_mirrors['mirror1']['url_prefix']
|
||||
url_file = os.path.join(url_prefix, 'targets', 'file1.txt')
|
||||
client_filepath = os.path.join(self.client_directory, 'file1.txt')
|
||||
|
||||
# Generate the fileinfo of 'file.txt' to compare it to what is downloaded.
|
||||
# The download should complete, albeit slowly (the slow retrieval server
|
||||
# sets a limit on the delay.)
|
||||
filepath = os.path.join(self.repository_directory, 'targets', 'file1.txt')
|
||||
length, hashes = tuf.util.get_file_details(filepath)
|
||||
fileinfo = tuf.formats.make_fileinfo(length, hashes)
|
||||
|
||||
try:
|
||||
server_process = self._start_slow_server('mode_2')
|
||||
six.moves.urllib.request.urlretrieve(url_file, client_filepath)
|
||||
|
||||
# Verify the expected file size and hash of the downloaded file.
|
||||
length, hashes = tuf.util.get_file_details(client_filepath)
|
||||
download_fileinfo = tuf.formats.make_fileinfo(length, hashes)
|
||||
self.assertEqual(fileinfo, download_fileinfo)
|
||||
|
||||
finally:
|
||||
# Terminate the slow retrieval (mode 2) server.
|
||||
self._stop_slow_server(server_process)
|
||||
"""
|
||||
|
||||
|
||||
def test_with_tuf_mode_1(self):
|
||||
# Simulate a slow retrieval attack.
|
||||
|
|
@ -297,6 +234,7 @@ def test_with_tuf_mode_1(self):
|
|||
|
||||
# Verify that 'file1.txt' is the culprit.
|
||||
self.assertEqual(url_file, mirror_url)
|
||||
print(repr(mirror_error))
|
||||
self.assertTrue(isinstance(mirror_error, tuf.DownloadLengthMismatchError))
|
||||
|
||||
else:
|
||||
|
|
@ -330,6 +268,7 @@ def test_with_tuf_mode_2(self):
|
|||
|
||||
# Verify that 'file1.txt' is the culprit.
|
||||
self.assertEqual(url_file, mirror_url)
|
||||
print(repr(mirror_error))
|
||||
self.assertTrue(isinstance(mirror_error, tuf.DownloadLengthMismatchError))
|
||||
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -65,11 +65,14 @@ def test_A1_tempfile_close_temp_file(self):
|
|||
|
||||
|
||||
def _extract_tempfile_directory(self, config_temp_dir=None):
|
||||
"""[Helper] Takes a directory (essentially specified in the conf.py as
|
||||
'temporary_directory') and substitutes tempfile.TemporaryFile() with
|
||||
tempfile.mkstemp() in order to extract actual directory of the stored
|
||||
tempfile. Returns the config's temporary directory (or default temp
|
||||
directory) and actual directory."""
|
||||
"""
|
||||
Takes a directory (essentially specified in the conf.py as
|
||||
'temporary_directory') and substitutes tempfile.TemporaryFile() with
|
||||
tempfile.mkstemp() in order to extract actual directory of the stored
|
||||
tempfile. Returns the config's temporary directory (or default temp
|
||||
directory) and actual directory.
|
||||
"""
|
||||
|
||||
# Patching 'tuf.conf.temporary_directory'.
|
||||
tuf.conf.temporary_directory = config_temp_dir
|
||||
|
||||
|
|
@ -102,12 +105,20 @@ def test_A2_tempfile_init(self):
|
|||
# directory. The location of the temporary files is set in 'tuf.conf.py'.
|
||||
|
||||
# Test: Expected input verification.
|
||||
config_temp_dirs = [None, self.make_temp_directory()]
|
||||
for config_temp_dir in config_temp_dirs:
|
||||
config_temp_dir, actual_dir = \
|
||||
self._extract_tempfile_directory(config_temp_dir)
|
||||
self.assertEqual(config_temp_dir, actual_dir)
|
||||
|
||||
# Assumed 'tuf.conf.temporary_directory' is 'None' initially.
|
||||
temp_file = tuf.util.TempFile()
|
||||
temp_file_directory = os.path.dirname(temp_file.temporary_file.name)
|
||||
self.assertEqual(tempfile.gettempdir(), temp_file_directory)
|
||||
|
||||
saved_temporary_directory = tuf.conf.temporary_directory
|
||||
temp_directory = self.make_temp_directory()
|
||||
tuf.conf.temporary_directory = temp_directory
|
||||
temp_file = tuf.util.TempFile()
|
||||
temp_file_directory = os.path.dirname(temp_file.temporary_file.name)
|
||||
self.assertEqual(temp_directory, temp_file_directory)
|
||||
|
||||
tuf.conf.temporary_directory = saved_temporary_directory
|
||||
|
||||
# Test: Unexpected input handling.
|
||||
config_temp_dirs = [self.random_string(), 123, ['a'], {'a':1}]
|
||||
for config_temp_dir in config_temp_dirs:
|
||||
|
|
|
|||
|
|
@ -87,10 +87,11 @@ def __init__(self, prefix='tuf_temp_'):
|
|||
"""
|
||||
|
||||
self._compression = None
|
||||
|
||||
# If compression is set then the original file is saved in 'self._orig_file'.
|
||||
self._orig_file = None
|
||||
temp_dir = tuf.conf.temporary_directory
|
||||
if temp_dir is not None and isinstance(temp_dir, str):
|
||||
if temp_dir is not None and tuf.formats.PATH_SCHEMA.matches(temp_dir):
|
||||
try:
|
||||
self.temporary_file = tempfile.NamedTemporaryFile(prefix=prefix,
|
||||
dir=temp_dir)
|
||||
|
|
|
|||
Loading…
Reference in a new issue