add ability to delete files/dirs through API, but only in qiita test …

qiita_pet/handlers/cloud_handlers/__init__.py

@@ -1,9 +1,16 @@
 from .file_transfer_handlers import (FetchFileFromCentralHandler,
-                                     PushFileToCentralHandler)
+                                     PushFileToCentralHandler,
+                                     DeleteFileFromCentralHandler)
+from qiita_core.util import is_test_environment
 
 __all__ = ['FetchFileFromCentralHandler']
 
 ENDPOINTS = [
     (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))

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()

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()