diff options
| -rw-r--r-- | gn3/file_utils.py | 29 | ||||
| -rw-r--r-- | tests/unit/test_file_utils.py | 47 | ||||
| -rw-r--r-- | tests/unit/upload-data.tar.gz | bin | 0 -> 445 bytes |
3 files changed, 75 insertions, 1 deletions
diff --git a/gn3/file_utils.py b/gn3/file_utils.py index a6929e1..d9301bb 100644 --- a/gn3/file_utils.py +++ b/gn3/file_utils.py @@ -2,9 +2,12 @@ import hashlib import json import os +import shutil +import tarfile from functools import partial from typing import Dict +from werkzeug.utils import secure_filename from gn3.settings import APP_DEFAULTS @@ -46,3 +49,29 @@ def jsonfile_to_dict(json_file: str) -> Dict: data = json.load(_file) return data raise FileNotFoundError + + +def extract_uploaded_file(gzipped_file, target_dir: str) -> Dict: + """Get the (directory) hash of extracted contents of GZIPPED_FILE; and move +contents to TARGET_DIR/<dir-hash>. + + """ + tar_target_loc = os.path.join(target_dir, + secure_filename(gzipped_file.filename)) + gzipped_file.save(tar_target_loc) + try: + # Extract to "tar_target_loc/tempdir" + tar = tarfile.open(tar_target_loc) + tar.extractall( + path=os.path.join(target_dir, "tempdir")) + tar.close() + # pylint: disable=W0703 + except Exception: + return {"status": 128, "error": "gzip failed to unpack file"} + dir_hash = get_dir_hash(tar_target_loc) + if os.path.exists(os.path.join(target_dir, dir_hash)): + shutil.rmtree(os.path.join(target_dir, 'tempdir')) + else: + os.rename(os.path.join(target_dir, "tempdir"), + os.path.join(target_dir, dir_hash)) + return {"status": 0, "token": dir_hash} diff --git a/tests/unit/test_file_utils.py b/tests/unit/test_file_utils.py index f5efa02..166e576 100644 --- a/tests/unit/test_file_utils.py +++ b/tests/unit/test_file_utils.py @@ -1,11 +1,22 @@ """Test cases for procedures defined in file_utils.py""" import os +import shutil import unittest +from dataclasses import dataclass +from typing import Callable from unittest import mock +from gn3.file_utils import extract_uploaded_file +from gn3.file_utils import get_dir_hash from gn3.file_utils import jsonfile_to_dict from gn3.file_utils import lookup_file -from gn3.file_utils import get_dir_hash + + +@dataclass +class MockFile: + """Mock File object returned by request.files""" + filename: str + save: Callable class TestFileUtils(unittest.TestCase): @@ -63,3 +74,37 @@ non-existent""" self.assertRaises(FileNotFoundError, jsonfile_to_dict, "/non-existent-dir") + + @mock.patch("gn3.file_utils.secure_filename") + def test_extract_uploaded_file(self, mock_file): + """Test that the gzip file is extracted to the right location""" + file_loc = os.path.join( + os.path.dirname(__file__), + "upload-data.tar.gz") + mock_file.return_value = file_loc + mock_fileobj = MockFile(save=mock.MagicMock(), + filename="upload-data.tar.gz") + result = extract_uploaded_file(mock_fileobj, "/tmp") + mock_fileobj.save.assert_called_once_with(file_loc) + mock_file.assert_called_once_with("upload-data.tar.gz") + # Clean up! + shutil.rmtree(os.path.join("/tmp", + "d41d8cd98f00b204e9800998ecf8427e")) + self.assertEqual(result, + {"status": 0, + "token": "d41d8cd98f00b204e9800998ecf8427e"}) + + @mock.patch("gn3.file_utils.secure_filename") + def test_extract_uploaded_file_non_existent_gzip(self, mock_file): + """Test that the right error message is returned when there is a problem +extracting the file""" + file_loc = os.path.join( + os.path.dirname(__file__), + "CTtyodSTh5") # Does not exist! + mock_file.return_value = file_loc + mock_fileobj = MockFile(save=mock.MagicMock(), + filename="") + result = extract_uploaded_file(mock_fileobj, "/tmp") + self.assertEqual(result, + {"status": 128, + "error": "gzip failed to unpack file"}) diff --git a/tests/unit/upload-data.tar.gz b/tests/unit/upload-data.tar.gz new file mode 100644 index 0000000..86077d9 --- /dev/null +++ b/tests/unit/upload-data.tar.gz Binary files differ |