generatlized strip_extension and added tests.

workflow/scripts/map_reads.py

@@ -59,15 +59,6 @@ def get_args():
 
 ## Functions
 
-def strip_extensions(filename, extensions):
-    '''Strips extensions to get basename of file.'''
-
-    basename = filename
-    for extension in extensions:
-        basename = basename.rpartition(extension)[0] or basename
-
-    return basename
-
 
 def check_tools():
     '''Checks for required componenets on user system'''
@@ -92,7 +83,7 @@ def check_tools():
 def generate_sa(fastq, reference):
     '''Use BWA to generate Suffix Arrays.'''
 
-    fastq_basename = os.path.basename(strip_extensions(fastq, STRIP_EXTENSIONS))
+    fastq_basename = os.path.basename(utils.strip_extensions(fastq, STRIP_EXTENSIONS))
 
     bwa_aln_params = '-q 5 -l 32 -k 2'
 
@@ -184,16 +175,16 @@ def main():
     # Run alignment for either PE or SE
     if paired: # paired-end data
         fastq_r1_basename = os.path.basename(
-            strip_extensions(fastq[0], STRIP_EXTENSIONS))
+            utils.strip_extensions(fastq[0], STRIP_EXTENSIONS))
         fastq_r2_basename = os.path.basename(
-            strip_extensions(fastq[1], STRIP_EXTENSIONS))
+            utils.strip_extensions(fastq[1], STRIP_EXTENSIONS))
         fastq_basename = fastq_r1_basename + fastq_r2_basename
 
         bam_filename = align_pe(fastq, sai, reference, fastq_basename)
 
     else:
         fastq_basename = os.path.basename(
-            strip_extensions(fastq[0], STRIP_EXTENSIONS))
+            utils.strip_extensions(fastq[0], STRIP_EXTENSIONS))
 
         bam_filename = align_se(fastq, sai, reference, fastq_basename)
 

workflow/scripts/utils.py

@@ -46,3 +46,13 @@ def run_pipe(steps, outfile=None):
             p = p_next
     out, err = p.communicate()
     return out, err
+
+
+def strip_extensions(filename, extensions):
+    '''Strips extensions to get basename of file.'''
+
+    basename = filename
+    for extension in extensions:
+        basename = basename.rpartition(extension)[0] or basename
+
+    return basename

workflow/tests/test_utils.py

@@ -6,6 +6,8 @@ import shlex
 import utils
 
 
+STRIP_EXTENSIONS = ['.gz', '.fq', '.fastq', '.fa', '.fasta']
+
 @pytest.fixture
 def steps():
     steps = []
@@ -51,3 +53,18 @@ def test_run_last_step_file(steps_2, capsys, tmpdir):
     output, errors = capsys.readouterr()
     assert "last step shlex" in output
     assert check_output in tmp_outfile.read()
+
+
+def test_strip_extensions():
+    filename = utils.strip_extensions('ENCFF833BLU.fastq.gz',STRIP_EXTENSIONS)
+    assert filename == 'ENCFF833BLU'
+
+
+def test_strip_extensions_not_valid():
+    filename = utils.strip_extensions('ENCFF833BLU.not.valid',STRIP_EXTENSIONS)
+    assert filename == 'ENCFF833BLU.not.valid'
+
+
+def test_strip_extensions_missing_basename():
+    filename = utils.strip_extensions('.fastq.gz',STRIP_EXTENSIONS)
+    assert filename == '.fastq'