initial tests + docstrings

Author: sjanssen2

qiita_client/qiita_client.py

@@ -7,6 +7,7 @@
 # -----------------------------------------------------------------------------
 
 import os
+import shutil
 import time
 import requests
 import threading
@@ -742,11 +743,53 @@ def _process_files_per_sample_fastq(self, files, prep_info,
 
         return sample_names, prep_info
 
-    def fetch_file_from_central(self, filepath):
+    def fetch_file_from_central(self, filepath, prefix=''):
+        """Moves content of a file from Qiita's central base_data_dir to a
+           local plugin file-system.
+
+           By default, this is exactly the same location, i.e. the return
+           filepath is identical to the requested one and nothing is moved /
+           copied.
+           However, for less tight plugin couplings, file content can be
+           transferred via https for situations where the plugin does not have
+           native access to Qiita's overall base_data_dir.
+
+        Parameters
+        ----------
+        filepath : str
+            The filepath in Qiita's central base_data_dir to the requested
+            file content
+        prefix : str
+            Primarily for testing: prefix the target filepath with this
+            filepath prefix to
+            a) in 'filesystem' mode: create an actual file copy (for testing)
+               If prefix='', nothing will be copied/moved
+            b) in 'https' mode: flexibility to locate files differently in
+               plugin local file system.
+
+        Returns
+        -------
+        str : the filepath of the requested file within the local file system
+        """
+        target_filepath = filepath
+        if prefix != '':
+            # strip off root
+            if filepath.startswith(os.path.abspath(os.sep)):
+                target_filepath = target_filepath[
+                    len(os.path.abspath(os.sep)):]
+            # prefix filepath with given prefix
+            target_filepath = os.path.join(prefix, target_filepath)
+
         if self._plugincoupling == 'filesystem':
-            return filepath
+            if prefix != '':
+                # create necessary directory locally
+                os.makedirs(os.path.dirname(target_filepath), exist_ok=True)
+
+                shutil.copyfile(filepath, target_filepath)
+
+            return target_filepath
 
-        if self._plugincoupling == 'https':
+        elif self._plugincoupling == 'https':
             logger.debug('Requesting file %s from qiita server.' % filepath)
 
             # actual call to Qiita central to obtain file content
@@ -755,23 +798,24 @@ def fetch_file_from_central(self, filepath):
                 rettype='content')
 
             # create necessary directory locally
-            os.makedirs(os.path.dirname(filepath), exist_ok=True)
+            os.makedirs(os.path.dirname(target_filepath), exist_ok=True)
 
             # write retrieved file content
-            with open(filepath, 'wb') as f:
+            with open(target_filepath, 'wb') as f:
                 f.write(content)
 
-            return filepath
+            return target_filepath
 
-        raise ValueError(
-            ("File communication protocol '%s' as defined in plugins "
-             "configuration is NOT defined.") % self._plugincoupling)
+        else:
+            raise ValueError(
+                ("File communication protocol '%s' as defined in plugins "
+                 "configuration is NOT defined.") % self._plugincoupling)
 
     def push_file_to_central(self, filepath):
         if self._plugincoupling == 'filesystem':
             return filepath
 
-        if self._plugincoupling == 'https':
+        elif self._plugincoupling == 'https':
             logger.debug('Submitting file %s to qiita server.' % filepath)
 
             self.post(

qiita_client/tests/test_qiita_client.py

@@ -7,6 +7,7 @@
 # -----------------------------------------------------------------------------
 
 from unittest import TestCase, main
+import filecmp
 from os import remove, close
 from os.path import basename, exists
 from tempfile import mkstemp
@@ -389,12 +390,37 @@ def test_fetch_file_from_central(self):
         self.tester._plugincoupling = 'filesystem'
 
         ainfo = self.tester.get("/qiita_db/artifacts/%s/" % 1)
-        print("STEFAN", ainfo)
+        fp = ainfo['files']['raw_forward_seqs'][0]['filepath']
 
-        # fp_query = '/home/runner/work/qiita/qiita/qiita_db/support_files/
-        # test_data/templates/FASTA_QUAL_preprocessing_sample_template.txt'
-        # fp_res = fetch_file_from_central('')
-        pass
+        # mode: filesystem, prefix='': no copy, directly return given fp
+        fp_obs = self.tester.fetch_file_from_central(fp)
+        self.assertEqual(fp, fp_obs)
+
+        # mode: filesystem, prefix='/karl': make file copy
+        self.clean_up_files.append('/karl' + fp)
+        fp_obs = self.tester.fetch_file_from_central(fp, prefix='/karl')
+        self.assertEqual('/karl' + fp, fp_obs)
+        self.assertTrue(filecmp.cmp(fp, fp_obs, shallow=False))
+
+        # non existing mode
+        with self.assertRaises(ValueError):
+            self.tester._plugincoupling = 'foo'
+            self.tester.fetch_file_from_central(fp)
+
+    def test_push_file_to_central(self):
+        self.tester._plugincoupling = 'filesystem'
+
+        ainfo = self.tester.get("/qiita_db/artifacts/%s/" % 1)
+        fp = ainfo['files']['raw_forward_seqs'][0]['filepath']
+
+        # mode: filesystem
+        fp_obs = self.tester.push_file_to_central(fp)
+        self.assertEqual(fp, fp_obs)
+
+        # non existing mode
+        with self.assertRaises(ValueError):
+            self.tester._plugincoupling = 'foo'
+            self.tester.push_file_to_central(fp)
 
 
 if __name__ == '__main__':