diff --git a/qiita_pet/handlers/cloud_handlers/__init__.py b/qiita_pet/handlers/cloud_handlers/__init__.py index d10344a05..7cc90db4f 100644 --- a/qiita_pet/handlers/cloud_handlers/__init__.py +++ b/qiita_pet/handlers/cloud_handlers/__init__.py @@ -1,5 +1,7 @@ from .file_transfer_handlers import (FetchFileFromCentralHandler, - PushFileToCentralHandler) + PushFileToCentralHandler, + DeleteFileFromCentralHandler) +from qiita_core.util import is_test_environment __all__ = ['FetchFileFromCentralHandler'] @@ -7,3 +9,8 @@ (r"/cloud/fetch_file_from_central/(.*)", FetchFileFromCentralHandler), (r"/cloud/push_file_to_central/", PushFileToCentralHandler) ] + +if is_test_environment(): + ENDPOINTS.append( + (r"/cloud/delete_file_from_central/(.*)", + DeleteFileFromCentralHandler)) diff --git a/qiita_pet/handlers/cloud_handlers/file_transfer_handlers.py b/qiita_pet/handlers/cloud_handlers/file_transfer_handlers.py index 74f994c79..cb11108c1 100644 --- a/qiita_pet/handlers/cloud_handlers/file_transfer_handlers.py +++ b/qiita_pet/handlers/cloud_handlers/file_transfer_handlers.py @@ -5,6 +5,7 @@ from tornado.gen import coroutine import zipfile from io import BytesIO +from shutil import rmtree from qiita_core.util import execute_as_transaction, is_test_environment from qiita_db.handlers.oauth2 import authenticate_oauth @@ -104,7 +105,7 @@ def get(self, requested_filepath): filename_directory = "qiita-main-data.zip" if os.path.isdir(filepath): - # Test if this directory is manages by Qiita's DB as directory + # Test if this directory is managed by Qiita's DB as directory # Thus we can prevent that a lazy client simply downloads the whole # basa_data_directory if not is_directory(filepath): @@ -256,3 +257,45 @@ def post(self): '\n'.join(map(lambda x: ' - %s' % x, objs)))) self.finish() + + +class DeleteFileFromCentralHandler(RequestHandler): + # Note: this function is NOT available in productive instances! + @authenticate_oauth + @coroutine + @execute_as_transaction + def get(self, requested_filepath): + if not is_test_environment(): + raise HTTPError(403, reason=( + "You cannot delete files through this API endpoint, when " + "Qiita is not in test-mode!")) + + # ensure we have an absolute path, i.e. starting at / + filepath = os.path.join(os.path.sep, requested_filepath) + # use a canonic version of the filepath + filepath = os.path.abspath(filepath) + + # canonic version of base_data_dir + basedatadir = os.path.abspath(qiita_config.base_data_dir) + + if not filepath.startswith(basedatadir): + # attempt to access files outside of the BASE_DATA_DIR + raise HTTPError(403, reason=( + "You cannot delete file '%s', which is outside of " + "the BASE_DATA_DIR of Qiita!" % filepath)) + + if not os.path.exists(filepath): + raise HTTPError(403, reason=( + "The requested file %s is not present " + "in Qiita's BASE_DATA_DIR!" % filepath)) + + if os.path.isdir(filepath): + rmtree(filepath) + self.write("Deleted directory %s from BASE_DATA_DIR of QIita" % + filepath) + else: + os.remove(filepath) + self.write("Deleted file %s from BASE_DATA_DIR of Qiita" % + filepath) + + self.finish() diff --git a/qiita_pet/handlers/cloud_handlers/tests/test_file_transfer_handlers.py b/qiita_pet/handlers/cloud_handlers/tests/test_file_transfer_handlers.py index 84b3d140b..9bf10c053 100644 --- a/qiita_pet/handlers/cloud_handlers/tests/test_file_transfer_handlers.py +++ b/qiita_pet/handlers/cloud_handlers/tests/test_file_transfer_handlers.py @@ -263,5 +263,77 @@ def test_is_directory(self): self.assertTrue(obs) +class DeleteFileFromCentralHandlerTests(OauthTestingBase): + def setUp(self): + super(DeleteFileFromCentralHandlerTests, self).setUp() + self.endpoint = '/cloud/delete_file_from_central/' + self.base_data_dir = qdb.util.get_db_files_base_dir() + self._clean_up_files = [] + + def tearDown(self): + for fp in self._clean_up_files: + if exists(fp): + if isdir(fp): + rmtree(fp) + else: + remove(fp) + + def test_post(self): + # check if error is raised when NOT providing a filepath + obs = self.get_authed(self.endpoint) + self.assertEqual(obs.status_code, 403) + self.assertIn("You cannot delete file '/', which", obs.reason) + + # check if error is raised when deleting something in productive mode + # we need to let qiita thinks for this test, to NOT be in test mode + with TRN: + TRN.add("UPDATE settings SET test = False") + TRN.execute() + obs = self.get_authed(self.endpoint) + with TRN: + TRN.add("UPDATE settings SET test = True") + TRN.execute() + self.assertEqual(obs.status_code, 403) + self.assertEqual("You cannot delete files through this API endpoint" + ", when Qiita is not in test-mode!", obs.reason) + + # check if error is raised when deleting existing file outside of base + # dir + obs = self.get_authed(self.endpoint + 'home') + self.assertEqual(obs.status_code, 403) + self.assertIn("You cannot delete file '/home', which", obs.reason) + + # check if a file can be deleted + # step 1: create file + fp_file = join(self.base_data_dir, 'deleteme') + with open(fp_file, 'w') as f: + f.write("this file shall be deleted") + self._clean_up_files.append(fp_file) + # step 2: ensure file exists + self.assertTrue(exists(fp_file)) + # step 3: delete file via API + obs = self.get_authed(self.endpoint + fp_file) + self.assertEqual(obs.status_code, 200) + self.assertIn("Deleted file %s from BASE_DATA_DIR" % fp_file, + str(obs.content)) + # step 4: ensure file does not exist anymore + self.assertFalse(exists(fp_file)) + + # check if a directory can be deleted + # step 1: create directory + fp_dir = join(self.base_data_dir, 'deletemeDir') + makedirs(fp_dir) + self._clean_up_files.append(fp_dir) + # step 2: ensure file exists + self.assertTrue(exists(fp_dir)) + # step 3: delete file via API + obs = self.get_authed(self.endpoint + fp_dir) + self.assertEqual(obs.status_code, 200) + self.assertIn("Deleted directory %s from BASE_DATA_DIR" % fp_dir, + str(obs.content)) + # step 4: ensure file does not exist anymore + self.assertFalse(exists(fp_dir)) + + if __name__ == "__main__": main()