mirror of
https://github.com/theupdateframework/python-tuf
synced 2026-05-24 10:08:28 +00:00
test_util.py and util.py need to be reviewed.
This commit is contained in:
parent
11d5e99060
commit
615416a9dc
2 changed files with 45 additions and 28 deletions
|
|
@ -48,9 +48,11 @@ def tearDown(self):
|
|||
self.temp_fileobj.close_temp_file()
|
||||
|
||||
|
||||
# TODO
|
||||
def testUtil_A1_close_temp_file(self):
|
||||
pass
|
||||
|
||||
def testUtil_A1_CloseTempFile(self):
|
||||
# Was the temporary file closed?
|
||||
self.temp_fileobj.close_temp_file()
|
||||
self.assertTrue(self.temp_fileobj.temporary_file.closed)
|
||||
|
||||
|
||||
|
||||
|
|
@ -106,19 +108,31 @@ def testUtil_A2_Init(self):
|
|||
|
||||
|
||||
|
||||
# TODO
|
||||
def testTempFile_read(self):
|
||||
pass
|
||||
def testUtil_A3_TempFile_Read(self):
|
||||
filepath = self.make_temp_data_file(data = '1234567890')
|
||||
fileobj = open(filepath, 'rb')
|
||||
|
||||
# Patching 'temp_fileobj.temporary_file'.
|
||||
self.temp_fileobj.temporary_file = fileobj
|
||||
|
||||
# Test: Expected input.
|
||||
self.assertEquals(self.temp_fileobj.read(), '1234567890')
|
||||
self.assertEquals(self.temp_fileobj.read(4), '1234')
|
||||
|
||||
# Test: Unexpected input.
|
||||
for bogus_arg in ['abcd', ['abcd'], {'a':'a'}, -100]:
|
||||
self.assertRaises(tuf.FormatError, self.temp_fileobj.read, bogus_arg)
|
||||
|
||||
|
||||
|
||||
# TODO
|
||||
def testTempFile_write(self):
|
||||
pass
|
||||
def testUtil_A4_TempFile_Write(self):
|
||||
data = self.random_string()
|
||||
self.temp_fileobj.write(data)
|
||||
self.assertEquals(data, self.temp_fileobj.read())
|
||||
|
||||
|
||||
|
||||
def testUtil_A3_TempFile_Move(self):
|
||||
|
||||
def testUtil_A5_TempFile_Move(self):
|
||||
# Destination directory to save the temporary file in.
|
||||
dest_temp_dir = self.make_temp_directory()
|
||||
dest_path = os.path.join(dest_temp_dir, self.random_string())
|
||||
|
|
@ -160,7 +174,7 @@ def _decompress_file(self, compressed_filepath):
|
|||
|
||||
|
||||
|
||||
def testUtil_A4_TempFile_DecompressTempFileObject(self):
|
||||
def testUtil_A6_TempFile_DecompressTempFileObject(self):
|
||||
# Setup: generate a temp file (self.make_temp_data_file()),
|
||||
# compress it. Write it to self.temp_fileobj().
|
||||
filepath = self.make_temp_data_file()
|
||||
|
|
@ -262,7 +276,7 @@ def testUtil_B5_LoadJsonString(self):
|
|||
data = ['a', {'b': ['c', None, 30.3, 29]}]
|
||||
json_string = util.json.dumps(data)
|
||||
self.assertEquals(data, util.load_json_string(json_string))
|
||||
|
||||
|
||||
|
||||
|
||||
def testUtil_B6_LoadJsonFile(self):
|
||||
|
|
|
|||
|
|
@ -21,38 +21,33 @@
|
|||
"""
|
||||
|
||||
|
||||
import gzip
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import re
|
||||
import tempfile
|
||||
import gzip
|
||||
import shutil
|
||||
import logging
|
||||
import tempfile
|
||||
|
||||
import tuf.formats
|
||||
import logging
|
||||
import tuf.hash
|
||||
import tuf.log
|
||||
import tuf.conf
|
||||
import tuf.formats
|
||||
|
||||
|
||||
# See 'log.py' to learn how logging is handled in TUF
|
||||
logger = logging.getLogger('tuf.util')
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
class TempFile(object):
|
||||
"""
|
||||
<Purpose>
|
||||
A high-level temporary file that cleans itself up or can be manually
|
||||
cleaned up. This isn't a complete file-like object. The file functions
|
||||
that are supported make additional common-case safe assumptions. There
|
||||
that are supported make additional common-case safe assumptions. There
|
||||
are additional functions that aren't part of file-like objects. TempFile
|
||||
is used in download.py module.
|
||||
TODO:
|
||||
# We use TemporaryFile for the auto-delete aspects of it to ensure
|
||||
# we don't leave behind temp files.
|
||||
is used in download.py module to temporarily store downloaded data whild
|
||||
all security checks (file hashes/length) are performed.
|
||||
|
||||
"""
|
||||
|
||||
def _default_temporary_directory(self, prefix):
|
||||
|
|
@ -64,6 +59,7 @@ def _default_temporary_directory(self, prefix):
|
|||
raise tuf.Error(err)
|
||||
|
||||
|
||||
|
||||
def __init__(self, prefix='tuf_temp_'):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
@ -97,6 +93,7 @@ def __init__(self, prefix='tuf_temp_'):
|
|||
self._default_temporary_directory(prefix)
|
||||
|
||||
|
||||
|
||||
def flush(self):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
@ -116,6 +113,7 @@ def flush(self):
|
|||
self.temporary_file.flush()
|
||||
|
||||
|
||||
|
||||
def read(self, size=None):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
@ -144,6 +142,7 @@ def read(self, size=None):
|
|||
return self.temporary_file.read(size)
|
||||
|
||||
|
||||
|
||||
def write(self, data, auto_flush=True):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
@ -170,6 +169,7 @@ def write(self, data, auto_flush=True):
|
|||
self.flush()
|
||||
|
||||
|
||||
|
||||
def move(self, destination_path):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
@ -197,6 +197,7 @@ def move(self, destination_path):
|
|||
self.close_temp_file()
|
||||
|
||||
|
||||
|
||||
def seek(self, *args):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
@ -220,6 +221,7 @@ def seek(self, *args):
|
|||
self.temporary_file.seek(*args)
|
||||
|
||||
|
||||
|
||||
def decompress_temp_file_object(self, compression):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
@ -265,6 +267,7 @@ def decompress_temp_file_object(self, compression):
|
|||
self.temporary_file = gzip.GzipFile(fileobj=self.temporary_file, mode='rb')
|
||||
|
||||
|
||||
|
||||
def close_temp_file(self):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
@ -297,7 +300,7 @@ def close_temp_file(self):
|
|||
|
||||
|
||||
|
||||
# TODO: remove 'repository_directory'. Instead specify hash algorithm.
|
||||
|
||||
def get_file_details(file_path):
|
||||
"""
|
||||
<Purpose>
|
||||
|
|
|
|||
Loading…
Reference in a new issue